Asyncio.gather와 Asyncio.잠깐만요.

Asyncio.gather와 Asyncio.잠깐만요.

asyncio.gather 유사한 용도가 있는 것 같습니다.실행/대기하고 싶은 비동기 작업이 많이 있습니다(다음 작업이 시작되기 전에 완료될 때까지 기다릴 필요는 없습니다.구문도 다르고 세부 사항도 다르지만, 기능이 이렇게 겹치는 두 가지 기능이 있다는 것은 매우 어색한 것 같습니다.제가 무엇을 빠뜨리고 있나요?

일반적인 경우("많은 작업에 대해 실행 및 결과 가져오기")와 비슷하지만 각 기능에는 다음과 같은 특정 기능이 있습니다.

asyncio.gather()

상위 수준의 작업을 그룹화할 수 있는 미래 인스턴스를 반환합니다.

import asyncio
from pprint import pprint

import random


async def coro(tag):
    print(">", tag)
    await asyncio.sleep(random.uniform(1, 3))
    print("<", tag)
    return tag


loop = asyncio.get_event_loop()

group1 = asyncio.gather(*[coro("group 1.{}".format(i)) for i in range(1, 6)])
group2 = asyncio.gather(*[coro("group 2.{}".format(i)) for i in range(1, 4)])
group3 = asyncio.gather(*[coro("group 3.{}".format(i)) for i in range(1, 10)])

all_groups = asyncio.gather(group1, group2, group3)

results = loop.run_until_complete(all_groups)

loop.close()

pprint(results)

내 은 콜을 통해 할 수 .group2.cancel()또는 심지어all_groups.cancel()도해 주세요..gather(..., return_exceptions=True) ,

asyncio.wait()

첫 번째 작업 완료 후 또는 지정된 타임아웃 후 정지 대기 기능을 지원하여 작업 정밀도를 낮춥니다.

import asyncio
import random


async def coro(tag):
    print(">", tag)
    await asyncio.sleep(random.uniform(0.5, 5))
    print("<", tag)
    return tag


loop = asyncio.get_event_loop()

tasks = [coro(i) for i in range(1, 11)]

print("Get first result:")
finished, unfinished = loop.run_until_complete(
    asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED))

for task in finished:
    print(task.result())
print("unfinished:", len(unfinished))

print("Get more results in 2 seconds:")
finished2, unfinished2 = loop.run_until_complete(
    asyncio.wait(unfinished, timeout=2))

for task in finished2:
    print(task.result())
print("unfinished2:", len(unfinished2))

print("Get all other results:")
finished3, unfinished3 = loop.run_until_complete(asyncio.wait(unfinished2))

for task in finished3:
    print(task.result())

loop.close()

asyncio.wait입니다.asyncio.gather.

알 수 있듯이, 「」는 「」입니다.asyncio.gather주로 결과 수집에 초점을 맞추고 있습니다.많은 미래를 기다리고 결과를 정해진 순서로 반환합니다.

asyncio.wait 것이 미해결 과제를 입니다.또한 결과를 직접 제공하는 대신 완료 및 보류 중인 작업을 제공합니다.값을 수동으로 수집해야 합니다.

모든 첫 번째 가 끝날 할 수 .wait.

매우 중요한 차이점은 예외에 관한 이들 2가지 기능의 기본 동작입니다.

---

이 예를 사용하여 예외를 발생시키는 코루틴을 시뮬레이션합니다.

import asyncio
import random


async def a_flaky_tsk(i):
    await asyncio.sleep(i)  # bit of fuzz to simulate a real-world example

    if i % 2 == 0:
        print(i, "ok")
    else:
        print(i, "crashed!")
        raise ValueError

coros = [a_flaky_tsk(i) for i in range(10)]

---

await asyncio.gather(*coros)outputs - outputs - 력 - - -

0 ok
1 crashed!
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 20, in <module>
    asyncio.run(main())
  File "/Users/dev/.pyenv/versions/3.8.2/lib/python3.8/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Users/dev/.pyenv/versions/3.8.2/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 17, in main
    await asyncio.gather(*coros)
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError

바와 같이 지수 1실행하지 못했습니다.

---

★★★★★★★★★★★★★★★★★.await asyncio.wait(coros)중 몇 이 실패해도 실행됩니다.

