Как я могу поделиться asyncio.Queue между несколькими потоками? - PullRequest
Новая
       7

Как я могу поделиться asyncio.Queue между несколькими потоками?

0 голосов
/ 24 сентября 2018

Этот вопрос отличается от Есть ли способ использовать asyncio.Queue в нескольких потоках?

У меня есть 2 цикла событий asyncio, работающих в двух разных потоках,Thread1 создает данные для Thread2 через asyncio.Queue () .

Один из потоков выдает исключение: got Future <Future pending> attached to a different loop

Теперь это верно, потому что у меня естьединую очередь, которую я использую в разных циклах. Как я могу разделить очередь между двумя циклами, в двух разных потоках?

Пример кода: 

q = asyncio.Queue()

async def producer(q):
    await asyncio.sleep(3)
    q.put(1)

def prod_work(q):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(producer(q))

async def consumer(q):
    await asyncio.sleep(3)
    res = await q.get()

def cons_work(q):
    loop2 = asyncio.new_event_loop()
    asyncio.set_event_loop(loop2)
    loop2.run_until_complete(consumer(q))

def worker(q):
    # main thread - uses this threads loop  
    prod = threading.Thread(target=prod_work, args=(q,))

    # separate thread - uses NEW loop
    cons = threading.Thread(target=cons_work, args=(q,))

    prod.start()
    cons.start()

worker(q)

трассировка полного стека: 

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/local/Cellar/python/3.7.0/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 917, in _bootstrap_inner
    self.run()
  File "/usr/local/Cellar/python/3.7.0/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "asyncio_examples.py", line 24, in cons_work
    loop2.run_until_complete(consumer(q))
  File "/usr/local/Cellar/python/3.7.0/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 568, in run_until_complete
    return future.result()
  File "asyncio_examples.py", line 18, in consumer
    res = await q.get()
  File "/usr/local/Cellar/python/3.7.0/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/queues.py", line 159, in get
    await getter
RuntimeError: Task <Task pending coro=<consumer() running at asyncio_examples.py:18> cb=[_run_until_complete_cb() at /usr/local/Cellar/python/3.7.0/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py:150]> got Future <Future pending> attached to a different loop
...