python - 防止 BufferedReader 关闭文件列表的文件

Question

我有一个扩展 BufferedReader 的类，以及一个文件流列表b.close() 为除最后一个流之外的所有流调用，我想保持流打开我该怎么做？

谢谢

class TestReader(BufferedReader):    passdef test(streams):    for stream in streams:        b=TestReader(stream)        do_something(b)    #all the streams except streams[-1] are closed, how do I prevent this?streams=[open('test1.txt','rb'),open('test2.txt','rb')]test(streams)streams.do_something_else()

Answer 1

即使在实现中 BufferedIOBase 类包装了一个 IOBase 对象，它们的接口(interface)是一个流(一切 继承自 IOBase)，所以 IOBase 对象的普通行为是在它们超出范围时关闭自己。 BufferedIOBase 只是将 close() 调用委托(delegate)给底层流。

您不应将 BufferedReader 视为流包装器(尽管这是它的实现方式)，而应将其视为现有流的类型转换。两个流的状态完全绑定(bind)在一起。但是，您可以使用 detach() 解除绑定(bind)流，但这会使 BufferedIOBase 对象无用。

此外，当模式为 rb 时，io.open 已经返回一个 BufferedReader，因此您是双缓冲。您应该改用 io.FileIO。

你有几个选择:

1. 创建一个新的流和一个新的底层文件描述符，并传递文件名而不是流。这是最简单、最安全的选择。

2. 创建原始文件描述符并根据需要从中创建流。这需要一些注意多个流不会同时使用相同的文件描述符。例如:

   fd = os.open('test.txt', os.O_RDONLY)
   file1 = FileIO(fd, 'r', closefd=False)
   file2 = FileIO(fd, 'r', closefd=False)
   
   file1.read(100)
   assert file1.tell() == 100
   file2.read(100)
   assert file1.tell() == 200
   

3. detach() 在您的 BufferedIOBase 对象关闭其流之前底层流。 (记得倒流!)

   def test(streams):
       for stream in streams:
           b=TestReader(stream)
           do_something(b)
           wrappedstream = b.detach()
           assert wrappedstream is stream
   

   您甚至可以在析构函数中实现它:

   class TestReader(BufferedReader):
       def __del__(self):
           self.detach()
           # self.raw will not be closed,
           # rather left in the state it was in at detachment
   

   或者，如果您认为语义有误，则完全禁用 close() 委托(delegate):

   class TestReader(BufferedReader):
       def close(self):
           self.closed = True
   

我不知道你在做什么(可能你需要不同的设计)，但这就是我实现我看到的代码的方式:

from io import FileIO, BufferedReader
import io
import os

class TestReader(BufferedReader):
    pass

def test(streams):
    for stream in streams:
        b = TestReader(stream)

def test_reset(streams):
    """Will try to leave stream state unchanged"""
    for stream in streams:
        pos = stream.tell()
        b = TestReader(stream)
        do_something(b)
        b.detach()
        stream.seek(pos)



filenames = ['test1.txt', 'test2.txt']

# option 1: just make new streams

streams = [FileIO(name, 'r') for name in filenames]
test(streams)
streams = [io.open(name, 'rb') for name in filenames]
#etc


# option 2: use file descriptors
fds = [os.open(name, os.O_RDONLY) for name in filenames]
#closefd = False means "do not close fd on __del__ or __exit__"
#this is only an option when you pass a fd instead of a file name
streams = [FileIO(fd, 'r', closefd=False) for fd in fds]
test(streams)
streams = []
for fd in fds:
    os.lseek(fd, 0, os.SEEK_SET)
    streams.append(io.open(fd, 'rb', closefd=False))
    # you can also .seek(0) on the BufferedReader objects
    # instead of os.lseek on the fds


# option 3: detach

streams = [FileIO(name, 'r') for name in filenames]
test_reset(streams)
# streams[*] should still be in the same state as when you passed it in