防止BufferedReader关闭文件列表中的文件

1 投票

1 回答

1554 浏览

提问于 2025-04-17 09:00

我有一个类，它是从BufferedReader这个类扩展出来的，还有一组文件流。除了最后一个流以外，其他的流都调用了b.close()来关闭。我想保持这些流是打开的，我该怎么做呢？

谢谢！


class TestReader(BufferedReader):
    pass

def 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()

资源管理 bufferedreader 文件流管理文件关闭策略

1 个回答

虽然在实现中，BufferedIOBase 类是用来包装一个 IOBase 对象的，但它们的接口其实是一个流（所有的东西 都继承自 IOBase）。所以，普通的 IOBase 对象在超出作用范围时会自动关闭自己。BufferedIOBase 只是把 close() 的调用转发给底层的流。

你不应该把 BufferedReader 看作是一个流的包装器（虽然它确实是这样实现的），而应该把它看作是对一个已有流的类型转换。 这两个流的状态是完全绑定在一起的。不过，你可以通过 detach() 来解除绑定，但这样会让 BufferedIOBase 对象变得无用。

另外，当模式是 rb 时，io.open 已经返回了一个 BufferedReader，所以你实际上是在进行双重缓冲。你应该使用 io.FileIO 来代替。

你有几个选择：

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

创建原始文件描述符，并根据需要从它们创建流。这需要小心，确保多个流不会同时使用同一个文件描述符。例如：

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

在你的 BufferedIOBase 对象关闭其流之前，先 detach() 底层流。（记得要重置流！）

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() 的转发：

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

回答于 2025-04-17 由 Python大师

分享举报

防止BufferedReader关闭文件列表中的文件

1 个回答

撰写回答