写入失败时数据去哪了一行 error 背后可能是几百条设备数据消失。本文从可运行源码出发拆解数据保底三道防线指数退避重试兜住瞬时错误DiskSpool 原子写落盘兜住持久故障重启后按序回放把数据捞回来。写文件为何必须 tmp→fsync→rename一文讲透。凌晨两点告警把你从梦里拽醒。日志里躺着一行 errorWrite failed, retries exhausted。你点了确认准备回去继续睡。但就在这一行日志的背后这一批发往数据库的传感器数据已经彻底消失了。写入失败的真正问题从来不是“要不要重试”而是失败的那一刻数据去哪了这是 TDengine 工业物联网实战系列的第 5 篇。第 4 篇搭好了异步写入管线这一篇补齐最后一块拼图——数据保底三板斧重试、落盘、回放。没有兜底的写链路等于裸奔先看一个朴素的写入实现try: client.insert(batch) except Exception: log.error(write failed) # 然后呢没有然后了抛异常、打日志看起来没什么问题。可异常被吞掉的那一刻这批数据就永远离开了内存。你以为只是少写几条实际上整批数据都没了。所以”不丢数据”不是靠承诺而是靠制度每一种失败都有一条对应的防线三道防线环环相扣重试吃掉瞬时错误数据库抖动几秒就能自愈的场景根本不需要惊动别人落盘持久故障直接把内存中的数据写到磁盘先活下来再说回放进程重启后把磁盘上的数据按序补回数据库一个很常见的场景数据库连接池短暂耗尽几秒钟就恢复。如果没有重试这一条数据就直接蒸发了。重试的意义就在这里——大部分“看起来严重”的故障其实是瞬时的。第一板斧重试要退避也要随机重试的逻辑全部封装在common/retry.py里核心函数只有 22 行asyncdefretry_async(operation:Callable[[],Awaitable[T]],*,attempts:int,base_delay:float,retryable:tuple[type[Exception],...](OSError,TimeoutError),)-T:Retry an async operation with capped exponential backoff and jitter.last_error:Exception|NoneNoneforattemptinrange(attempts1):try:returnawaitoperation()exceptretryableaserror:last_errorerrorifattemptattempts:raisedelaymin(base_delay*(2**attempt),30.0)delay*random.uniform(0.8,1.2)awaitasyncio.sleep(delay)assertlast_errorisnotNoneraiselast_error逐行拆解四个设计决策值得细品第一attempts 1次机会。默认 5 次重试实际上是 1 次原始尝试再加 5 次重试一共 6 次出手机会。注意循环是range(attempts 1)最后一次异常后直接raise不会白等一个用不上的退避。第二指数退避但封顶 30 秒。base_delay * 2**attempt逐次翻倍默认配置下等待序列是 0.25s、0.5s、1s、2s、4s最后一次失败直接raise不白等。如果退避不设上限第 10 次重试的理论等待会到 128 秒——一个 30 秒的上限防止退避时间在长故障里失控。第三±20% 的抖动。这是整个函数最容易被忽略、也最救命的细节下一节专门讲。第四retryable白名单机制。默认只重试OSError和TimeoutError顺带覆盖ConnectionError它是OSError的子类。业务错误——比如数据格式非法——重试一万次也还是失败不如尽快失败抛出去别堵在队列里拖垮整个管线。还有一个异步的隐形福利await asyncio.sleep(delay)不阻塞事件循环。重试等待期间管线还能继续处理其他数据不会被一个卡住的请求堵死整条链路。没有抖动的退避是另一种故障讲一个分布式系统里的经典事故数据库宕机所有客户端几乎同时检测到连接失败。数据库恢复的那一瞬间几百个客户端的退避计时器同时到点在同一秒发起重试——数据库被再次打挂。这不是段子是真实的故障模式叫重试风暴retry storm。抖动的本质是让“所有人在同一时刻重试”变成“大家在 0.8x 到 1.2x 的区间内随机散开”。配置里base_delay0.25s第 2 次重试的理论等待是 0.5 秒加了抖动后实际落在 0.4 到 0.6 秒之间。看起来只是一个小小的随机扰动却能把请求在时间轴上摊开避免恢复瞬间的流量尖峰。关键就是这一行delay*random.uniform(0.8,1.2)一个乘法救活一个系统。别小看这行代码——它把所有人同时重试变成随机散开是重试代码里性价比最高的一行。第二板斧落盘写文件也要原子重试解决瞬时错误但数据库要是挂了 10 分钟呢6 次重试全部失败数据不能丢内存也兜不住——进程一重启就全没了。这时候必须落盘。DiskSpool 的原子写实现在writer/spool.py里asyncdefstore(self,records:Sequence[Record])-Path:ifnotrecords:raiseValueError(cannot spool an empty batch)awaitself.initialize()asyncwithself._lock:current_sizeawaitself.size_bytes()payload\n.join(self._serialize(record)forrecordinrecords)\nencodedpayload.encode()ifcurrent_sizelen(encoded)self.max_bytes:raiseOSError(disk spool capacity exceeded)stemf{os.getpid()}-{uuid4().hex}temporaryself.directory/f{stem}.tmpreadyself.directory/f{stem}.readydefwrite_file()-None:withtemporary.open(xb)asstream:stream.write(encoded)stream.flush()os.fsync(stream.fileno())temporary.replace(ready)awaitasyncio.to_thread(write_file)returnready注意看写文件的核心三段式1. 写.tmp临时文件。xb排他创建文件已存在直接报错防止同 pid 并发重复写同一个文件。文件名用{pid}-{uuid4}避免多进程/多线程互相覆盖。2.flush()fsync()。这是最容易偷懒的一步也是必须的一步。flush()只是把 Python 缓冲区的数据交给操作系统此时数据还在页缓存里断电照样丢。os.fsync()强制把页缓存刷到物理磁盘落盘才真正发生。两条缺一不可这正是写文件要原子的地基。3.temporary.replace(ready)。rename 在 POSIX 上是原子操作要么旧文件要么新文件不会出现半截状态。配合文件名后缀——写的时候叫.tmp写完 rename 成.ready——消费端永远只扫*.ready一个写了一半的临时文件永远不会被当成有效数据读走。这就是文件协议读侧只看.ready写侧先落.tmp。还有两个细节值得一提。容量上限是硬约束不是摆设。store()先算size_bytes()所有.ready文件之和超了spool_max_bytes默认 1GB直接抛OSError(disk spool capacity exceeded)。注意它没有静默丢弃也没有悄悄覆盖——宁可让调用方收到异常、触发告警也不让磁盘悄悄写满。写文件也串行化。多 worker 同时落盘时asyncio.Lock保证同一时刻只有一个store在跑所有阻塞 IOmkdir、write、fsync、stat都丢进asyncio.to_thread不卡事件循环。第 4 篇里连接复用 锁串行的模式在磁盘上又出现了一次。落盘触发点重试耗尽之后第 4 篇讲_flush时留了个钩子重试耗尽怎么办答案就在_flush的异常分支里asyncdef_flush(self,batch:Sequence[Record],index:int)-None:ifnotbatch:returntry:awaitself._write_with_retry(batch)exceptExceptionaserror:logger.exception(batch exhausted retries,extra{batch_size:len(batch),worker:index},)RECORDS_FAILED.labels(self.writer.transport,type(error).__name__).inc(len(batch))ifself.settings.spool_enabled:awaitself.spool.store(batch)finally:for_inbatch:self.queue.task_done()QUEUE_DEPTH.set(self.queue.qsize())整批落盘6 次尝试全部失败后这一批最多 1000 条整体spool.store不是一条一条拆开写——保持批次完整性回放时一次写回指标先行RECORDS_FAILED按传输方式和异常类型打标签Prometheus 里能看到失败在往哪个方向积累不吞异常spool 满了抛OSError会继续向上传播进程层拿到告警——数据宁可报错也不许无声消失。第三板斧回放重启后把数据捞回来落盘只是让数据活下来真正回到数据库要靠回放。回放发生在启动时、开始接收新数据之前# collector/service.pyasyncdefstart(self)-None:awaitself.pipeline.start()replayedawaitself.pipeline.replay_spool()logger.info(collector started,extra{replayed:replayed})replay_spool的实现asyncdefreplay_spool(self)-int:ifnotself.settings.spool_enabled:return0replayed0asyncforpathinself.spool.files():recordsawaitself.spool.load(path)try:awaitself._write_with_retry(records)exceptException:logger.exception(spool replay failed,extra{path:str(path)})breakawaitself.spool.acknowledge(path)replayedlen(records)returnreplayed三个决策值得单独说按序回放。files()对*.ready做排序再逐个处理——文件按写入先后顺序回放时间戳乱序的坑从源头避免。时序数据最怕乱序写入回放自然也要按序。回放也走重试。_write_with_retry完全复用数据库刚重启回放的第一批可能还会失败指数退避再兜一轮。重试不是写路径的专利回放路径同样有资格。失败就停但不删。某个文件回放失败比如数据库又挂了立刻break退出循环——不删除文件、也不跳过剩下的等下次启动再试。宁可晚到不可丢这个原则从写路径贯彻到回放路径。端到端串一遍一次故障的完整旅程把三道防线串起来看一次真实的故障闭环t0 数据库连接池耗尽 t00s 第一批写入失败 → 指数退避重试 t04s 6 次尝试全部失败 → RECORDS_FAILED 整批落盘 spool t010s 进程重启发布/崩溃 t010s start(): 先 replay_spool按序读 .ready 文件 t011s 回放走 retry_async写入成功 → acknowledge 删除文件 t012s 开始接收新数据一切照常每一步的失败都有对应的防线每一道防线都独立可验重试次数、落盘文件数、回放成功数都是日志和指标里的数字。不丢数据不是一句口号而是一条条可观测的链路。写在最后回头看这三板斧的定位重试 → 兜住瞬时错误几秒钟的自愈场景 落盘 → 兜住持久故障内存里的数据先上磁盘 回放 → 兜住进程重启磁盘上的数据按序补回它们分别回答三个问题失败了吗再试一次。还失败写盘。重启了捞回来。第 4 篇的管线负责写得快这一篇的保底负责不丢数据。下一篇换个视角离开写入侧实测三种连接方式WebSocket / REST / Native的吞吐与延迟差异——看看同样的数据走不同的门进 TDengine差别到底有多大。觉得有用点个关注持续获取优质内容。
写入失败时数据去哪了一行 error 背后可能是几百条设备数据消失。本文从可运行源码出发拆解数据保底三道防线指数退避重试兜住瞬时错误DiskSpool 原子写落盘兜住持久故障重启后按序回放把数据捞回来。写文件为何必须 tmp→fsync→rename一文讲透。凌晨两点告警把你从梦里拽醒。日志里躺着一行 errorWrite failed, retries exhausted。你点了确认准备回去继续睡。但就在这一行日志的背后这一批发往数据库的传感器数据已经彻底消失了。写入失败的真正问题从来不是“要不要重试”而是失败的那一刻数据去哪了这是 TDengine 工业物联网实战系列的第 5 篇。第 4 篇搭好了异步写入管线这一篇补齐最后一块拼图——数据保底三板斧重试、落盘、回放。没有兜底的写链路等于裸奔先看一个朴素的写入实现try: client.insert(batch) except Exception: log.error(write failed) # 然后呢没有然后了抛异常、打日志看起来没什么问题。可异常被吞掉的那一刻这批数据就永远离开了内存。你以为只是少写几条实际上整批数据都没了。所以”不丢数据”不是靠承诺而是靠制度每一种失败都有一条对应的防线三道防线环环相扣重试吃掉瞬时错误数据库抖动几秒就能自愈的场景根本不需要惊动别人落盘持久故障直接把内存中的数据写到磁盘先活下来再说回放进程重启后把磁盘上的数据按序补回数据库一个很常见的场景数据库连接池短暂耗尽几秒钟就恢复。如果没有重试这一条数据就直接蒸发了。重试的意义就在这里——大部分“看起来严重”的故障其实是瞬时的。第一板斧重试要退避也要随机重试的逻辑全部封装在common/retry.py里核心函数只有 22 行asyncdefretry_async(operation:Callable[[],Awaitable[T]],*,attempts:int,base_delay:float,retryable:tuple[type[Exception],...](OSError,TimeoutError),)-T:Retry an async operation with capped exponential backoff and jitter.last_error:Exception|NoneNoneforattemptinrange(attempts1):try:returnawaitoperation()exceptretryableaserror:last_errorerrorifattemptattempts:raisedelaymin(base_delay*(2**attempt),30.0)delay*random.uniform(0.8,1.2)awaitasyncio.sleep(delay)assertlast_errorisnotNoneraiselast_error逐行拆解四个设计决策值得细品第一attempts 1次机会。默认 5 次重试实际上是 1 次原始尝试再加 5 次重试一共 6 次出手机会。注意循环是range(attempts 1)最后一次异常后直接raise不会白等一个用不上的退避。第二指数退避但封顶 30 秒。base_delay * 2**attempt逐次翻倍默认配置下等待序列是 0.25s、0.5s、1s、2s、4s最后一次失败直接raise不白等。如果退避不设上限第 10 次重试的理论等待会到 128 秒——一个 30 秒的上限防止退避时间在长故障里失控。第三±20% 的抖动。这是整个函数最容易被忽略、也最救命的细节下一节专门讲。第四retryable白名单机制。默认只重试OSError和TimeoutError顺带覆盖ConnectionError它是OSError的子类。业务错误——比如数据格式非法——重试一万次也还是失败不如尽快失败抛出去别堵在队列里拖垮整个管线。还有一个异步的隐形福利await asyncio.sleep(delay)不阻塞事件循环。重试等待期间管线还能继续处理其他数据不会被一个卡住的请求堵死整条链路。没有抖动的退避是另一种故障讲一个分布式系统里的经典事故数据库宕机所有客户端几乎同时检测到连接失败。数据库恢复的那一瞬间几百个客户端的退避计时器同时到点在同一秒发起重试——数据库被再次打挂。这不是段子是真实的故障模式叫重试风暴retry storm。抖动的本质是让“所有人在同一时刻重试”变成“大家在 0.8x 到 1.2x 的区间内随机散开”。配置里base_delay0.25s第 2 次重试的理论等待是 0.5 秒加了抖动后实际落在 0.4 到 0.6 秒之间。看起来只是一个小小的随机扰动却能把请求在时间轴上摊开避免恢复瞬间的流量尖峰。关键就是这一行delay*random.uniform(0.8,1.2)一个乘法救活一个系统。别小看这行代码——它把所有人同时重试变成随机散开是重试代码里性价比最高的一行。第二板斧落盘写文件也要原子重试解决瞬时错误但数据库要是挂了 10 分钟呢6 次重试全部失败数据不能丢内存也兜不住——进程一重启就全没了。这时候必须落盘。DiskSpool 的原子写实现在writer/spool.py里asyncdefstore(self,records:Sequence[Record])-Path:ifnotrecords:raiseValueError(cannot spool an empty batch)awaitself.initialize()asyncwithself._lock:current_sizeawaitself.size_bytes()payload\n.join(self._serialize(record)forrecordinrecords)\nencodedpayload.encode()ifcurrent_sizelen(encoded)self.max_bytes:raiseOSError(disk spool capacity exceeded)stemf{os.getpid()}-{uuid4().hex}temporaryself.directory/f{stem}.tmpreadyself.directory/f{stem}.readydefwrite_file()-None:withtemporary.open(xb)asstream:stream.write(encoded)stream.flush()os.fsync(stream.fileno())temporary.replace(ready)awaitasyncio.to_thread(write_file)returnready注意看写文件的核心三段式1. 写.tmp临时文件。xb排他创建文件已存在直接报错防止同 pid 并发重复写同一个文件。文件名用{pid}-{uuid4}避免多进程/多线程互相覆盖。2.flush()fsync()。这是最容易偷懒的一步也是必须的一步。flush()只是把 Python 缓冲区的数据交给操作系统此时数据还在页缓存里断电照样丢。os.fsync()强制把页缓存刷到物理磁盘落盘才真正发生。两条缺一不可这正是写文件要原子的地基。3.temporary.replace(ready)。rename 在 POSIX 上是原子操作要么旧文件要么新文件不会出现半截状态。配合文件名后缀——写的时候叫.tmp写完 rename 成.ready——消费端永远只扫*.ready一个写了一半的临时文件永远不会被当成有效数据读走。这就是文件协议读侧只看.ready写侧先落.tmp。还有两个细节值得一提。容量上限是硬约束不是摆设。store()先算size_bytes()所有.ready文件之和超了spool_max_bytes默认 1GB直接抛OSError(disk spool capacity exceeded)。注意它没有静默丢弃也没有悄悄覆盖——宁可让调用方收到异常、触发告警也不让磁盘悄悄写满。写文件也串行化。多 worker 同时落盘时asyncio.Lock保证同一时刻只有一个store在跑所有阻塞 IOmkdir、write、fsync、stat都丢进asyncio.to_thread不卡事件循环。第 4 篇里连接复用 锁串行的模式在磁盘上又出现了一次。落盘触发点重试耗尽之后第 4 篇讲_flush时留了个钩子重试耗尽怎么办答案就在_flush的异常分支里asyncdef_flush(self,batch:Sequence[Record],index:int)-None:ifnotbatch:returntry:awaitself._write_with_retry(batch)exceptExceptionaserror:logger.exception(batch exhausted retries,extra{batch_size:len(batch),worker:index},)RECORDS_FAILED.labels(self.writer.transport,type(error).__name__).inc(len(batch))ifself.settings.spool_enabled:awaitself.spool.store(batch)finally:for_inbatch:self.queue.task_done()QUEUE_DEPTH.set(self.queue.qsize())整批落盘6 次尝试全部失败后这一批最多 1000 条整体spool.store不是一条一条拆开写——保持批次完整性回放时一次写回指标先行RECORDS_FAILED按传输方式和异常类型打标签Prometheus 里能看到失败在往哪个方向积累不吞异常spool 满了抛OSError会继续向上传播进程层拿到告警——数据宁可报错也不许无声消失。第三板斧回放重启后把数据捞回来落盘只是让数据活下来真正回到数据库要靠回放。回放发生在启动时、开始接收新数据之前# collector/service.pyasyncdefstart(self)-None:awaitself.pipeline.start()replayedawaitself.pipeline.replay_spool()logger.info(collector started,extra{replayed:replayed})replay_spool的实现asyncdefreplay_spool(self)-int:ifnotself.settings.spool_enabled:return0replayed0asyncforpathinself.spool.files():recordsawaitself.spool.load(path)try:awaitself._write_with_retry(records)exceptException:logger.exception(spool replay failed,extra{path:str(path)})breakawaitself.spool.acknowledge(path)replayedlen(records)returnreplayed三个决策值得单独说按序回放。files()对*.ready做排序再逐个处理——文件按写入先后顺序回放时间戳乱序的坑从源头避免。时序数据最怕乱序写入回放自然也要按序。回放也走重试。_write_with_retry完全复用数据库刚重启回放的第一批可能还会失败指数退避再兜一轮。重试不是写路径的专利回放路径同样有资格。失败就停但不删。某个文件回放失败比如数据库又挂了立刻break退出循环——不删除文件、也不跳过剩下的等下次启动再试。宁可晚到不可丢这个原则从写路径贯彻到回放路径。端到端串一遍一次故障的完整旅程把三道防线串起来看一次真实的故障闭环t0 数据库连接池耗尽 t00s 第一批写入失败 → 指数退避重试 t04s 6 次尝试全部失败 → RECORDS_FAILED 整批落盘 spool t010s 进程重启发布/崩溃 t010s start(): 先 replay_spool按序读 .ready 文件 t011s 回放走 retry_async写入成功 → acknowledge 删除文件 t012s 开始接收新数据一切照常每一步的失败都有对应的防线每一道防线都独立可验重试次数、落盘文件数、回放成功数都是日志和指标里的数字。不丢数据不是一句口号而是一条条可观测的链路。写在最后回头看这三板斧的定位重试 → 兜住瞬时错误几秒钟的自愈场景 落盘 → 兜住持久故障内存里的数据先上磁盘 回放 → 兜住进程重启磁盘上的数据按序补回它们分别回答三个问题失败了吗再试一次。还失败写盘。重启了捞回来。第 4 篇的管线负责写得快这一篇的保底负责不丢数据。下一篇换个视角离开写入侧实测三种连接方式WebSocket / REST / Native的吞吐与延迟差异——看看同样的数据走不同的门进 TDengine差别到底有多大。觉得有用点个关注持续获取优质内容。
