在这种情况下如何正确使用 asyncio？

Posted 2023-03-07

【中文标题】在这种情况下如何正确使用 asyncio？

【英文标题】：How to properly use asyncio in this context?

【发布时间】：2022-01-11 21:01:29

【问题描述】：

我有一个在基于 websocket 的服务器中使用事件循环的想法（基于 websockets 库），但我很难做到这一点，因为我真的不知道如何使用asyncio 并编写自己的协程。

# let us define a dictionary of callbacks awaiting to be triggered
CALLBACKS = dict()

# let us have an infinite loop, which reads new messages from a websocket connection
async def websocket_connection(ws, path):
     async for msg in ws:
         # we received a message
         msg = json.loads(msg)
         msg_id = msg['id']
         msg_data = msg['data']
         # we check, if there is a callback waiting to be triggered with such id
         if msg_id in CALLBACKS:
             # if such id exists, we get the callback function
             callback = CALLBACKS[msg_id]
             # we delete the callback from the dictionary
             del CALLBACKS[msg_id]
             # we call it passing the data from the message in there
             callback(msg_data)
         else:
             # we don't have such id, throw some error
             raise Exception("bad id " + msg_id)


# this is the function which submits a message
async def submit(ws, data):
    # it generates a random request id
    from uuid import uuid4
    request_id = str(uuid4())
    # registers a callback, which will simply return the received data
    CALLBACKS[request_id] = lambda data:  # <- this is the part I don't know what it should really do to be able to return the value from the await
    # after it encodes the message as json
    msg = json.dumps('id': request_id, 'data': data)
    # and sends via the websocket
    await ws.send(msg)

# and let us have the actual function, which will be the actual script
async def websocket_script(ws):
    # this is what I would like to be able to do in the end.
    # pass in commands to the websocket and await their completion
    sum = int(await submit(ws, "eval": "2+2"))
    division = float(await submit(ws, "eval": "3/2"))

websocket_connection 和 websocket_script 需要并排运行才能正常工作。我想gather 或其他一些async 函数会起作用。我真的很想摆脱回调，因为这是首先使用asyncio 的最初目的。

如何做到这一点？这似乎是asyncio 的工作。

【参考方案1】：

如果我理解正确的话，我想你搜索的是asyncio.Future。

FUTURES = dict()

async def websocket_connection(ws, path):
    # ...
    fut = FUTURES[msg_id]
    fut.set_result(msg_data)  # will "unblock" await fut below
    del FUTURES[msg_id]
    # ...


async def submit(ws, data):
    from uuid import uuid4
    request_id = str(uuid4())

    fut = asyncio.Future()
    FUTURES[request_id] = fut

    msg = json.dumps('id': request_id, 'data': data)    
    await ws.send(msg)
    # ...
    
    return (await fut)


async def websocket_script(ws):
    sum = int(await submit(ws, "eval": "2+2"))
    division = float(await submit(ws, "eval": "3/2"))

【讨论】：

спасибо, нигде больше не видел решения。