Python StreamIO 从同一个流中读取和写入

Question

我正在寻找一种 Python 方式来读取和写入流（在 IOBase 层次结构中），而无需通过 seek() 和 tell() 管理流位置。这类似于套接字流的预期工作方式，其中接收端将能够在接收字节时连续读取。

我尝试过使用 BufferedReader 和 BufferedWriter 都附加到同一个原始流，但一个操作会影响另一个流的位置。 seek()/tell() 似乎直接传递到底层原始流。是否有其他 IOBase 类型可用于有效地充当 IO 流，其中支持并发输入和输出而无需管理流位置（类似于 C++ stringstream >> 和

谢谢！

>>> import io
>>> buf = io.BytesIO()
>>> r = io.BufferedReader(buf)
>>> w = io.BufferedWriter(buf)
>>> w.write('foo bar')
7L
>>> r.read(1)
''
>>> r.tell()
0L
>>> w.flush()
>>> r.tell()
7L
>>> r.flush()
>>> r.tell()
7L
>>> w.tell()
7L
>>> buf.tell()
7L

Answer 1

我认为以前的答案不是线程安全的......尽管可以这样做，例如：

class BytesLoop:
    def __init__(self, s=b''):
        self.buffer = s
        self.lock = threading.Lock()

    def read(self, n=-1):
        with self.lock:
            chunk = self.buffer[:n]
            self.buffer = self.buffer[n:]
        return chunk

    def write(self, s):
        with self.lock:
            self.buffer += s

Answer 2

以下简单的字符串示例比上面 Serge 的第二个示例运行得快很多（至少在流中保存的数据相对较小的情况下）并且适用于对同一流的简单读取和写入：

class BytesLoop:
    def __init__(self, s=b''):
        self.buffer = s

    def read(self, n=-1):
        chunk = self.buffer[:n]
        self.buffer = self.buffer[n:]
        return chunk

    def write(self, s):
        self.buffer += s

Answer 3

你不能直接。

套接字类似于一对文件描述符，一个用于读取，一个用于写入。这就是为什么您可以在套接字上进行读写而无需在不同操作之间使用查找的原因。

如果您真的想用StringIO 或BytesIO 模拟套接字，只需构建一个包含一对或它们的自定义类。

可能是这样的：

class BytesSocket(io.IOBase):
    def __init__(self, inputText):
        self.input = io.BytesIO(inputText)
        self.output = io.BytesIO()
    def read(self, n=-1):
        return self.input.read(n)
    def readinto(self, b):
        return self.input.readinto(b)
    def write(self, b):
        return self.output.write(b)
    def getoutvalue(self):
        return self.output.getvalue()

如果您需要一个环回伪套接字（阅读之前编写的内容），您可以使用：

class BytesLoop(io.IOBase):
    def __init__(self, inputText=''):
        self.buf = inputText
    def read(self, n=-1):
        inp = io.BytesIO(self.buf)
        b = inp.read(n)
        self.buf = self.buf[len(b):]
        return b
    def readinto(self, b):
        inp = io.BytesIO(buf)
        l = inp.readinto(b)
        self.buf = self.buf[l:]
        return l
    def write(self, b):
        outp = io.BytesIO()
        l = outp.write(b)
        self.buf += outp.getvalue()
        return l
    def getvalue(self):
        return self.buf

由于str（或unicode）是一个不可变的序列，它需要为每个io操作重新编写。您可以轻松地使用可变的字符列表来代替。您可以使用l = [c for c in s ] 将字符串转换为列表，而使用s = ''.join(l) 可以完成相反的操作。