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。
