如果任务失败，我如何召回&QUOT；TASK&QOOT；？ [英] How can I recall "task" if task is failed?

问题描述

有函数make_request向API发出http请求。而且我每秒不能发出超过3个请求。

我做过类似的事情

coroutines = [make_request(...) for ... in ...]
tasks = []
for coroutine in coroutines:
   tasks.append(asyncio.create_task(coroutine))
   await asyncio.sleep(1 / 3)
responses = asyncio.gather(*tasks)

而且我每小时不能发出超过1000个请求。(可能，我可以延迟3600 / 1000。)如果互联网连接中断了怎么办？我应该再次尝试提出请求。

我可以这样包装make_request：

async def try_make_request(...):
   while True:
      try:
         return await make_request(...)
      exception APIError as err:
         logging.exception(...)

在这种情况下，每秒可能会发出3个以上的请求。

我找到了that解决方案，但我不确定这是否为最佳解决方案

pending = []
coroutines = [...]
for coroutine in coroutines:
    pending.append(asyncio.create_task(coroutine))
    await asyncio.sleep(1 / 3)
result = []
while True:
    finished, pending = await asyncio.wait(
        pending, return_when=asyncio.FIRST_EXCEPTION
    )
    for task in finished:
        exc = task.exception()
        if isinstance(exc, APIError) and exc.code == 29:
            pending.add(task.get_coro()) # since python 3.8
        if exc:
            logging.exception(...)
        else:
            result.append(task.result())
    if not pending:
        break

推荐答案

如果我理解正确，发起连接的频率不能超过3.6秒。实现这一点的一种方法是设置一个计时器，该计时器在每次启动连接时都会重置，并且在3.6秒后过期，从而允许启动下一个连接。例如：

class Limiter:
    def __init__(self, delay):
        self.delay = delay
        self._ready = asyncio.Event()
        self._ready.set()

    async def wait(self):
        # wait in a loop because if there are multiple waiters,
        # the wakeup can be spurious
        while not self._ready.is_set():
            await self._ready.wait()
        # We got the slot and can proceed with the download.
        # Before doing so, clear the ready flag to prevent other waiters
        # from commencing downloads until the delay elapses again.
        self._ready.clear()
        asyncio.get_event_loop().call_later(self.delay, self._ready.set)

则try_make_request可能如下所示：

async def try_make_request(limiter, ...):
    while True:
        await limiter.wait()
        try:
            return await make_request(...)
        exception APIError as err:
            logging.exception(...)

.和主协程可以全部并行等待try_make_request%s：

limiter = Limiter(3600/1000)
responses = await asyncio.gather(*[try_make_request(limiter, ...) for ... in ...])

这篇关于如果任务失败，我如何召回&QUOT；TASK&QOOT；？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！