从 StringIO 读取直到遇到某个字节的快速方法答案

【问题标题】：fast way to read from StringIO until some byte is encountered从 StringIO 读取直到遇到某个字节的快速方法
【发布时间】：2012-01-06 23:11:15
【问题描述】：

假设我有一些StringIO（来自cStringIO）。我想从中读取缓冲区，直到遇到一些字符/字节，比如'Z'，所以：

stringio = StringIO('ABCZ123')
buf = read_until(stringio, 'Z')  # buf is now 'ABCZ'
# strinio.tell() is now 4, pointing after 'Z'

在 Python 中最快的方法是什么？谢谢

【问题讨论】：

标签： python performance optimization stream stringio

【解决方案1】：

我很失望这个问题只有一个关于堆栈溢出的答案，因为这是一个有趣且相关的问题。无论如何，由于只有 ovgolovin 提供解决方案，而且我认为它可能很慢，所以我想到了一个更快的解决方案：

def foo(stringio):
    datalist = []
    while True:
        chunk = stringio.read(256)
        i = chunk.find('Z')
        if i == -1:
            datalist.append(chunk)
        else:
            datalist.append(chunk[:i+1])
            break
        if len(chunk) < 256:
            break
    return ''.join(datalist)

这以块的形式读取 io（可能未在第一个块中找到结束字符）。它非常快，因为没有为每个字符调用 Python 函数，相反，最大限度地使用了 C 编写的 Python 函数。

这比 ovgolovin 的解决方案快 60 倍。我跑了 timeit 来检查它。

【讨论】：

非常好的解决方案！它解决了 Python 在函数调用上的沉重开销。唯一的缺点是您在内存中保留了一个冗余的datalist 对象。可以使用生成器而不是函数重写此代码（join 接受迭代器），因此内存中不会有临时冗余对象。
但是生成器版本有点慢：ideone.com/dQGe5（如果字符串很大（100 万个符号）- 那么生成器版本会快一点）。
对了，你为什么选择256符号块？（为什么不512 或1024？）
我希望最后一点。写chink.find('Z') 不是 Pythonic。可以改写为if 'Z' in chunk: ...
不，这不是最后一点。 chunk[:i] 应该是 chunk[:i+1]（因为我们需要包含 Z）。

【解决方案2】：

i = iter(lambda: stringio.read(1),'Z')
buf = ''.join(i) + 'Z'

这里iter在这种模式下使用：iter(callable, sentinel) -> iterator。

''.join(...) 相当有效。最后添加'Z'''.join(i) + 'Z'的操作不是很好。但可以通过在迭代器中添加'Z' 来解决：

from itertools import chain, repeat

stringio = StringIO.StringIO('ABCZ123')
i = iter(lambda: stringio.read(1),'Z')
i = chain(i,repeat('Z',1))
buf = ''.join(i)

另一种方法是使用生成器：

def take_until_included(stringio):
    while True:
        s = stringio.read(1)
        yield s
        if s=='Z':
            return

i = take_until_included(stringio)
buf = ''.join(i)

我做了一些效率测试。所描述技术的性能几乎相同：

http://ideone.com/dQGe5

【讨论】：

但是'Z'不是从流中获取的，不是吗？
@zaharpopov 不，它被丢弃了。所以我使用+'Z' 和chain(i,repeat('Z',1)) 来解决这个问题。我们知道我们将什么用作哨兵，因此我们可以轻松地将其手动添加到流中。
Спасибо 感谢您的努力，但请看我的回答

【解决方案3】：

#!/usr/bin/env python3
import io


def iterate_stream(stream, delimiter, max_read_size=1024):
    """ Reads `delimiter` separated strings or bytes from `stream`. """
    empty = '' if isinstance(delimiter, str) else b''
    chunks = []
    while 1:
        d = stream.read(max_read_size)
        if not d:
            break
        while d:
            i = d.find(delimiter)
            if i < 0:
                chunks.append(d)
                break
            chunks.append(d[:i+1])
            d = d[i+1:]
            yield empty.join(chunks)
            chunks = []
    s = empty.join(chunks)
    if s:
        yield s


if __name__ == '__main__':
    print(next(iterate_stream(io.StringIO('ABCZ123'), 'Z')))
    print(next(iterate_stream(io.BytesIO(b'ABCZ123'), b'Z')))

【讨论】：