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。
网友评论