Python 增量解码器怎样在最后一块检查残留字节

Python 的增量解码器不会把每个字节块都当成一条完整消息。正确配方是：处理中调用 decode(chunk, final=False)，让解码器保留暂时不够组成字符的尾部；确认输入流结束后，再调用一次 decode(b"", final=True)。这次调用会刷新内部缓冲，如果残留字节不能组成合法字符，就按 errors 策略处理，默认的 strict 通常会抛出 UnicodeDecodeError。

官方文档：https://docs.python.org/3/library/codecs.html

先把普通分块交给同一个解码器

增量解码器的价值在于“状态连续”：同一个实例接收多次 decode()，内部会记住上一个分块末尾还没有完成的字节。UTF-8 一个汉字可能占多个字节，如果网络或文件读取恰好在字符中间切开，前一块不应立即被当成坏数据。

先通过 codecs.getincrementaldecoder() 取得对应编码的解码器类，再创建一个实例。每个独立输入流应拥有自己的实例，不要把两个文件或两个客户端的字节交错喂给同一个状态机。

import codecs

# 取得 UTF-8 的增量解码器类；同一实例只服务一个输入流。
decoder_class = codecs.getincrementaldecoder("utf-8")
decoder = decoder_class(errors="strict")

# 普通分块先保持 final=False，让解码器暂存可能不完整的尾部。
text_parts = [
    decoder.decode(b"日", final=False),
    decoder.decode(b"本語", final=False),
]

# 这里只展示每次调用返回的已完成文本，不代表实际运行输出。
print("".join(text_parts))

前面的调用只负责消费“目前已经能确定”的字符。返回空字符串不等于输入丢失，也可能意味着当前分块只是某个多字节字符的一部分，剩余字节正在解码器的内部状态里等待下一块。

在最后一块用 final=True 刷新内部缓冲

final=True 是流尾信号，不是“这次传入的参数比较重要”的标记。它告诉解码器：后面不会再有字节了，因此必须把输入完整解码并刷新所有缓冲。如果最后留下一个不完整的 UTF-8 序列，解码器就不能继续等待下一块。

最小的流尾处理可以固定成下面这一段。传入空字节串是为了只触发收尾动作；如果最后一块还没有先交给解码器，也可以把它直接作为这次调用的第一个参数。

import codecs

decoder = codecs.getincrementaldecoder("utf-8")(errors="strict")
parts = []

# 前面的块可以在字符边界之外结束，解码器会保留状态。
parts.append(decoder.decode(b"hello ", final=False))
parts.append(decoder.decode("世界".encode("utf-8"), final=False))

try:
    # 输入流到此结束；final=True 会强制检查仍留在缓冲中的字节。
    parts.append(decoder.decode(b"", final=True))
except UnicodeDecodeError as error:
    # 收尾失败要单独记录，不能伪装成普通的空字符串。
    print(f"流尾解码失败：{error}")
else:
    print("".join(parts))

如果流尾完整，最后一次调用通常只返回剩余的已完成文本或空字符串；如果尾部不完整，默认 strict 会把问题暴露出来。文章中的代码只说明调用关系，具体输出取决于输入字节。

把不完整字节、错误策略和空输出分开

最容易混淆的是三种情况：普通分块暂时没有可输出字符、最后一块确实没有剩余内容，以及最后一块存在不完整字节。它们都可能在某次调用中看到空字符串，但处理含义不同。

errors="replace" 可以用替换字符继续产出文本，errors="ignore" 会丢弃无法解码的部分。日志、协议解析和数据落库通常更适合保留默认的 strict，先知道输入被截断，再决定是否允许降级；展示型文本才可能选择替换策略。

把分块来源接成可复用的迭代器配方

文件读取、网络接收和消息队列虽然来源不同，但可以统一成“不断产生 bytes，结束后收尾”的接口。关键是把 final=True 放在迭代结束之后，并保证异常不被生成器的空值掩盖。

import codecs
from collections.abc import Iterable, Iterator

def decode_chunks(chunks: Iterable[bytes], encoding: str = "utf-8") -> Iterator[str]:
    # 每个输入迭代器创建独立解码器，避免不同流共享残留状态。
    decoder = codecs.getincrementaldecoder(encoding)(errors="strict")

    for chunk in chunks:
        if not isinstance(chunk, bytes):
            # 先拒绝错误输入类型，避免错误落在编码器内部才暴露。
            raise TypeError("chunks 必须产生 bytes")
        # 普通块不宣告结束，可能的半个字符由解码器暂存。
        text = decoder.decode(chunk, final=False)
        if text:
            yield text

    # 迭代器耗尽才代表流尾；这里负责检查最后的残留字节。
    tail = decoder.decode(b"", final=True)
    if tail:
        yield tail

# 下面只给出调用形状；输入可以来自文件、网络或消息队列。
for text in decode_chunks([b"A", "中".encode("utf-8")]):
    print(text)

这个配方把“分块处理”和“结束确认”分开：循环内只处理可继续到来的数据，循环结束后才做一次最终检查。若上游能提前报告传输错误，应先保留原始错误信息，再决定是否仍然调用收尾。

明确 final=True 不能用于每个分块

如果每次收到网络分片都传入 final=True，解码器会把每一块都当成完整输入；字符恰好跨块时，前一块就可能提前触发错误，后续块也无法补齐它。相反，如果永远使用 final=False，截断的最后字节可能一直留在内部缓冲，直到调用方主动收尾才会暴露。

因此，最小判断规则只有两条：还会有下一块时使用 final=False，确认不会再有输入时使用 final=True。需要查看解码器当前状态时可使用 getstate()，但不要把状态元组当成业务协议直接持久化；它是实现接口的一部分，适合调试和受控恢复。

为什么最后一次要传空字节串？因为前面的分块已经交给解码器，空字节串只触发流尾刷新；如果最后一块尚未处理，也可以把最后一块直接传给 decode(..., final=True)。final=True 一定会返回非空文本吗？不一定，完整状态下它可能只返回空字符串。能否用字符串直接喂给增量解码器？文本编码的增量解码器通常接收字节输入，分块接口应保持 bytes 类型。