백그라운드 스레드와 함께 사용하면 비동기 대기열이 중단됩니다.

Question

asyncio.Queue은 읽을 수있는 동일한 스레드에서만 푸시 될 수 있습니다. 예를 들어 :

import asyncio 
from threading import Thread 
import time 

q = asyncio.Queue() 

def produce(): 
    for i in range(100): 
     q.put_nowait(i) 
     time.sleep(0.1) 

async def consume(): 
    while True: 
     i = await q.get() 
     print('consumed', i) 

Thread(target=produce).start() 
asyncio.get_event_loop().run_until_complete(consume()) 

는 중단 후

consumed 0 

를 인쇄합니다. 내가 뭘 놓치고 있니?

Answer 1

call asyncio methods from another thread을 직접 입력 할 수 없습니다.

중 하나가 loop.call_soon_threadsafe를 사용

loop.call_soon_threadsafe(q.put_nowait, i) 

또는 asyncio.run_coroutine_threadsafe :

loop는 주 스레드에 asyncio.get_event_loop() 에 의해 반환되는 루프

future = asyncio.run_coroutine_threadsafe(q.put(i), loop) 

.

Answer 2

janus 라이브러리는 대기열에 비동기 이벤트 루프와 클래식 스레드 코드의 두 가지 얼굴을 제공합니다.

README가 라이브러리를 잘 설명한다고 생각합니다.