如何在Python中以FIFO的方式處理多個WebSockets？

Question

我在寫一個處理兩個WebSocket的函數，每個WebSocket的響應都會改變一個共享的DataFrame df。

import json 
import asyncio 
import websockets 

@asyncio.coroutine 
def printResponse(df, dataSocket, quoteSocket, dataRequest, quoteRequest): 

    yield from dataSocket.send(dataRequest) 
    yield from quoteSocket.send(quoteRequest) 

    response = yield from dataSocket.recv()  # skip first response 
    response = yield from quoteSocket.recv() # skip first response 

    while True: 

     response = yield from dataSocket.recv() 
     print("<< {}".format(json.loads(response))) 
     df = changeRecord(df, response) 

     response = yield from quoteSocket.recv() 
     print("<< {}".format(json.loads(response))) 
     df = changeRecord(df, response) 

我不確定，但目前的代碼似乎輪流處理兩個WebSockets。我想以「先進先出」的方式處理響應，無論它來自哪個WebSocket。我應該如何改變才能達到這個目標？

Answer 1

因爲您在同一個while循環中使用兩個yield from語句，所以它將按順序處理它們，然後重複無限次。

所以它會一直等到從dataSocket得到響應，然後它會等到從quoteSocket得到響應，然後清洗並重復。

Tasks()對於您正在嘗試執行的操作非常有用，因爲它們允許協同程序彼此獨立運行。所以，如果你在他們自己的任務包裝器中啓動兩個單獨的協程，那麼每個人都將等待他們自己的下一個響應，而不必打擾另一個。

例如：

import json 
import asyncio 
import websockets 

@asyncio.coroutine 
def coroutine_1(df, dataSocket): 
    yield from dataSocket.send(dataRequest) 
    response = yield from dataSocket.recv()  # skip first response 
    while True: 
     response = yield from dataSocket.recv() 
     print("<< {}".format(json.loads(response))) 
     df = changeRecord(df, response) 

@asyncio.coroutine 
def coroutine_2(df, quoteSocket): 
    yield from quoteSocket.send(quoteRequest) 
    response = yield from quoteSocket.recv() # skip first response 
    while True: 
     response = yield from quoteSocket.recv() 
     print("<< {}".format(json.loads(response))) 
     df = changeRecord(df, response) 

@asyncio.coroutine 
def printResponse(df, dataSocket, quoteSocket): 

    websocket_task_1 = asyncio.ensure_future(coroutine_1(df, dataSocket)) 
    websocket_task_2 = asyncio.ensure_future(coroutine_2(df, quoteSocket)) 

    yield from asyncio.wait([websocket_task_1, websocket_task_2])