我已经写了以下异步池的代码。在__aexit__
中,当任务完成后,我会取消工作任务。但是当我运行代码时,工作任务没有被取消,代码一直在运行。这是任务的样子:<Task pending coro=<AsyncPool._worker() running at \async_pool.py:17> wait_for=<Future cancelled>>
。虽然asyncio.wait_for
被取消了,但工作任务并没有被取消。
class AsyncPool:
def __init__(self,coroutine,no_of_workers,timeout):
self._loop = asyncio.get_event_loop()
self._queue = asyncio.Queue()
self._no_of_workers = no_of_workers
self._coroutine = coroutine
self._timeout = timeout
self._workers = None
async def _worker(self):
while True:
try:
ret = False
queue_item = await self._queue.get()
ret = True
result = await asyncio.wait_for(self._coroutine(queue_item), timeout = self._timeout,loop= self._loop)
except Exception as e:
print(e)
finally:
if ret:
self._queue.task_done()
async def push_to_queue(self,item):
self._queue.put_nowait(item)
async def __aenter__(self):
assert self._workers == None
self._workers = [asyncio.create_task(self._worker()) for _ in range(self._no_of_workers)]
return self
async def __aexit__(self,type,value,traceback):
await self._queue.join()
for worker in self._workers:
worker.cancel()
await asyncio.gather(*self._workers, loop=self._loop, return_exceptions =True)
使用Asyncpool:
async def something(item):
print("got", item)
await asyncio.sleep(item)
async def main():
async with AsyncPool(something, 5, 2) as pool:
for i in range(10):
await pool.push_to_queue(i)
asyncio.run(main())
我终端中的输出:
![在此输入图片描述](https://istack.dev59.com/J8ZpR.webp)
main()
像这样 来复制它时,我得到了一个CancelledError
。这是因为gather()
等待取消的任务。在gather
调用中添加return_exceptions=True
可以修复此错误,并且不会打印任何警告。 - user4815162342gather
完成),程序退出,因此任务显然无法继续运行。你能否编辑问题以包含可以复现问题的可运行的代码? - user4815162342