Блог

 import queue

qq = queue.Queue()
qq.put('hi')

class MyApp():

    def __init__(self, q):
        self._queue = q

    def _process_item(self, item):
        print(f'Processing this item: {item}')

    def get_item(self):
        try:
            item = self._queue.get_nowait()
            self._process_item(item)
        except queue.Empty:
            pass

    async def listen_for_orders(self):  
        '''
        Asynchronously check the orders queue for new incoming orders
        '''
        while True:
            self.get_item()
            await asyncio.sleep(0)      

a = MyApp(qq)

loop = asyncio.get_event_loop()

loop.run_until_complete(a.listen_for_orders())
  

С использованием Python 3.6.

Я пытаюсь написать обработчик событий, который постоянно прослушивает сообщения в queue и обрабатывает их (в данном случае печатает их). Но он должен быть асинхронным — мне нужно иметь возможность запускать его в терминале (IPython) и вручную передавать данные в queue (по крайней мере, изначально, для тестирования).

Этот код не работает — он блокируется навсегда.

Как мне заставить это выполняться вечно, но возвращать управление после каждой итерации while цикла?

Спасибо.

примечание: Чтобы цикл событий работал с IPython (версия 7.2), я использую этот код из ib_insync библиотеки, я использую эту библиотеку для решения реальной проблемы в примере выше.

Вам нужно создать свою очередь asyncio.Queue и добавлять элементы в очередь потокобезопасным способом. Например:

 qq = asyncio.Queue()

class MyApp():
    def __init__(self, q):
        self._queue = q

    def _process_item(self, item):
        print(f'Processing this item: {item}')

    async def get_item(self):
        item = await self._queue.get()
        self._process_item(item)

    async def listen_for_orders(self):  
        '''
        Asynchronously check the orders queue for new incoming orders
        '''
        while True:
            await self.get_item()

a = MyApp(qq)

loop = asyncio.get_event_loop()

loop.run_until_complete(a.listen_for_orders())
  

Ваш другой поток должен помещать данные в очередь следующим образом:

 loop.call_soon_threadsafe(qq.put_nowait, <item>)
  

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

Это не асинхронная очередь. Вам нужно использовать asyncio.Queue

 qq = queue.Queue()
  

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

 loop.run_until_complete(a.listen_for_orders())
  

Вы прокомментировали:

У меня есть другой поток, который опрашивает внешний сетевой ресурс на предмет данных (интенсивный ввод-вывод) и сбрасывает входящие сообщения в этот поток.

Напишите этот код асинхронно, чтобы у вас был:

 async def run():
    while 1:
        item = await get_item_from_network()
        process_item(item)

loop = asyncio.get_event_loop()
loop.run_until_complete( run() )
  

Если вы не хотите этого делать, то вы можете выполнить пошаговое выполнение цикла, хотя вы и не хотите этого делать.

 import asyncio


def run_once(loop):
    loop.call_soon(loop.stop)
    loop.run_forever()


loop = asyncio.get_event_loop()

for x in range(100):
    print(x)
    run_once(loop)
  

Затем вы просто вызываете свою функцию async, и каждый раз, когда вы вызываете run_once, она проверяет вашу (очередь asyncio) и передает управление вашей функции listen for orders, если в очереди есть элемент.