跳到主要内容
版本：1.2.2

LineBuffer

类上下文源码

tbuf = LineBuffer(maxlen: int = 20)

with tbuf:
    ...

基础的行缓存句柄。

该缓存为文字流提供了一个轮换存储空间。文字的存储单位不是字符，而是行。存储空间的最大行数是有限的。

别名​

该类可以按以下方式之一获取

import syncstream


syncstream.LineBuffer
syncstream.mproc.LineBuffer

参数​

参数类型必选
说明
maxlenint所存储的最大行数。该值和deque.maxlen的含义相同。

方法​

close​

tbuf.close()

关闭 IO。该方法只会生效一次。第二次调用则无任何效果。

clear​

tbuf.clear()

清空整个缓存。

该方法会清空该缓存内的整个存储空间，以及当前正在写的最后一行的流。然而，该方法不会清空该缓存的任何镜像或拷贝。该方法是线程安全的，且应当永远可以成功执行。

new_line​

tbuf.new_line()

清空当前的临时缓存，并人为地触发一个“新行”信号。若当前的临时缓存包含数据，则会先将数据移动到存储区，然后再创建新行。若当前的临时缓存的已经是新行，则不做任何处理。

该方法等价于

if tbuf.last_line.tell() > 0:
    write('\n')

flush​

tbuf.flush()

刷新当前正在写入行的数据流（临时缓存）。

parse_lines​

tbuf.parse_lines(lines)

处理行。

每次write()方法将要向缓存写入新行时，调用此方法。默认的行为是向存储空间添加一条新的信息。

用户自行定制形如“正则表达式搜索”的处理方法、并以此继承、重载该方法。该方法的默认行为透过以下私有方法实现：

tbuf.storage.extend(lines: Sequence[str])

输入​

参数类型必选
说明
linesSequence[str]要写入存储空间的字符串序列。用户可以捕获任一行，并从中提取某种特定的信息。

read​

lines: Sequence[str] = tbuf.read(size: int | None = None)

从缓存中获取所存储的各项记录。read()方法是线程安全的，并且不会干扰write()方法的指针。

若当前正在写入的行不为空，调用该方法时，则会将其视为最后一条记录。

输入​

参数类型必选
说明
sizeint | None若设为None，则会返回整个存储区。若设为一个整数，则会返回最后size条信息。

输出​

参数类型
说明
linesSequence[str]所返回的信息。这些信息已经按断行符分拆过。

write​

n_bytes: int = tbuf.write(data: str)

写缓存的方法。源数据data的形式与文字IO的数据形式相同。每当data包含断行符时，会向存储区置入一条新的记录。write()方法是进程安全的。

输入​

参数类型必选
说明
datastr要写入的信息。会自动检测其中包含的断行符。

输出​

参数类型
说明
n_bytesint所要写入缓存的byte数目。该值直接从输入data计算。

fileno​

tbuf.fileno() -> Never

返回文件 ID。

由于该缓存不使用文件 ID，故而该方法会抛出OSError。

isatty​

is_atty: False = tbuf.fileno()

流是否连接到了终端/TTY 上。返回False。

输出​

参数类型
说明
is_attybool恒为False.

readable​

is_readable: bool = tbuf.readable()

流是否可读。只要流还未关闭，其就是可读的。

若流不再可读，调用read()会抛出OSError。

输出​

参数类型
说明
is_readablebool若缓存未关闭，返回True。否则，返回False。

writable​

is_writable: bool = tbuf.writable()

流是否可写。只要流还未关闭，其就是可读的。

若流不再可写，调用read()会抛出OSError。

输出​

参数类型
说明
is_writablebool若缓存未关闭，返回True。否则，返回False。

seekable​

is_seekable: False = tbuf.seekable()

流是否支持随机访问。该缓存不支持。

输出​

参数类型
说明
is_seekablebool恒为False。

seek​

tbuf.seek() -> Never

由于缓存不支持随机访问，调用该方法会抛出OSError。

属性​

maxlen​

maxlen: int | None = tbuf.maxlen

缓存中的最大长度（即行数）。

若该值为None，则表明缓存没有最大长度限制。

closed​

is_closed: bool = tbuf.closed

检查缓存是否已经被关闭。

运算符​

__len__​

n_buffer_items: int = len(tbuf)

缓存中的行/条目数。

范例​

在线程之间同步信息​

sync_among_threads.py
from syncstream.mproc import LineBuffer

tbuf = LineBuffer(5)

with tbuf:
    for i in range(10):
        print(i)

print(tbuf.read())