asyncio queue потребительская сопрограмма

Это должен быть asyncio.Task?

Да, создайте его с помощью asyncio.ensure_future или loop.create_task.

Что делать, если очередь становится пустой из-за того, что в течение нескольких секунд данные не принимаются?

Просто используйте queue.get, чтобы дождаться товар доступен:

async def consume(queue):
    while True:
        item = await queue.get()
        print(item)

Есть ли более чистый способ, чем использование глобальной переменной для моей очереди?

Да, просто передайте его как аргумент сопрограмме потребителя и протоколу потока:

class StreamProtocol(asyncio.Protocol):
    def __init__(self, loop, queue):
        self.loop = loop
        self.queue = queue

    def data_received(self, data):
        for message in data.decode().splitlines():
            self.queue.put_nowait(message.rstrip())

    def connection_lost(self, exc):
        self.loop.stop()

Как я могу убедиться, что мой потребитель не остановится (run_until_complete)?

После закрытия соединения используйте queue.join ждать, пока очередь не опустеет.

Полный пример:

loop = asyncio.get_event_loop()
queue = asyncio.Queue()
# Connection coroutine
factory = lambda: StreamProtocol(loop, queue)
connection = loop.create_connection(factory, '127.0.0.1', '42')
# Consumer task
consumer = asyncio.ensure_future(consume(queue))
# Set up connection
loop.run_until_complete(connection)
# Wait until the connection is closed
loop.run_forever()
# Wait until the queue is empty
loop.run_until_complete(queue.join())
# Cancel the consumer
consumer.cancel()
# Let the consumer terminate
loop.run_until_complete(consumer)
# Close the loop
loop.close()

В качестве альтернативы вы также можете использовать потоки:

async def tcp_client(host, port, loop=None):
    reader, writer = await asyncio.open_connection(host, port, loop=loop)
    async for line in reader:
        print(line.rstrip())
    writer.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(tcp_client('127.0.0.1', 42, loop))
loop.close()