flask is deployed using gevent-websocket + gunicorn

pip3 install gevent-websocket
pip3 install gunicorn

app.py demo

from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer

app = Flask(__name__)

@app.route('/echo/')
def echo():
    if request.environ.get('wsgi.websocket'):
        ws = request.environ['wsgi.websocket']
        while True:
            msg = ws.receive()
            ws.send(msg)

if __name__ == '__main__':
    http_server = WSGIServer(('', 5000), app, handler_class=WebSocketHandler)
    http_server.serve_forever()

Use gunicorn to start and specify gevent-websocket

gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker app:app

django uses Django-websocket

https://github.com/archever/p...

Thank you everyone, I couldn’t find a solution on Baidu for a long time. The master above should be able to use it, but I don’t understand it very well. Thank you, I really gained something by going to Google, so I decisively abandoned Baidu

This is on someone else’s github, you can use it

# install ws4py
# pip install ws4py
# easy_install ws4py
from ws4py.client.threadedclient import WebSocketClient

class DummyClient(WebSocketClient):
    def opened(self):
        self.send("www.baidu.com")

    def closed(self, code, reason=None):
        print "Closed down", code, reason

    def received_message(self, m):
        print m

if __name__ == '__main__':
    try:
        ws = DummyClient('ws://10.222.138.163:1889/websocket', protocols=['chat'])
        ws.connect()
        ws.run_forever()
    except KeyboardInterrupt:
        ws.close()

It is recommended to use tornado, which supports websocket. The backend of Zhihu is built using tornado