0 ok
1 crashed!
2 ok
3 crashed!
4 ok
5 crashed!
6 ok
7 crashed!
8 ok
9 crashed!
Task exception was never retrieved
future: <Task finished name='Task-10' coro=<a_flaky_tsk() done, defined at /Users/dev/PycharmProjects/trading/xxx.py:6> exception=ValueError()>
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError
Task exception was never retrieved
future: <Task finished name='Task-8' coro=<a_flaky_tsk() done, defined at /Users/dev/PycharmProjects/trading/xxx.py:6> exception=ValueError()>
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError
Task exception was never retrieved
future: <Task finished name='Task-2' coro=<a_flaky_tsk() done, defined at /Users/dev/PycharmProjects/trading/xxx.py:6> exception=ValueError()>
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError
Task exception was never retrieved
future: <Task finished name='Task-9' coro=<a_flaky_tsk() done, defined at /Users/dev/PycharmProjects/trading/xxx.py:6> exception=ValueError()>
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError
Task exception was never retrieved
future: <Task finished name='Task-3' coro=<a_flaky_tsk() done, defined at /Users/dev/PycharmProjects/trading/xxx.py:6> exception=ValueError()>
Traceback (most recent call last):
  File "/Users/dev/PycharmProjects/trading/xxx.py", line 12, in a_flaky_tsk
    raise ValueError
ValueError

---

물론, 이 동작은, 다음의 어느쪽인가를 사용해 변경할 수 있습니다.

asyncio.gather(..., return_exceptions=True)

또는,

asyncio.wait([...], return_when=asyncio.FIRST_EXCEPTION)

---

하지만 여기서 끝나지 않아요!

★★★★★★Task exception was never retrieved를 참조해 주세요.

asyncio.wait()를 실행할 때까지 하위 태스크에서 예외를 다시 생성하지 않습니다.await일 뿐, 할 수 (로그내의 트레이스는 일 뿐입니다).

done, pending = await asyncio.wait(coros)
for tsk in done:
    try:
        await tsk
    except Exception as e:
        print("I caught:", repr(e))

출력 -

0 ok
1 crashed!
2 ok
3 crashed!
4 ok
5 crashed!
6 ok
7 crashed!
8 ok
9 crashed!
I caught: ValueError()
I caught: ValueError()
I caught: ValueError()
I caught: ValueError()
I caught: ValueError()

---

, 외, 외, 외의 예외를 잡기 는, 외, 외, on, on, on, on, on, on의 예외를 잡기 위해서,asyncio.gather() - , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , .

results = await asyncio.gather(*coros, return_exceptions=True)
for result_or_exc in results:
    if isinstance(result_or_exc, Exception):
        print("I caught:", repr(result_or_exc))

(기존과 같은 출력)

또한 목록을 지정하는 것만으로 wait()에 coroutines 그룹을 지정할 수 있습니다.

result=loop.run_until_complete(asyncio.wait([
        say('first hello', 2),
        say('second hello', 1),
        say('third hello', 4)
    ]))

gather() 내의 그룹화는 여러 개의 코루틴을 지정하는 것만으로 이루어집니다.

result=loop.run_until_complete(asyncio.gather(
        say('first hello', 2),
        say('second hello', 1),
        say('third hello', 4)
    ))

앞서 말한 모든 답변에 덧붙여, 나는 다른 행동들에 대해 말하고 싶다.gather()그리고.wait()취소될 경우를 대비해서요

Gather() 취소

한다면gather()(아직 완료되지 않은) 모든 제출 대기 파일도 취소됩니다.

Wait() 취소

이 경우,wait()ing 태스크가 취소되면 단순히 ing 태스크가CancelledError대기한 태스크는 그대로 유지됩니다.

간단한 예:

import asyncio


async def task(arg):
    await asyncio.sleep(5)
    return arg


async def cancel_waiting_task(work_task, waiting_task):
    await asyncio.sleep(2)
    waiting_task.cancel()
    try:
        await waiting_task
        print("Waiting done")
    except asyncio.CancelledError:
        print("Waiting task cancelled")

    try:
        res = await work_task
        print(f"Work result: {res}")
    except asyncio.CancelledError:
        print("Work task cancelled")


async def main():
    work_task = asyncio.create_task(task("done"))
    waiting = asyncio.create_task(asyncio.wait({work_task}))
    await cancel_waiting_task(work_task, waiting)

    work_task = asyncio.create_task(task("done"))
    waiting = asyncio.gather(work_task)
    await cancel_waiting_task(work_task, waiting)


