与工作者"线程"相同的asyncio.Queues

Question

我正在试图弄清楚如何移植一个线程程序来使用asyncio.我有很多代码可以围绕几个标准库同步Queues,基本上是这样的:

import queue, random, threading, time

q = queue.Queue()

def produce():
    while True:
        time.sleep(0.5 + random.random())  # sleep for .5 - 1.5 seconds
        q.put(random.random())

def consume():
    while True: 
        value = q.get(block=True)
        print("Consumed", value)

threading.Thread(target=produce).start()
threading.Thread(target=consume).start()

一个线程创建值(可能是用户输入),另一个线程用它们做某事.关键是这些线程在有新数据之前一直处于空闲状态,此时它们会唤醒并对其执行某些操作.

我正在尝试使用asyncio实现这种模式,但我似乎无法弄清楚如何让它"去".

我的尝试看起来或多或少都像这样(并且根本不做任何事情).

import asyncio, random

q = asyncio.Queue()

@asyncio.coroutine
def produce():
    while True: 
        q.put(random.random())
        yield from asyncio.sleep(0.5 + random.random())

@asyncio.coroutine
def consume():
    while True:
        value = yield from q.get()
        print("Consumed", value)

# do something here to start the coroutines. asyncio.Task()? 

loop = asyncio.get_event_loop()
loop.run_forever()

我尝试过使用协同程序的变体,不使用它们,在Tasks中包装东西,试图让它们创建或返回期货等.

我开始认为我对如何使用asyncio有错误的想法(也许这种模式应该以我不知道的不同方式实现).任何指针将不胜感激.

Answer 1

对,就是这样.任务是你的朋友:

import asyncio, random

q = asyncio.Queue()

@asyncio.coroutine
def produce():
    while True:
        yield from q.put(random.random())
        yield from asyncio.sleep(0.5 + random.random())

@asyncio.coroutine
def consume():
    while True:
        value = yield from q.get()
        print("Consumed", value)


loop = asyncio.get_event_loop()
loop.create_task(produce())
loop.create_task(consume())
loop.run_forever()

asyncio.ensure_future 也可以用于任务创建.

请记住:q.put()是一个协程,所以你应该使用yield from q.put(value).

UPD

转自asyncio.Task()/ asyncio.async()新品牌的API loop.create_task()和asyncio.ensure_future()示例所示.

Answer 2

这是我在制作中使用的内容,转移到要点:https://gist.github.com/thehesiod/7081ab165b9a0d4de2e07d321cc2391d