SocketIO是一個基于websocket的封裝的傳輸框架。在大多數對數據量要求不高的場景里，可以用于快速搭建實時數據流。SocketIO最大的優點應該是它對數據可以進行json封裝，因而減去了傳統socket通信中的拆包粘包的工序。
Flask-SocketIO允許在Flask的框架下直接構建SOcketIO服務，實現方式非常簡單. 這里是一個基本的消息接收應答的例子
from flask import Flask, request, make_response, jsonify, session from flask_cors import CORS from flask_socketio import SocketIO '''==============================''' '''flask web server config''' '''==============================''' app = Flask(__name__, static_url_path='') app.config['SECRET_KEY'] = 'secret!' cors = CORS(app,resources={r"/*":{"origins":"*"}}) # Restful使用跨域CORS訪問時通過CORS進行支持 sio = SocketIO(app) '''websocket應答消息''' @sio.on('message', namespace='/ws') def ws_message(data): ask = {'result':'OK'} sio.emit('message_back', json.dumps(ack), namespace='/ws') '''restful訪問，返回json''' @app.route('/api', methods=['GET']) def api_message(): resp = make_response(jsonify({'result':'OK'}) return resp def main(): pass if __name__ == '__main__': main() sio.run(app, port=8092, host='0.0.0.0', use_reloader=False, debug=False) while True: time.sleep(10) # 防止程序退出
在簡單的應用中，app和websocket可以同時共存。當服務器處于并發模式的時候，例如服務器通過多個線程向socketio emit消息的時候，如果簡單采用如下的方式:
import threading thread_send_msg = threading.Thread(task_send_msg) thread_send_msg.start() def task_send_msg(): while True: time.sleep(1) msg sio.emit('message', json.dumps(msg), namespace='/ws')
你會發現，socketio的發送間隔會出現模型的延遲，而且間隔也會變得不是每秒發送一次。
那么可否采用multiprocessing?
import multiprocessing proc_send_msg = multiprocessing.Process(task_send_msg) proc_send_msg.start() def task_send_msg(): while True: time.sleep(1) msg sio.emit('message', json.dumps(msg), namespace='/ws')
測試的結果仍然會出現問題，甚至會出現flask服務完全不響應。
所以問題是什么?
查閱documentation和面向stackoverflow編程之后，發現原因在于，socketio內部采用的是協程任務調度，這樣如果把emit的行為放置在線程或著進程內時，并沒有解決并發emit的沖突問題。我們需要回歸到coroutine的調度模式本身。在這里，采用了gevent，也可以根據自己的需求使用eventlet或者其他的模塊。gevent并發模式也比較簡單:
task_1(): gevent.sleep(1) print 'running task 1' task_2(): gevent.sleep(1) print 'running task 2' gevent.addall([ gevent.spawn(task_1), gevent.spawn(task_2) ])
上述的操作就是在gevent內孵化(spawn)兩個協程，然后每個協程每一秒通過gevent.sleep()讓出gvent供其他的協程使用。通過這個模式，我們將emit并發事件修改為這樣:
socketio_msg_queue = Queue(maxsize=5000) def gtask_sockeio_emit(): while True: gevent.sleep(0.01) try: msg = socketio_msg_queue.get_nowait() except: continue print msg sio.emit(msg['on'], json.dumps(msg['data']), namespace='/ws/rt_market') gevent.addAll([ gtask_socketio_emit ])
這段代碼中，我們進一步使用了一個隊列將各個線程可能產生的emit事件推送到一個隊列之中，然后在一個統一的協程內進行發送。
那么問題解決了么?還沒有。。。
當把emit放入協程之后我們又發現，flask框架不響應了。
最大的可能就是gevent本身搶占了進程的資源，導致restful沒法響應。
第一個想到的方法是把flask放入單獨一個進程，例如這樣:
def task_start_flask(): 'start the rest and ws server' sio.run(app, port=80, host='0.0.0.0', use_reloader=False, debug=False) proc_flask = multiprocessing.Process(target=task_start_flask) proc_flask.start()
結果兩邊都無法訪問了。
分析之后，猜測可能是因為socketio建立需要通過web請求，因此flask在哪里啟動，所有的socketio通信就會堆積在哪里，分開啟動并不能解決問題。所以，第二次，我們把所有的內容都放在gevent里統一調度:
gevent.joinall([ gevent.spawn(gtask_sockeio_emit), gevent.spawn(task_start_flask), ])
至此問題解決。
后記:SocketIO從協議本身上看效率并不高，如果需要更高效率，最好還是采用原生websocket。
