При использовании asyncio, как вы разрешаете выполнение всех задач перед закрытием цикла событий

Ответ 1

Вы можете восстановить незавершенные задачи и запустить цикл до тех пор, пока они не закончатся, затем закройте цикл или выйдите из программы.

pending = asyncio.Task.all_tasks()
loop.run_until_complete(asyncio.gather(*pending))

• pending - список ожидающих задач.
• asyncio.gather() позволяет подождать сразу несколько задач.

Если вы хотите, чтобы все задачи выполнялись внутри сопрограммы (возможно, у вас есть "главная" сопрограмма), вы можете сделать это таким образом, например:

@asyncio.coroutine
def do_something_periodically():
    while True:
        asyncio.async(my_expensive_operation())
        yield from asyncio.sleep(my_interval)
        if shutdown_flag_is_set:
            print("Shutting down")
            break

    yield from asyncio.gather(*asyncio.Task.all_tasks())

Кроме того, в этом случае, поскольку все задачи создаются в одной сопрограмме, вы уже имеете доступ к задачам:

@asyncio.coroutine
def do_something_periodically():
    tasks = []
    while True:
        tasks.append(asyncio.async(my_expensive_operation()))
        yield from asyncio.sleep(my_interval)
        if shutdown_flag_is_set:
            print("Shutting down")
            break

    yield from asyncio.gather(*tasks)

Ответ 2

Начиная с Python 3.7 в приведенном выше ответе используется несколько устаревших API (asyncio.async и Task.all_tasks, @asyncio.coroutine, yield from и т.д.), И вам лучше использовать это:

import asyncio


async def my_expensive_operation(expense):
    print(await asyncio.sleep(expense, result="Expensive operation finished."))


async def do_something_periodically(expense, interval):
    while True:
        asyncio.create_task(my_expensive_operation(expense))
        await asyncio.sleep(interval)


loop = asyncio.get_event_loop()
coro = do_something_periodically(1, 1)

try:
    loop.run_until_complete(coro)
except KeyboardInterrupt:
    coro.close()
    tasks = asyncio.all_tasks(loop)
    expensive_tasks = {task for task in tasks if task._coro.__name__ != coro.__name__}
    loop.run_until_complete(asyncio.gather(*expensive_tasks))

Ответ 3

Вы можете также рассмотреть возможность использования asyncio.shield, хотя, выполнив этот способ, вы не получите ВСЕ выполняющиеся задачи, а только экранированные. Но это все еще может быть полезно в некоторых сценариях.

Кроме того, начиная с Python 3.7 мы также можем использовать высокоуровневый метод API asynio.run здесь. Как разработчик ядра Python, Юрий Селиванов предлагает: https://youtu.be/ReXxO_azV-w?t=636
Примечание: функция asyncio.run была добавлена в asyncio в Python 3.7 на временной основе.

Надеюсь, это поможет!

import asyncio


async def my_expensive_operation(expense):
    print(await asyncio.sleep(expense, result="Expensive operation finished."))


async def do_something_periodically(expense, interval):
    while True:
        asyncio.create_task(my_expensive_operation(expense))
        # using asyncio.shield
        await asyncio.shield(asyncio.sleep(interval))


coro = do_something_periodically(1, 1)

if __name__ == "__main__":
    try:
        # using asyncio.run
        asyncio.run(coro)
    except KeyboardInterrupt:
        print('Cancelled!')

Ответ 4

Используйте сопрограмму-обертку, которая ждет, пока счетчик ожидающих заданий не станет равным 1, прежде чем вернуться.

async def loop_job():
    asyncio.create_task(do_something_periodically())
    while len(asyncio.Task.all_tasks()) > 1:  # Any task besides loop_job() itself?
        await asyncio.sleep(0.2)

asyncio.run(loop_job())

Ответ 5

Я не уверен, что это то, о чем вы просили, но у меня была похожая проблема, и вот окончательное решение, которое я нашел.

Код совместим с Python 3 и использует только общедоступные API-интерфейсы asyncio (что означает отсутствие хаков _coro и устаревших API).

import asyncio

async def fn():
  await asyncio.sleep(1.5)
  print('fn')

async def main():
    print('main start')
    asyncio.create_task(fn()) # run in parallel
    await asyncio.sleep(0.2)
    print('main end')


def async_run_and_await_all_tasks(main):
  def get_pending_tasks():
      tasks = asyncio.Task.all_tasks()
      pending = [task for task in tasks if task != run_main_task and not task.done()]
      return pending

  async def run_main():
      await main()

      while True:
          pending_tasks = get_pending_tasks()
          if len(pending_tasks) == 0: return
          await asyncio.gather(*pending_tasks)

  loop = asyncio.new_event_loop()
  run_main_coro = run_main()
  run_main_task = loop.create_task(run_main_coro)
  loop.run_until_complete(run_main_task)

# asyncio.run(main()) # does not print from fn task, because main finishes earlier
async_run_and_await_all_tasks(main)

вывод (как и ожидалось):

main start
main end
fn

Эта функция async_run_and_await_all_tasks заставит python вести себя в режиме nodejs: выходить только при отсутствии незавершенных задач.