跨服务通信兜底机制-Java 回传失败无持久重试队列，报告可能静默丢失。

情况AI模拟面试的RAG部分中将生成的报告回传到java端时如若产生网络错误产生的报告直接丢失或者重复发送。为此利用如下机制Tenacity(Python 第三方重试框架)角色系统的“减震器”。技术细节它是一个通过 Python 装饰器Decorator实现的高级重试库。你使用了它的wait_exponential指数退避算法这比简单的time.sleep()强大得多它是工业界解决“拥堵重试”的标准算法。HTTP Headers(网络协议层自定义标识)角色系统的“防伪标签”。技术细节你利用了 HTTP 协议的扩展性在 Request Header 中自定义了X-Idempotency-Key字段来携带session_id。这是一种非侵入式的数据传递不会破坏原有 JSON Payload 的结构。Redis List(内存数据库数据结构)角色系统的“冷冻保险箱”。技术细节你利用了 Redis 的双端队列特性。通过lpush左侧推入命令将 Python 字典序列化为 JSON 字符串后安全落盘。Redis 的极速响应保证了兜底写入时不会拖垮主线程。解决问题的方法论️ 第一步用“容错重试”解决偶发波动问题场景发送请求的那一秒公司的路由器抖动了或者 Java 后端正好在处理垃圾回收GC卡顿了 1 秒。如果直接报错这份报告就凭空消失了。解决逻辑你封装了_do_push_with_retry允许代码在遇到 HTTP 异常时“再试几次”。而且为了防止把 Java 服务器打垮你要求每次重试的等待时间按指数级拉长2秒、4秒、8秒这种“克制的重试”能消化掉 90% 的偶发网络故障。️ 第二步用“幂等设计”解决重复执行问题场景重试机制是一把双刃剑。如果 Java 其实已经成功收到了报告存进数据库了仅仅是在给 Python 回复200 OK时网络断了。此时 Python 误以为没发成功触发重试再次发送。这就导致 Java 数据库里存了两条一样的面试成绩解决逻辑你引入了**幂等性Idempotency**思维。在发送时贴上唯一的单号session_id。这个架构设计把防重的压力合法地转移给了 Java 端——Java 端在存库前必须先校验这个单号如果处理过就直接无视并返回成功。彻底绝杀了“数据重复”的 P0 级 Bug。️ 第三步用“死信队列 (DLQ)”解决彻底宕机问题场景如果 Java 后台正在进行长达半小时的版本升级重启无论你重试多少次都会失败。程序最终抛出异常此时如果依然打印一句错误就结束数据还是会丢失。解决逻辑你秉持了**“最终一致性Eventual Consistency”**的底层逻辑。当重试弹药全部打光后你在except块中捕获了最终异常将原本要丢弃的数据打包成“死信Dead Letter”静悄悄地塞进 Redis 的failed_reports_queue里。数据没有消失只是转移了阵地等待未来系统恢复后进行手动或定时的补偿发送。# 1. 失败落库的兜底逻辑 async def save_failed_report_to_redis(user_id: str, session_id: str, payload: dict): 当所有重试都失败后把报告塞进 Redis 死信队列等待后续手动/定时补偿 try: dead_letter { session_id: session_id, user_id: user_id, payload: payload, timestamp: time.time() } # lpush: 从左边推进一个名为 failed_reports_queue 的列表 await redis_client.lpush(failed_reports_queue, json.dumps(dead_letter, ensure_asciiFalse)) print(f [极致兜底] 会话 {session_id} 的报告已安全存入 Redis 死信队列数据未丢失。) except Exception as e: print(f [灾难] 死信队列写入失败报告彻底丢失: {e}) # 2. 带自动重试机制的发送逻辑 # retry 魔法如果遇到网络报错自动重试。 # wait_exponential(multiplier1, min2, max10)指数退避等待第一次等2秒第二次等4秒第三次等8秒 # stop_after_attempt(3)最多疯狂重试 3 次不行就放弃抛出异常 retry(waitwait_exponential(multiplier1, min2, max10), stopstop_after_attempt(3)) async def _do_push_with_retry(payload: dict): 执行实际的 HTTP 请求失败会自动重试 async with httpx.AsyncClient() as client: response await client.post( settings.JAVA_REPORT_URL, # ⚠️ 统一配置项名称确保 .env 里叫这个名字 jsonpayload, headers{ X-Idempotency-Key: payload[sessionId] # 极其关键的幂等键防止 Java 端重复存库 }, timeout15.0 ) # 如果 Java 端返回 500/502/503 等非 2xx 的错误码主动抛出异常强制触发 Tenacity 进行重试 response.raise_for_status() return response # 3. 对外暴露的业务调度函数 async def push_report_to_java(user_id: str, session_id: str, report_data: dict): 将生成的报告推送给 Java 业务端已接入重试与死信队列保障 # 1. 组装发给 Java 的终极 Payload (这部分你写得很完美) payload { userId: user_id, sessionId: session_id, score: report_data.get(comprehensiveScore, 0), reportDetail: json.dumps(report_data, ensure_asciiFalse), contentShortcomings: report_data.get(contentAnalysis, {}).get(shortcomings, ), expressionShortcomings: report_data.get(expressionAnalysis, {}).get(shortcomings, ) } try: print(f 正在将 {session_id} 的报告发送至 Java 业务中心...) # 核心修改点丢弃旧的直接发送逻辑调用我们刚刚写好的带装甲的发送函数 # 如果这里成功了说明要么是一次成功要么是经过几次重试后成功了 await _do_push_with_retry(payload) print(f✅ [跨服协同] 会话 {session_id} 报告推送成功) except Exception as e: # 核心兜底点如果代码走到这里说明 Tenacity 已经拼命重试了 3 次但 Java 服务器依然处于瘫痪状态 print(f❌ [网络绝望] 连续 3 次连接 Java 后端失败: {e}) # 触发终极绝招失败落库把 Payload 塞进 Redis 死信队列等待日后补偿 await save_failed_report_to_redis(user_id, session_id, payload)#