Python Asyncio - 运行时错误 : Cannot close a running event loop

Question

我正在尝试解决此错误:RuntimeError: Cannot close a running event loop在我的异步过程中。我相信它正在发生，因为在任务仍然挂起时出现故障，然后我尝试关闭事件循环。我想我需要在关闭事件循环之前等待剩余的响应，但我不确定如何在我的具体情况下正确完成。

 def start_job(self):

        if self.auth_expire_timestamp < get_timestamp():
            api_obj = api_handler.Api('Api Name', self.dbObj)
            self.api_auth_resp = api_obj.get_auth_response()
            self.api_attr = api_obj.get_attributes()


        try:
            self.queue_manager(self.do_stuff(json_data))
        except aiohttp.ServerDisconnectedError as e:
            logging.info("Reconnecting...")
            api_obj = api_handler.Api('API Name', self.dbObj)
            self.api_auth_resp = api_obj.get_auth_response()
            self.api_attr = api_obj.get_attributes()
            self.run_eligibility()

async def do_stuff(self, data):

    tasks = []

    async with aiohttp.ClientSession() as session:
        for row in data:
            task = asyncio.ensure_future(self.async_post('url', session, row))
            tasks.append(task)
        result = await asyncio.gather(*tasks)
    self.load_results(result)


def queue_manager(self, method):
    self.loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(method)
    self.loop.run_until_complete(future)


async def async_post(self, resource, session, data):
        async with session.post(self.api_attr.api_endpoint + resource, headers=self.headers, data=data) as response:
            resp = []
            try:
                headers = response.headers['foo']
                content = await response.read()
                resp.append(headers)
                resp.append(content)
            except KeyError as e:
                logging.error('KeyError at async_post response')
                logging.error(e)
        return resp


def shutdown(self):
    //need to do something here to await the remaining tasks and then I need to re-start a new event loop, which i think i can do, just don't know how to appropriately stop the current one.
    self.loop.close() 
    return True

我该如何处理错误并正确关闭事件循环，以便我可以开始一个新的并基本上重新启动整个程序并继续。

编辑:

这就是我现在正在尝试的，基于 this SO answer .不幸的是，这个错误很少发生，所以除非我能强制它，否则我将不得不等待，看看它是否有效。在我的 queue_manager方法我把它改成这样:

try:
   self.loop.run_until_complete(future)
except Exception as e:
   future.cancel()
   self.loop.run_until_complete(future)
   future.exception()

更新:

我摆脱了 shutdown()方法并将其添加到我的 queue_manager()相反，它似乎可以正常工作:

  try:
        self.loop.run_until_complete(future)
    except Exception as e:
        future.cancel()
        self.check_in_records()
        self.reconnect()
        self.start_job()
        future.exception()

Answer 1

要回答最初所说的问题，无需close()。一个正在运行的循环，您可以为整个程序重用相同的循环。

鉴于更新中的代码，您的 queue_manager可能看起来像这样:

try:
    self.loop.run_until_complete(future)
except Exception as e:
    self.check_in_records()
    self.reconnect()
    self.start_job()

取消 future没有必要，据我所知没有效果。这与 referenced answer 不同。对 KeyboardInterrupt 有特别 react , 特别是因为它是由 asyncio 本身提出的。 KeyboardInterrupt可以通过 run_until_complete 传播没有真正完成的 future 。在 asyncio 中正确处理 Ctrl-C 非常困难，甚至是不可能的(详情请参阅 here)，但幸运的是，问题根本与 Ctrl-C 无关，而是与协程引发的异常有关。 (注意 KeyboardInterrupt 不继承自 Exception ，所以在 Ctrl-C 的情况下，except 正文甚至不会执行。)

I was canceling the future because in this instance there are remaining tasks pending and i want to essentially remove those tasks and start a fresh event loop.

这是一件正确的事情，但是(更新的)问题中的代码只是取消了一个 future ，那个已经传递给 run_until_complete .回想一下，future 是稍后提供的结果值的占位符。一旦提供了值，就可以通过调用 future.result() 来检索它。 .如果 future 的“值(value)”是一个异常(exception)， future.result()将引发该异常。 run_until_complete有一个约定，只要给定的 future 产生一个值，它就会运行事件循环，然后它返回那个值。如果“值”实际上是要引发的异常，则 run_until_complete将重新提高它。例如:

loop = asyncio.get_event_loop()
fut = loop.create_future()
loop.call_soon(fut.set_exception, ZeroDivisionError)
# raises ZeroDivisionError, as that is the future's result,
# manually set
loop.run_until_complete(fut)

当所讨论的 future 实际上是 Task ，一个特定于异步的对象，它将协程包装到 Future ，这样 future 的结果就是协程返回的对象。如果协程引发异常，则检索结果将重新引发异常， run_until_complete 也会如此。 :

async def fail():
    1/0

loop = asyncio.get_event_loop()
fut = loop.create_task(fail())
# raises ZeroDivisionError, as that is the future's result,
# because the coroutine raises it
loop.run_until_complete(fut)

处理任务时， run_until_complete Finish 意味着协程也已经完成，返回值或引发异常，由 run_until_complete 确定回归或提升。

另一方面，通过安排要恢复的任务和 await 来取消任务。暂停它以提高 CancelledError 的表达式.除非任务专门捕获并抑制此异常(行为良好的 asyncio 代码不应该这样做)，否则任务将停止执行并且 CancelledError将成为它的结果。但是，如果协程在 cancel() 时已经完成。被调用，然后 cancel()不能做任何事情，因为没有待处理的 await注入(inject) CancelledError进入。