如何在Python中以先进先出的方式处理多个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()对你想做的事情很好，因为它们允许协同程序相互独立地操作。因此，如果您在各自的Task包装器中启动两个独立的协程，那么每个协程都将等待自己的下一个响应，而不必干扰对方。

例如:

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])