使用 Python“请求”库读取流式 http 响应

Question

我正在尝试使用Kubernetes 提供的事件流 使用requests 模块的api。我遇到了看起来像 缓冲问题：requests 模块似乎滞后了一个事件。

我的代码看起来像这样：

r = requests.get('http://localhost:8080/api/v1beta1/watch/services',
                 stream=True)

for line in r.iter_lines():
    print 'LINE:', line

由于 Kubernetes 发出事件通知，此代码将仅显示 新事件进入时发出的最后一个事件，这使得它 对于需要响应服务的代码几乎完全没用 添加/删除事件。

我已经通过在子进程中生成 curl 而不是使用 requests 库：

p = subprocess.Popen(['curl', '-sfN',
                      'http://localhost:8080/api/watch/services'],
                     stdout=subprocess.PIPE,
                     bufsize=1)

for line in iter(p.stdout.readline, b''):
    print 'LINE:', line

这可行，但会牺牲一些灵活性。有没有办法 避免requests 库的这种缓冲问题？

Answer 1

此行为是由于 iter_lines 的错误实现 requests 库中的方法。

iter_lines 迭代 chunk_size 块中的响应内容 使用iter_content 迭代器的数据。如果少于 chunk_size 可用于从远程读取的数据字节 服务器（通常在读取最后一行时会出现这种情况 输出），读操作将阻塞直到chunk_size字节 数据可用。

我已经编写了自己的 iter_lines 例程，并且运行正常：

import os


def iter_lines(fd, chunk_size=1024):
    '''Iterates over the content of a file-like object line-by-line.'''

    pending = None

    while True:
        chunk = os.read(fd.fileno(), chunk_size)
        if not chunk:
            break

        if pending is not None:
            chunk = pending + chunk
            pending = None

        lines = chunk.splitlines()

        if lines and lines[-1]:
            pending = lines.pop()

        for line in lines:
            yield line

    if pending:
        yield(pending)

这是可行的，因为os.read 将返回少于chunk_size 字节 而不是等待缓冲区填满。