python - 从 Starlette 中的同步迭代器通过 websocket 发送数据
问题描述
我有一个同步迭代器,它来自第三方包。迭代器查询外部服务并产生一些数据。如果没有数据,则迭代器等待它。我将WebSocketEndpoint
Starlette 的子类化为通过 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 语句表明,迭代器产生数据,但实际上没有发送任何数据。如果我放return
after websocket.send_json()
,它将正确发送迭代器的第一个结果,但循环将在之后完成。为什么?
另一个问题是迭代器完全阻塞了应用程序的执行。我理解它为什么会发生,但由于它是一个 Web 服务,并且迭代器旨在工作,直到客户端与 Websocket 断开连接,它很容易阻塞我的整个应用程序。如果我有 10 个工作人员,那么 10 个 websocket 客户端将阻止应用程序,并且在其中一个 websocket 断开连接之前将无法执行任何操作。我该如何解决?
解决方案
这是一个第三方对象,根本不是异步的。
这就是问题所在 - 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)
,一切都应该正常工作。
推荐阅读
- jspdf - 添加 jsPDF 字体?
- c# - 我们的项目不断使用具有非静态嵌套类的静态外部类——这可能有任何陷阱吗?
- javascript - 尝试在父组件之前将 ref 数据传递给组件的 prop
- reactjs - gatsby build → WebpackError: TypeError: Cannot read property 'GoogleAuthProvider' of undefined
- sql-server - SQL INFORMATION_SCHEMA.VIEWS
- c# - UWP C# Windows.Web.HttpClient 内存泄漏
- java - 如何在 Kotlin 的 viewPager 中添加动画?
- javascript - JavaScript 网格复制为字符串
- javascript - 如何记住溢出的滚动位置:滚动div?
- windows - WIN 10 PostgreSQL DB symbols are unreadable