0

我有一个同步迭代器，它来自第三方包。迭代器查询外部服务并产生一些数据。如果没有数据，则迭代器等待它。我将WebSocketEndpointStarlette 的子类化为通过 websocket 从迭代器发送新数据。不幸的是，我似乎不明白某些东西，而且我的代码没有按预期工作。这是一个稍微简化的代码：

import time

from starlette.endpoints import WebSocketEndpoint
from starlette.websockets import WebSocket


class Iterator:
    """This is a third-party object, not asynchronous at all."""

    def __init__(self):
        self._stop = False

    def __iter__(self):
        self.i = 0
        return self

    def __next__(self):
        if self._stop:
            raise StopIteration

        time.sleep(5)
        self.i += 1
        print(self.i)
        return self.i

    def cancel(self):
        self._stop = True


class MyWebSocket(WebSocketEndpoint):
    def __init__(self, scope, receive, send) -> None:
        super().__init__(scope, receive, send)

        self.iterator = Iterator()

    async def on_connect(self, websocket: WebSocket) -> None:
        await super().on_connect(websocket)

        for message in self.iterator:
            await websocket.send_json({"message": message})

    async def on_disconnect(self, websocket: WebSocket, close_code: int) -> None:
        await super().on_disconnect(websocket, close_code)

        self.iterator.cancel()

第一个问题 - 代码不通过 websocket 发送任何数据。print 语句表明，迭代器产生数据，但实际上没有发送任何数据。如果我放returnafter websocket.send_json()，它将正确发送迭代器的第一个结果，但循环将在之后完成。为什么？

另一个问题是迭代器完全阻塞了应用程序的执行。我理解它为什么会发生，但由于它是一个 Web 服务，并且迭代器旨在工作，直到客户端与 Websocket 断开连接，它很容易阻塞我的整个应用程序。如果我有 10 个工作人员，那么 10 个 websocket 客户端将阻止应用程序，并且在其中一个 websocket 断开连接之前将无法执行任何操作。我该如何解决？

4

1 回答 1

0

这是一个第三方对象，根本不是异步的。

这就是问题所在 - asyncio 是单线程的，因此您的迭代器必须完全不阻塞（例如在对内置集合进行迭代时），或者您必须使用异步迭代器和async for在等待时暂停执行的循环下一项。

在处理第三方阻塞函数时，您可以将其合并到异步代码中，使用run_in_executor它将函数提交到线程池并暂停当前协程直到函数完成。您不能直接将迭代器传递给run_in_executor，但您可以创建一个包装器，该包装器采用同步迭代器并运行每个单独的__next__through调用run_in_executor，提供异步迭代器的接口。例如：

async def wrap_iter(iterable):
    loop = asyncio.get_event_loop()
    it = iter(iterable)

    DONE = object()
    def get_next_item():
        # Get the next item synchronously.  We cannot call next(it)
        # directly because StopIteration cannot be transferred
        # across an "await".  So we detect StopIteration and
        # convert it to a sentinel object.
        try:
            return next(it)
        except StopIteration:
            return DONE

    while True:
        # Submit execution of next(it) to another thread and resume
        # when it's done.  await will suspend the coroutine and
        # allow other tasks to execute while waiting.
        next_item = await loop.run_in_executor(None, get_next_item)
        if next_item is DONE:
            break
        yield next_item

现在您可以替换for message in self.iterator为async for message in wrap_iter(self.iterator)，一切都应该正常工作。

于 2019-09-07T21:50:52.937 回答