MQ 异步 回调外部系统集成模式详解一、要解决的核心问题当你的系统需要和外部系统第三方服务、下游平台协作完成一件事时天然面临三个难题外部系统慢或不稳定如果同步等待它返回你的主流程会被拖慢甚至拖垮处理是长流程对方可能需要人工介入或长时间处理不可能在一次 HTTP 请求内返回结果耦合风险外部系统宕机不应该导致你的核心业务不可用MQ 异步 回调就是解决这类问题的经典组合拳用 MQ 异步解决发起请求这一侧的解耦与削峰用回调解决获取结果这一侧的长流程异步通知注博客https://blog.csdn.net/badao_liumang_qizhi二、两个独立的概念先讲清2.1 消息队列MQ / Message Queue生产者把消息投递到队列就立即返回消费者独立地从队列取消息处理。核心价值价值说明解耦生产者不直接依赖消费者双方只认识队列异步生产者不等消费者处理完投递即返回削峰突发流量堆在队列里消费者按自己节奏消费容错消费失败可重试消费者宕机消息不丢2.2 回调Callback / WebhookA 系统请求 B 系统做一件耗时的事B 不在当次请求里返回结果而是事后主动调用 A 预留的接口把结果推回来。同步调用A 问 B「做完了吗」→ B 当场答请求-响应一来一回回调A 告诉 B「做完了打我这个电话」→ B 稍后主动打回来事件通知两次独立调用回调也叫Webhook当通知方式是 HTTP 回调时。三、组合起来一次完整的集成链路以向外部系统提交一个待处理任务并接收处理结果为例整个链路分两个方向【发起方向 —— MQ 异步】 业务系统 --(1)写业务数据-- DB 业务系统 --(2)发MQ------- 队列 | 消费者 --(3)消费-------- 消费者 --(4)调用外部系统提交任务-- 外部系统 消费者 --(5)返回任务ID(ack)------ 外部系统 // 只确认收到了,不是最终结果 消费者 --(6)保存任务ID与业务的映射-- DB ...... 外部系统内部处理 / 人工操作耗时不定 ...... 【结果方向 —— 回调】 外部系统 --(7)处理完,回调-- 业务系统的回调接口 业务系统 --(8)用任务ID反查业务-- DB 业务系统 --(9)处理结果,更新业务状态 业务系统 --(10)返回ack----- 外部系统关键点第(2)~(5)步用MQ 异步业务主流程只管写数据发消息不等外部系统响应飞快第(6)步保存任务ID ↔ 业务ID 的映射关系这是连接发起和回调两个独立调用的桥梁第(7)步的回调解决外部处理是长流程、结果不确定何时产生的问题四、通用示例代码4.1 发起侧业务提交 发 MQServicepublicclassTaskSubmitService{ResourceprivateTaskRepositorytaskRepository;ResourceprivateSubmitTaskMqSendermqSender;/** * 提交任务:只写本地数据 发MQ,不同步等外部系统 */Transactional(rollbackForException.class)publicLongsubmit(TaskRequestrequest){// 1. 写本地业务数据,状态待提交TaskEntitytasknewTaskEntity();task.setBizData(request.getData());task.setStatus(TaskStatus.PENDING.getCode());taskRepository.save(task);// 2. 事务提交后再发MQ(关键:避免消息发出但本地事务回滚)TransactionSynchronizationManager.registerSynchronization(newTransactionSynchronization(){OverridepublicvoidafterCommit(){mqSender.send(task.getId());}});returntask.getId();}}为什么事务提交后再发 MQ如果在事务里直接发 MQ可能出现消息已发出去、本地事务却回滚了导致消费者处理一个数据库里不存在的任务。注册afterCommit回调能保证消息只在本地数据真正落库后才发出。4.2 消费侧调用外部系统 保存映射ComponentSlf4jpublicclassSubmitTaskMqConsumer{ResourceprivateTaskRepositorytaskRepository;ResourceprivateExternalSystemClientexternalClient;RabbitListener(queuestask.submit.queue)publicvoidconsume(LongtaskId){log.info(消费提交任务MQ,taskId{},taskId);try{TaskEntitytasktaskRepository.findById(taskId).orElse(null);if(tasknull){log.warn(任务不存在,taskId{},taskId);return;// 幂等:消息对应数据不存在,直接丢弃}// 调用外部系统提交,外部只返回受理ID,不是最终结果ExternalResultresultexternalClient.submitTask(task.getBizData());// 关键:保存 外部任务ID - 本地业务ID 的映射,供回调时反查task.setExternalTaskId(result.getExternalTaskId());task.setStatus(TaskStatus.SUBMITTED.getCode());taskRepository.save(task);log.info(提交外部系统成功,taskId{},externalTaskId{},taskId,result.getExternalTaskId());}catch(Exceptione){// 消费异常:记录日志,交由MQ重试机制(或进死信队列)log.warn(提交任务消费异常,taskId{},taskId,e);throwe;// 抛出以触发重试}}}4.3 回调侧接收外部结果 更新业务RestControllerRequestMapping(/api/callback)Slf4jpublicclassTaskCallbackController{ResourceprivateTaskRepositorytaskRepository;/** * 外部系统处理完后主动回调此接口 * 入参格式由外部系统约定(常见form或json) */PostMapping(/task-result)publicResponseEntityBooleanonTaskResult(CallbackParamparam){log.info(收到外部回调,externalTaskId{},result{},param.getExternalTaskId(),param.getResult());// 1. 校验:关键字段为空直接返回成功(幂等容错,避免对方重试风暴)if(param.getExternalTaskId()null){log.warn(回调externalTaskId为空);returnResponseEntity.ok(Boolean.TRUE);}// 2. 用外部任务ID反查本地业务(发起侧保存的映射在此发挥作用)TaskEntitytasktaskRepository.findByExternalTaskId(param.getExternalTaskId());if(tasknull){log.warn(回调未查到本地任务,externalTaskId{},param.getExternalTaskId());returnResponseEntity.ok(Boolean.TRUE);// 幂等:查不到也返回成功}// 3. 幂等:已处理过的回调不重复处理if(TaskStatus.isFinished(task.getStatus())){log.info(任务已是终态,忽略重复回调,taskId{},task.getId());returnResponseEntity.ok(Boolean.TRUE);}// 4. 根据回调结果更新业务状态try{task.setStatus(param.isSuccess()?TaskStatus.SUCCESS.getCode():TaskStatus.FAILED.getCode());task.setResultMsg(param.getMessage());taskRepository.save(task);log.info(回调处理完成,taskId{},status{},task.getId(),task.getStatus());}catch(Exceptione){log.warn(回调处理异常,externalTaskId{},param.getExternalTaskId(),e);// 返回失败,让外部系统按其重试策略重试returnResponseEntity.ok(Boolean.FALSE);}returnResponseEntity.ok(Boolean.TRUE);}}五、这个模式里的关键设计点5.1 任务ID映射串联两次独立调用的桥梁发起时拿到外部系统的externalTaskId必须和本地bizId建立持久化映射。回调时外部系统只会带externalTaskId靠它反查本地业务。没有这个映射回调回来你不知道它对应哪笔业务。5.2 幂等性最重要回调可能被重复调用网络重试、对方故障恢复后补发所以回调接口必须幂等查不到记录 → 返回成功不报错已是终态 → 直接返回成功不重复处理用唯一键/状态机防止重复更新MQ 消费也可能重复消费侧同样要幂等。5.3 事务与消息的一致性事务提交后再发 MQ如 4.1避免消息发了、数据回滚了进阶方案本地消息表 / 事务消息如 RocketMQ 事务消息/ Outbox 模式保证写业务和发消息的最终一致5.4 回调接口的安全与返回约定鉴权回调接口要校验来源签名、token、IP 白名单防止被伪造调用返回值约定明确告诉对方成功/失败对方据此决定是否重试如返回 200true 表示已接收Content-Type按外部系统约定常见application/x-www-form-urlencoded或application/json5.5 可观测性全链路用关键 IDbizId、externalTaskId贯穿日志这样线上能把发起 → 提交 → 回调三段独立调用串成一条链路排查。5.6 兜底与补偿回调不是 100% 可靠可能丢失。生产级方案通常补充定时对账任务扫描长时间停在已提交未回调的记录主动查询外部系统状态超时处理超过约定时长未回调的任务标记异常/告警六、与其他方案的对比方案适用场景优点缺点同步调用外部处理快、结果立即可得简单直观外部慢会拖垮主流程强耦合轮询Polling外部无回调能力实现简单实时性差、空轮询浪费资源MQ 异步 回调外部处理慢/长流程、要求解耦解耦、削峰、不阻塞主流程复杂度高、需处理幂等/对账长连接推送(SSE/WebSocket)需要实时推送实时性强连接管理复杂不适合跨系统七、常见误区在事务内直接发 MQ可能消息发出但事务回滚 → 数据不一致。应事务后发或用事务消息。回调接口不做幂等对方重试导致重复处理如重复扣款、重复更新。忘记保存任务ID映射回调回来无法关联本地业务。回调完全信任、不鉴权回调接口被伪造请求攻击。只依赖回调、无兜底对账回调一旦丢失业务永远停在中间态。MQ 消费吞异常不重试catch后不抛出也不记录消息假装消费成功数据丢失。八、一句话总结MQ 异步让你发起请求时不被外部系统拖累解耦、削峰、不阻塞回调让你获取结果时不必苦等长流程事件驱动、异步通知。两者之间靠任务ID映射串联靠幂等 事务一致性 对账兜底保证可靠。它是对接慢、长、不稳定外部系统的标准范式。