python - 从 Python 队列中读取行

Question

我正在涉足 Python 线程。我创建了一个供应商线程，它通过队列从 *nix(串行)/dev 返回字符/行数据。

作为练习，我想一次一行地使用队列中的数据(使用“\n”作为行终止符)。

我当前的(简单的)解决方案是一次只将 1 个字符放入队列，因此消费者一次只会 get() 一个字符。 (这是一个安全的假设吗？)这种方法目前允许我执行以下操作:

...
return_buffer = []
while True:
    rcv_data = queue.get(block=True)
    return_buffer.append(rcv_data)        
    if rcv_data == "\n":
        return return_buffer

这似乎可行，但是当我一次 put() 2 个字符时，我肯定会导致它失败。

我想让接收逻辑更通用并且能够处理多字符 put()。

我的下一个方法是 rcv_data.partition("\n")，将“剩余”放在另一个缓冲区/列表中，但这需要在队列旁边处理临时缓冲区。(我想另一种方法是一次只 put() 一行，但这有什么乐趣呢？)

是否有一种更优雅的方式来一次读取队列中的一行？

Answer 1

这可能是生成器的一个很好的用途。它会在 yield 后准确地从停止的地方开始，因此减少了您需要的存储量和缓冲区交换量(我无法谈论它的性能)。

def getLineGenerator(queue, splitOn):
    return_buffer = []
    while True:
        rcv_data = queue.get(block=True) # We can pull any number of characters here.
        for c in rcv_data:
            return_buffer.append(c)
            if c == splitOn:
                yield return_buffer
                return_buffer = []


gen = getLineGenerator(myQueue, "\n")
for line in gen:
    print line.strip()

编辑:

一旦 J.F. Sebastian 指出行分隔符可以是多字符，我也必须解决这个问题。我还使用了 jdi 的回答中的 StringIO。同样，我无法谈论效率，但我相信它在所有情况下都是正确的(至少在我能想到的情况下)。这是未经测试的，因此可能需要一些调整才能实际运行。感谢 J.F. Sebastian 和 jdi 的回答，最终导致了这个问题。

def getlines(chunks, splitOn="\n"):
    r_buffer = StringIO()
    for chunk in chunks
        r_buffer.write(chunk)
        pos = r_buffer.getvalue().find(splitOn) # can't use rfind see the next comment
        while pos != -1: # A single chunk may have more than one separator
            line = r_buffer.getvalue()[:pos + len(splitOn)]
            yield line
            rest = r_buffer.getvalue().split(splitOn, 1)[1]
            r_buffer.seek(0)
            r_buffer.truncate()
            r_buffer.write(rest)
            pos = rest.find(splitOn) # rest and r_buffer are equivalent at this point. Use rest to avoid an extra call to getvalue

    line = r_buffer.getvalue();
    r_buffer.close() # just for completeness
    yield line # whatever is left over.

for line in getlines(iter(queue.get, None)): # break on queue.put(None)
    process(line)