Asyncio StreamReader 如何正确增量解码 UTF-8

2026-11-04数据小陈

不要对每次 await stream.read(n) 得到的 bytes 直接调用 decode('utf-8')。可以用 codecs.getincrementaldecoder('utf-8') 创建增量解码器，逐块传入 chunk，最后在 EOF 时调用 decode(b'', final=True) 收尾。这样，即使一个 UTF-8 字符被拆到多个 chunk 里，也能正确拼回。

问题在于，StreamReader.read(n) 返回的是 bytes 块，n 控制的是字节数，并不对应字符边界。UTF-8 的 ä、ß 这类字符会占多个字节，分块读取时很容易把 \xc3 和后续字节拆开。普通的 bytes.decode('utf-8') 默认把当前输入当作完整数据（final=True），一旦遇到不完整序列，就会抛出 UnicodeDecodeError: unexpected end of data。

按块做增量解码即可：

import asyncio
import codecs

async def get_text(stream):
    decoder = codecs.getincrementaldecoder('utf-8')(errors='strict')
    pieces = []

    while True:
        chunk = await stream.read(4)
        if not chunk:
            break
        text = decoder.decode(chunk, final=False)
        if text:
            pieces.append(text)

    tail = decoder.decode(b'', final=True)
    if tail:
        pieces.append(tail)

    return ''.join(pieces)

如果想把解码逻辑封装成类似 StreamReader 的异步包装类，可以这样写：

import codecs

class DecodingStreamReader:
    def __init__(self, stream, encoding='utf-8', errors='strict'):
        self.stream = stream
        self.decoder = codecs.getincrementaldecoder(encoding)(errors=errors)
        self._finished = False

    async def read(self, n=-1):
        if self._finished:
            return ''

        data = await self.stream.read(n)
        if not data:
            self._finished = True
            return self.decoder.decode(b'', final=True)

        text = self.decoder.decode(data, final=False)
        if self.stream.at_eof():
            self._finished = True
            text += self.decoder.decode(b'', final=True)
        return text

    def at_eof(self):
        return self.stream.at_eof()

注意：

final=True 只应在数据流结束时调用一次。提前调用会把尚未完整的多字节序列当成非法数据。
不要用 errors='ignore' 来处理跨 chunk 的问题。它会静默丢掉字节，可能破坏文本内容。
codecs.getreader('utf-8') 和 io.TextIOWrapper 属于同步流接口，不能直接套在需要 await 的 StreamReader 上。前者会报 TypeError: can't concat bytes to generator，后者常见报错是 AttributeError: 'FlowControlStreamReader' object has no attribute 'readable'。
如果响应还经过 gzip/br 等压缩，应先解压成 bytes，再做 UTF-8 增量解码。
如果按行处理，应在增量解码后的文本上做缓冲和分行，不要先按字节调用 splitlines()。