Python asyncio 대기열 예제

asyncio.queue는 비동기 작업 간의 안전한 통신을위한 큐 도구입니다. 1. 생산자는 Await Queue.put (항목)을 통해 데이터를 추가하고 소비자는 Await Queue.get ()를 사용하여 데이터를 얻습니다. 2. 프로세스가 완료 될 때마다 queue.join ()가 모든 작업이 완료되기까지 기다릴 수 있도록 queue.task_done ()을 호출해야합니다. 3. 소비자가 중지하도록 통지하기 위해 최종 신호로 아무것도 사용하지 않습니다. 4. 여러 소비자 인 경우 작업을 취소하기 전에 다수의 종말 신호를 보내거나 모든 작업이 처리되었습니다. 5. 큐는 용량을 제한하기 위해 최대 규모의 설정을 지원하고, 작동을 자동으로 매달리고 이벤트 루프를 차단하지 않습니다. 이 프로그램은 최종적으로 여전히 대기중인 소비자 작업을 취소 ()를 통해 종료하여 효율적인 비동기 생산자 소비자 모드를 완전히 실현합니다.

아래는 비동기 작업에서 생산자 소비자 모드 커뮤니케이션에 대기열을 사용하는 방법을 보여주는 실용적인 Python asyncio.Queue 예제입니다.

asyncio.Queue 란 무엇입니까?

asyncio.Queue 스레드-안전한 것으로, 여러 코 루틴간에 데이터를 안전하게 전달하는 데 사용되는 비동기식 보안 대기열입니다. 생산자가 데이터를 생성하고 소비자가 프로세스 데이터를 생성하는 시나리오에서 일반적으로 사용됩니다.

기본 예 : 생산자 및 소비자

 Asyncio 가져 오기
무작위로 가져옵니다

Async def producer (Queue, N) :
    범위 (n)의 i를 위해 :
        항목 = f "항목-{i}"
        대기 시간 대기
        print (f "생산 : {item}")
        # 비동기식 시간 소모 시뮬레이션 (예 : 네트워크 요청, IO)
        Asyncio.sleep (random.uniform (0.1, 0.5))를 기다립니다.
    # 종말 신호 보내기 대기 시간을 보내십시오.
    print ( "프로듀서 완성.")

비동기 DEF 소비자 (대기열) :
    사실이지만 :
        item = await queue.get ()
        항목이 없으면 :
            # 끝 신호를 받고 다른 소비자가 queue.task_done ()를받을 수 있도록 다시 한번하지 않습니다.
            부서지다
        print (f "소비 : {item}")
        Asyncio.sleep (random.uniform (0.2, 0.7)) # 처리 시간 큐 시뮬레이션 .task_done () # 마크 작업 완료 인쇄 ( "소비자 완료")

Async def main () :
    대기열 = asyncio.queue ()
    num_items = 5

    # 생산자 및 소비자는 동시에 Producer_Task = asyncio.create_task (Producer (Queue, Num_items))
    소비자 _task = asyncio.create_task (소비자 (큐))

    # 소비자가 모든 작업을 완료 할 때까지 기다립니다 (없음 포함)
    제작자 _task를 기다리고 있습니다
    Await Queue.join () # 모든 작업이 Task_done으로 표시 될 때까지 기다립니다.
    소비자 _task.cancel () # 소비자를 수동으로 취소합니다 (더 많은 작업을 기다리고 있기 때문에)

# run asyncio.run (main ())

키 포인트 설명

✅ queue.put() 및 queue.get()

• 모두가 기다릴 수 있으며 이벤트 루프를 차단하지 않습니다.
• 큐가 가득 차면 put() 가 매달려 있습니다. 대기열이 비어있을 때 get() 매달린다.

✅ queue.task_done()

• 소비자가 작업 처리를 완료 할 때마다 task_done() 호출해야합니다.
• queue.join() 모든 작업이 task_done() 로 표시되기를 기다립니다.

✅ 끝 신호로 None 사용하지 않습니다

• 소비자에게 중지하라고 통지하십시오.
• 여러 소비자가 최종 신호를 처리하도록주의하십시오 (여러 소비자가 처리하지 않도록하십시오).

✅ queue.join() vs task.cancel()

• queue.join() 은 모든 작업이 완료되기를 기다립니다.
• 소비자는 무한 루프이며 cancel() 로 중지해야합니다.

여러 소비자 예제 (선택 사항)

 Async def main_multiple_consumers () :
    대기열 = asyncio.queue ()
    num_items = 10
    NUM_CONSUMERS = 3

    # 여러 소비자 소비자 생성 = [asyncio.create_task (Queue (Queue)) _ in range (num_consumers)]
    producer_task = asyncio.create_task (producer (queue, num_items))

    제작자 _task를 기다리고 있습니다
    Await Queue.join () # 모든 품목이 처리 될 때까지 기다려 # 모든 소비자를 취소합니다 (더 많은 작업을 기다리고 있습니다)
    소비자의 C의 경우 :
        c.cancel ()
    # 그들이 취소 될 때까지 기다릴 때까지 Asyncio.gather (*소비자, return_exceptions = true)

# run # asyncio.run (main_multiple_consumers ())

팁

• 기본적으로 대기열의 크기 제한이 없습니다. 용량은 asyncio.Queue(maxsize=5) 에 의해 제한 될 수 있습니다.
• 대기열이 가득 차면 put() 공간이있을 때까지 자동으로 대기합니다.
• 오류가 없다고 확신하지 않는 한, 코 루틴에서 .get_nowait() 또는 .get_nowaiit () 또는 .put_nowait() 사용하지 마십시오. 예외가 발생하지 않습니다.

기본적으로 그게 다야. asyncio.Queue 비동기 크롤러, 작업 일정 및 메시지 처리 시스템을 작성하는 일반적인 도구입니다. 마스터 링은 효율적인 비동기 프로그램을 작성하는 데 매우 도움이됩니다.

위 내용은 Python asyncio 대기열 예제의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!