Python 上的非阻塞生成器答案

【问题标题】：Non-blocking generator on PythonPython 上的非阻塞生成器
【发布时间】：2013-11-13 05:26:09
【问题描述】：

我在 QT 应用程序中使用 requests 模块中的生成器函数，与请求流示例中的几乎相同：

import json
import requests

def get_stream():
    r = requests.get('http://httpbin.org/stream/20', stream=True)
    for line in r.iter_lines():
        if line:
            yield json.loads(line)

def consume_stream():
    for message in get_stream():
       #do something

但是，当没有传入响应时（例如，来自 Twitter 流 API 的不规则传入推文），生成器 get_stream 将阻止 consume_stream 方法。

这可能发生在任何情况下，生成器不会立即产生，而是必须等待传入的消息等，因此会阻塞消费者。

在 Python 中是否有任何模式可以让您以非阻塞方式使用生成器，即如果生成器产生，则处理它的结果，否则在下一个结果出现之前执行其他操作？

【问题讨论】：

您如何知道继续使用生成器？
抱歉回复晚了：consume_stream-方法基本上更新了一个 GUI-Widget (TreeModel) 并且“Abort”-Button 可用。现在，当get_stream 没有产生任何东西（例如等待传入消息/推文）时，此按钮会阻塞。我无法修改iter_lines-方法，但如果没有传入的推文，我想产生类似“等待推文......”的内容，并在有新推文到达时立即产生推文。因此，问题归结为一个问题是可以在消费者 for 循环中控制生成器的消耗还是必须等待......

标签： python generator yield

【解决方案1】：

您可以从 python 3.6 https://www.python.org/dev/peps/pep-0525/987654321@开始使用异步生成器

import json
import requests

async def get_stream():
    r = requests.get('http://httpbin.org/stream/20', stream=True)
    for line in r.iter_lines():
        if line:
            yield json.loads(line)

async def consume_stream():
   await for message in get_stream():
       #do something

【讨论】：

【解决方案2】：

如果您控制生成器函数，一种解决方案是让它在超时后抛出异常。也许是这样的：

def get_stream(timeout=None):
    while message=read_message(timeout=timout):
        yield message

如果发生超时条件，则让 read_message 抛出 TimeOutException 或其他东西。

当然，您仍然需要处理何时/如何重试/恢复的后勤问题。

【讨论】：

【解决方案3】：

正如西蒙在评论中所问的那样，它不能像您在示例中描述的那样简单。有很多细节需要注意。根据您的用例，有不同的解决方案或多或少有意义。你并没有提供太多关于你真正想做的事情的细节，所以我只是将你发送到http://twistedmatrix.com/trac/wiki/QTReactor 作为示例。有不同的解决方案/框架实现异步消息队列。我认为，这就是您要寻找的。p>

【讨论】：

【解决方案4】：

看看producer-consumer pattern。它通常在 python 中使用Queue 实现。

生产者，通常运行在一个线程或另一个进程中（Queue 支持任何一个），只是将消息放入队列中。消费者，只要有感觉，就会从队列中弹出消息。此操作支持timeout 参数。

【讨论】：