假设我有一些StringIO
(来自cStringIO
)。我想从中读取缓冲区,直到遇到某些字符/字节,例如“ Z”,所以:
stringio = StringIO('ABCZ123')
buf = read_until(stringio, 'Z') # buf is now 'ABCZ'
# strinio.tell() is now 4, pointing after 'Z'
用Python最快的方法是什么?谢谢
我非常失望,这个问题在堆栈溢出时仅得到一个答案,因为这是一个有趣且相关的问题。无论如何,由于只有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函数。
这比timeit
进行检查。
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)
另一种实现方法是使用generator:
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)
我做了一些效率测试。所描述技术的性能几乎相同:
#!/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')))