SocketIO是一個基于websocket的封裝的傳輸框架。在大多數(shù)對數(shù)據量要求不高的場景里，可以用于快速搭建實時數(shù)據流。SocketIO最大的優(yōu)點應該是它對數(shù)據可以進行json封裝，因而減去了傳統(tǒng)socket通信中的拆包粘包的工序。

Flask-SocketIO允許在Flask的框架下直接構建SOcketIO服務，實現(xiàn)方式非常簡單. 這里是一個基本的消息接收應答的例子

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可以同時共存。當服務器處于并發(fā)模式的時候，例如服務器通過多個線程向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')

你會發(fā)現(xiàn)，socketio的發(fā)送間隔會出現(xiàn)模型的延遲，而且間隔也會變得不是每秒發(fā)送一次。

那么可否采用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')

測試的結果仍然會出現(xiàn)問題，甚至會出現(xiàn)flask服務完全不響應。

所以問題是什么？

查閱documentation和面向stackoverflow編程之后，發(fā)現(xiàn)原因在于，socketio內部采用的是協(xié)程任務調度，這樣如果把emit的行為放置在線程或著進程內時，并沒有解決并發(fā)emit的沖突問題。我們需要回歸到coroutine的調度模式本身。在這里，采用了gevent，也可以根據自己的需求使用eventlet或者其他的模塊。gevent并發(fā)模式也比較簡單：

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）兩個協(xié)程，然后每個協(xié)程每一秒通過gevent.sleep()讓出gvent供其他的協(xié)程使用。通過這個模式，我們將emit并發(fā)事件修改為這樣：

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事件推送到一個隊列之中，然后在一個統(tǒng)一的協(xié)程內進行發(fā)送。

那么問題解決了么？還沒有。。。

當把emit放入協(xié)程之后我們又發(fā)現(xiàn)，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里統(tǒng)一調度：

    gevent.joinall([
        gevent.spawn(gtask_sockeio_emit),
        gevent.spawn(task_start_flask),
    ])

至此問題解決。

后記：SocketIO從協(xié)議本身上看效率并不高，如果需要更高效率，最好還是采用原生websocket。