asyncio.run(main())

출력:

asyncio.wait()
Waiting task cancelled
Work result: done
----------------
asyncio.gather()
Waiting task cancelled
Work task cancelled

---

신청중

때때로 결합할 필요가 있다.wait()그리고.gather()기능성.예를 들어 하나 이상의 작업이 완료될 때까지 기다린 후 나머지 보류 중인 작업을 취소해야 합니다.waiting취소된 다음 보류 중인 작업도 모두 취소됩니다.

예를 들어 연결 해제 이벤트와 작업 태스크가 있다고 가정해 보겠습니다.그리고 작업 결과를 기다리고 싶은데 연결이 끊겼으면 취소해 주세요.또는 여러 개의 병렬 요청을 하지만 적어도 하나의 응답이 완료되면 다른 모든 요청을 취소하십시오.

다음과 같이 할 수 있습니다.

import asyncio
from typing import Optional, Tuple, Set


async def wait_any(
        tasks: Set[asyncio.Future], *, timeout: Optional[int] = None,
) -> Tuple[Set[asyncio.Future], Set[asyncio.Future]]:
    tasks_to_cancel: Set[asyncio.Future] = set()
    try:
        done, tasks_to_cancel = await asyncio.wait(
            tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED
        )
        return done, tasks_to_cancel
    except asyncio.CancelledError:
        tasks_to_cancel = tasks
        raise
    finally:
        for task in tasks_to_cancel:
            task.cancel()


async def task():
    await asyncio.sleep(5)


async def cancel_waiting_task(work_task, waiting_task):
    await asyncio.sleep(2)
    waiting_task.cancel()
    try:
        await waiting_task
        print("Waiting done")
    except asyncio.CancelledError:
        print("Waiting task cancelled")

    try:
        res = await work_task
        print(f"Work result: {res}")
    except asyncio.CancelledError:
        print("Work task cancelled")


async def check_tasks(waiting_task, working_task, waiting_conn_lost_task):
    try:
        await waiting_task
        print("waiting is done")
    except asyncio.CancelledError:
        print("waiting is cancelled")

    try:
        await waiting_conn_lost_task
        print("connection is lost")
    except asyncio.CancelledError:
        print("waiting connection lost is cancelled")

    try:
        await working_task
        print("work is done")
    except asyncio.CancelledError:
        print("work is cancelled")


async def work_done_case():
    working_task = asyncio.create_task(task())
    connection_lost_event = asyncio.Event()
    waiting_conn_lost_task = asyncio.create_task(connection_lost_event.wait())
    waiting_task = asyncio.create_task(wait_any({working_task, waiting_conn_lost_task}))
    await check_tasks(waiting_task, working_task, waiting_conn_lost_task)


async def conn_lost_case():
    working_task = asyncio.create_task(task())
    connection_lost_event = asyncio.Event()
    waiting_conn_lost_task = asyncio.create_task(connection_lost_event.wait())
    waiting_task = asyncio.create_task(wait_any({working_task, waiting_conn_lost_task}))
    await asyncio.sleep(2)
    connection_lost_event.set()  # <---
    await check_tasks(waiting_task, working_task, waiting_conn_lost_task)


async def cancel_waiting_case():
    working_task = asyncio.create_task(task())
    connection_lost_event = asyncio.Event()
    waiting_conn_lost_task = asyncio.create_task(connection_lost_event.wait())
    waiting_task = asyncio.create_task(wait_any({working_task, waiting_conn_lost_task}))
    await asyncio.sleep(2)
    waiting_task.cancel()  # <---
    await check_tasks(waiting_task, working_task, waiting_conn_lost_task)


async def main():
    print("Work done")
    print("-------------------")
    await work_done_case()
    print("\nConnection lost")
    print("-------------------")
    await conn_lost_case()
    print("\nCancel waiting")
    print("-------------------")
    await cancel_waiting_case()


asyncio.run(main())

출력:

Work done
-------------------
waiting is done
waiting connection lost is cancelled
work is done

Connection lost
-------------------
waiting is done
connection is lost
work is cancelled

Cancel waiting
-------------------
waiting is cancelled
waiting connection lost is cancelled
work is cancelled

언급URL : https://stackoverflow.com/questions/42231161/asyncio-gather-vs-asyncio-wait