当前位置： 首页 > news >正文

从 0 到 1 构建零丢失 RabbitMQ 数据同步堡垒：第三方接口数据零丢失的终极方案

news 2025/8/31 6:12:11

引言：

  • 本文总字数：约 19800 字
  • 预计阅读时间：50 分钟

数据不丢失为何如此重要？

在当今数据驱动的业务环境中，数据的完整性直接关系到企业的决策质量和业务连续性。想象一下，如果你是电商平台的技术负责人，第三方接口第三方支付接口的交易数据因为某种原因丢失，可能导致财务对账错误、用户投诉激增，甚至引发合规风险。

根据 Gartner 的研究报告，数据丢失事件的平均处理成本本已达到 420 万美元，而对于金融关键业务数据丢失导致的间接接业务中断中断可能使企业每天损失高达 500 万美元。这就是为什么构建一个能够100% 保证数据不丢失的同步系统至关重要。

本文将以 "使用 RabbitMQ 接收第三方方接口数据并保存到本地" 为场景，深入探讨可能导致数据丢失的 12 种典型场景，提供经过生产验证的解决方案，并通过完整的代码示例例展示如何构建一个真正可靠靠的数据同步体系。

一、整体架构设计：数据安全的第一道防线

在开始编码之前，我们需要设计一个能够抵御各种数据丢失风险的整体架构。一个健壮靠靠的数据同步系统应该包含以下核心组件：

核心设计原则

  1. 双重存储机制：接收的数据先写入本地日志表，再发送到 RabbitMQ
  2. 全链路确认机制：从 API 接收、MQ 传递到最终入库，每一步都有确认
  3. 分层防御策略：针对不同环节可能出现的问题，设置专门的防护措施
  4. 实时监控告警：对关键指标进行监控，异常情况及时告警
  5. 自动恢复能力：出现异常时，系统能自动或半自动恢复数据

二、环境准备：基础设施搭建

2.1 技术栈选择

组件版本说明
JDK17采用最新 LTS 版本，获得更好的性能和安全特性
Spring Boot3.2.0简化应用开发和配置
Spring AMQP3.2.0RabbitMQ 的 Spring 集成方案
RabbitMQ3.13.0消息中间件核心
MyBatis-Plus3.5.5数据库访问框架
MySQL8.3.0关系型数据库
Lombok1.18.30简化 Java 代码
Commons Lang33.14.0工具类库
SpringDoc OpenAPI2.1.0Swagger3 接口文档

2.2 Maven 依赖配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.2.0</version><relativePath/></parent><groupId>com.jamguo.rabbitmq</groupId><artifactId>third-party-data-sync</artifactId><version>0.0.1-SNAPSHOT</version><name>third-party-data-sync</name><description>第三方接口数据同步系统，基于RabbitMQ保证数据不丢失</description><properties><java.version>17</java.version></properties><dependencies><!-- Spring Boot Starter Web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Spring AMQP (RabbitMQ) --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!-- Lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.30</version><scope>provided</scope></dependency><!-- Commons Lang3 --><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.14.0</version></dependency><!-- MyBatis-Plus --><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.5.5</version></dependency><!-- MySQL Driver --><dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>8.3.0</version><scope>runtime</scope></dependency><!-- SpringDoc OpenAPI (Swagger3) --><dependency><groupId>org.springdoc</groupId><artifactId>springdoc-openapi-starter-webmvc-ui</artifactId><version>2.1.0</version></dependency><!-- Spring Boot Starter Test --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!-- Spring AMQP Test --><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><!-- HikariCP 连接池 --><dependency><groupId>com.zaxxer</groupId><artifactId>HikariCP</artifactId><version>5.0.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build>
</project>

2.3 配置文件

spring:application:name: third-party-data-sync# 数据库配置datasource:type: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/data_sync_db?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=trueusername: rootpassword: roothikari:maximum-pool-size: 20minimum-idle: 5idle-timeout: 300000connection-timeout: 20000max-lifetime: 1800000pool-name: DataSyncHikariCP# RabbitMQ配置rabbitmq:addresses: localhost:5672username: adminpassword: adminvirtual-host: /connection-timeout: 10000# 生产者确认配置publisher-confirm-type: correlatedpublisher-returns: true# 消费者配置listener:simple:acknowledge-mode: manual  # 手动确认concurrency: 5max-concurrency: 10prefetch: 10  # 每次预取10条消息retry:enabled: true  # 启用重试max-attempts: 3  # 最大重试次数initial-interval: 1000  # 初始重试间隔(毫秒)multiplier: 2  # 重试间隔乘数max-interval: 10000  # 最大重试间隔(毫秒)# MyBatis-Plus配置
mybatis-plus:mapper-locations: classpath*:mapper/**/*.xmltype-aliases-package: com.jamguo.sync.entityconfiguration:map-underscore-to-camel-case: truelog-impl: org.apache.ibatis.logging.slf4j.Slf4jImplglobal-config:db-config:id-type: assign_idlogic-delete-field: deletedlogic-delete-value: 1logic-not-delete-value: 0# 日志配置
logging:level:com.jamguo.sync: debugorg.springframework.amqp: infoorg.springframework.jdbc.core: warnpattern:console: "%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n"file: "%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n"file:name: logs/data-sync.log# 接口配置
third-party:api:# 签名密钥sign-secret: "your-sign-secret-key"# 数据接收路径receive-path: "/api/third-party/receive"# 最大数据大小限制(MB)max-data-size: 10# 自定义配置
sync:# 数据处理重试次数max-retry-count: 5# 数据过期时间(天)data-expire-days: 30# 死信队列延迟时间(毫秒)dead-letter-delay: 3600000# 定时任务执行间隔(分钟)schedule-interval: 5

2.4 数据库表设计

-- 创建数据库
CREATE DATABASE IF NOT EXISTS data_sync_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;USE data_sync_db;-- 接收日志表：存储所有接收到的第三方数据
CREATE TABLE IF NOT EXISTS data_receive_log (id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID',data_id VARCHAR(64) NOT NULL COMMENT '第三方数据ID',data_type VARCHAR(32) NOT NULL COMMENT '数据类型',data_content TEXT NOT NULL COMMENT '数据内容',source_system VARCHAR(64) NOT NULL COMMENT '来源系统',receive_time DATETIME NOT NULL COMMENT '接收时间',status TINYINT NOT NULL DEFAULT 0 COMMENT '状态：0-初始，1-已发送到MQ，2-已处理成功，3-处理失败，4-已丢弃',mq_message_id VARCHAR(64) COMMENT 'MQ消息ID',process_time DATETIME COMMENT '处理时间',process_result VARCHAR(255) COMMENT '处理结果',retry_count TINYINT NOT NULL DEFAULT 0 COMMENT '重试次数',created_by VARCHAR(64) NOT NULL DEFAULT 'system' COMMENT '创建人',created_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',updated_by VARCHAR(64) NOT NULL DEFAULT 'system' COMMENT '更新人',updated_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',deleted TINYINT NOT NULL DEFAULT 0 COMMENT '删除标识：0-未删除，1-已删除',version INT NOT NULL DEFAULT 0 COMMENT '乐观锁版本',PRIMARY KEY (id),UNIQUE KEY uk_data_id (data_id, source_system, deleted),KEY idx_status_receive_time (status, receive_time),KEY idx_data_type (data_type),KEY idx_receive_time (receive_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='第三方数据接收日志表';-- 业务数据表：存储处理后的业务数据
CREATE TABLE IF NOT EXISTS business_data (id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID',data_id VARCHAR(64) NOT NULL COMMENT '第三方数据ID',source_system VARCHAR(64) NOT NULL COMMENT '来源系统',data_type VARCHAR(32) NOT NULL COMMENT '数据类型',business_no VARCHAR(64) COMMENT '业务编号',data_content JSON NOT NULL COMMENT '处理后的数据内容',status TINYINT NOT NULL DEFAULT 1 COMMENT '状态：1-有效，0-无效',created_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',updated_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',PRIMARY KEY (id),UNIQUE KEY uk_data_id (data_id, source_system),KEY idx_business_no (business_no),KEY idx_data_type (data_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='业务数据表';-- 数据同步异常表：记录处理异常的数据
CREATE TABLE IF NOT EXISTS data_sync_exception (id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID',receive_log_id BIGINT NOT NULL COMMENT '接收日志ID',data_id VARCHAR(64) NOT NULL COMMENT '第三方数据ID',error_type VARCHAR(32) NOT NULL COMMENT '错误类型',error_message TEXT NOT NULL COMMENT '错误信息',exception_stack TEXT COMMENT '异常堆栈',handle_status TINYINT NOT NULL DEFAULT 0 COMMENT '处理状态：0-未处理，1-已处理',created_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',updated_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',PRIMARY KEY (id),KEY idx_receive_log_id (receive_log_id),KEY idx_data_id (data_id),KEY idx_handle_status (handle_status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据同步异常表';

三、核心代码实现：层层设防的数据同步系统

3.1 常量定义

package com.jamguo.sync.constant;/*** 系统常量类** @author 果酱*/
public class SyncConstant {/*** RabbitMQ相关常量*/public static class RabbitMq {/*** 数据同步交换机*/public static final String DATA_SYNC_EXCHANGE = "data.sync.exchange";/*** 数据同步队列*/public static final String DATA_SYNC_QUEUE = "data.sync.queue";/*** 数据同步路由键*/public static final String DATA_SYNC_ROUTING_KEY = "data.sync.routing.key";/*** 死信交换机*/public static final String DEAD_LETTER_EXCHANGE = "data.sync.dlq.exchange";/*** 死信队列*/public static final String DEAD_LETTER_QUEUE = "data.sync.dlq.queue";/*** 死信路由键*/public static final String DEAD_LETTER_ROUTING_KEY = "data.sync.dlq.routing.key";/*** 重试交换机*/public static final String RETRY_EXCHANGE = "data.sync.retry.exchange";/*** 重试队列*/public static final String RETRY_QUEUE = "data.sync.retry.queue";/*** 重试路由键*/public static final String RETRY_ROUTING_KEY = "data.sync.retry.routing.key";}/*** 数据接收状态*/public static class ReceiveStatus {/*** 初始状态*/public static final int INIT = 0;/*** 已发送到MQ*/public static final int SENT_TO_MQ = 1;/*** 已处理成功*/public static final int PROCESSED_SUCCESS = 2;/*** 处理失败*/public static final int PROCESSED_FAILED = 3;/*** 已丢弃*/public static final int DISCARDED = 4;}/*** 异常处理状态*/public static class ExceptionHandleStatus {/*** 未处理*/public static final int UNHANDLED = 0;/*** 已处理*/public static final int HANDLED = 1;}/*** 消息头常量*/public static class MessageHeader {/*** 数据ID*/public static final String DATA_ID = "DATA_ID";/*** 接收日志ID*/public static final String RECEIVE_LOG_ID = "RECEIVE_LOG_ID";/*** 重试次数*/public static final String RETRY_COUNT = "RETRY_COUNT";}
}

3.2 实体类定义

package com.jamguo.sync.entity;import com.baomidou.mybatisplus.annotation.*;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;import java.time.LocalDateTime;/*** 数据接收日志实体类** @author 果酱*/
@Data
@TableName("data_receive_log")
@Schema(description = "数据接收日志实体")
public class DataReceiveLog {@TableId(type = IdType.AUTO)@Schema(description = "主键ID")private Long id;@Schema(description = "第三方数据ID")private String dataId;@Schema(description = "数据类型")private String dataType;@Schema(description = "数据内容")private String dataContent;@Schema(description = "来源系统")private String sourceSystem;@Schema(description = "接收时间")@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime receiveTime;@Schema(description = "状态：0-初始，1-已发送到MQ，2-已处理成功，3-处理失败，4-已丢弃")private Integer status;@Schema(description = "MQ消息ID")private String mqMessageId;@Schema(description = "处理时间")@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime processTime;@Schema(description = "处理结果")private String processResult;@Schema(description = "重试次数")private Integer retryCount;@Schema(description = "创建人")@TableField(fill = FieldFill.INSERT)private String createdBy;@Schema(description = "创建时间")@TableField(fill = FieldFill.INSERT)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime createdTime;@Schema(description = "更新人")@TableField(fill = FieldFill.INSERT_UPDATE)private String updatedBy;@Schema(description = "更新时间")@TableField(fill = FieldFill.INSERT_UPDATE)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime updatedTime;@Schema(description = "删除标识：0-未删除，1-已删除")@TableLogicprivate Integer deleted;@Schema(description = "乐观锁版本")@Versionprivate Integer version;
}

package com.jamguo.sync.entity;import com.baomidou.mybatisplus.annotation.*;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;import java.time.LocalDateTime;/*** 业务数据实体类** @author 果酱*/
@Data
@TableName("business_data")
@Schema(description = "业务数据实体")
public class BusinessData {@TableId(type = IdType.AUTO)@Schema(description = "主键ID")private Long id;@Schema(description = "第三方数据ID")private String dataId;@Schema(description = "来源系统")private String sourceSystem;@Schema(description = "数据类型")private String dataType;@Schema(description = "业务编号")private String businessNo;@Schema(description = "处理后的数据内容")private String dataContent;@Schema(description = "状态：1-有效，0-无效")private Integer status;@Schema(description = "创建时间")@TableField(fill = FieldFill.INSERT)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime createdTime;@Schema(description = "更新时间")@TableField(fill = FieldFill.INSERT_UPDATE)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime updatedTime;
}

package com.jamguo.sync.entity;import com.baomidou.mybatisplus.annotation.*;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;import java.time.LocalDateTime;/*** 数据同步异常实体类** @author 果酱*/
@Data
@TableName("data_sync_exception")
@Schema(description = "数据同步异常实体")
public class DataSyncException {@TableId(type = IdType.AUTO)@Schema(description = "主键ID")private Long id;@Schema(description = "接收日志ID")private Long receiveLogId;@Schema(description = "第三方数据ID")private String dataId;@Schema(description = "错误类型")private String errorType;@Schema(description = "错误信息")private String errorMessage;@Schema(description = "异常堆栈")private String exceptionStack;@Schema(description = "处理状态：0-未处理，1-已处理")private Integer handleStatus;@Schema(description = "创建时间")@TableField(fill = FieldFill.INSERT)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime createdTime;@Schema(description = "更新时间")@TableField(fill = FieldFill.INSERT_UPDATE)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private LocalDateTime updatedTime;
}

package com.jamguo.sync.dto;import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;import java.io.Serializable;/*** 第三方数据请求DTO** @author 果酱*/
@Data
@Schema(description = "第三方数据请求DTO")
public class ThirdPartyDataDTO implements Serializable {private static final long serialVersionUID = 1L;@Schema(description = "数据ID，唯一标识一条数据")private String dataId;@Schema(description = "数据类型")private String dataType;@Schema(description = "来源系统")private String sourceSystem;@Schema(description = "数据内容，JSON格式")private String dataContent;@Schema(description = "签名")private String sign;@Schema(description = "时间戳")private Long timestamp;
}

3.3 RabbitMQ 配置

package com.jamguo.sync.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import com.jamguo.sync.constant.SyncConstant;import lombok.extern.slf4j.Slf4j;import java.util.HashMap;
import java.util.Map;/*** RabbitMQ配置类，配置交换机、队列及绑定关系* 重点配置了死信队列和重试机制，防止消息丢失** @author 果酱*/
@Configuration
@Slf4j
public class RabbitMqConfig {/*** 死信队列延迟时间(毫秒)*/@Value("${sync.dead-letter-delay}")private long deadLetterDelay;/*** 创建数据同步交换机* 采用Direct类型，持久化** @return 数据同步交换机*/@Beanpublic DirectExchange dataSyncExchange() {// 参数说明：交换机名称、是否持久化、是否自动删除、附加参数DirectExchange exchange = new DirectExchange(SyncConstant.RabbitMq.DATA_SYNC_EXCHANGE,true,false,null);log.info("创建数据同步交换机: {}", SyncConstant.RabbitMq.DATA_SYNC_EXCHANGE);return exchange;}/*** 创建重试交换机** @return 重试交换机*/@Beanpublic DirectExchange retryExchange() {DirectExchange exchange = new DirectExchange(SyncConstant.RabbitMq.RETRY_EXCHANGE,true,false,null);log.info("创建重试交换机: {}", SyncConstant.RabbitMq.RETRY_EXCHANGE);return exchange;}/*** 创建死信交换机** @return 死信交换机*/@Beanpublic DirectExchange deadLetterExchange() {DirectExchange exchange = new DirectExchange(SyncConstant.RabbitMq.DEAD_LETTER_EXCHANGE,true,false,null);log.info("创建死信交换机: {}", SyncConstant.RabbitMq.DEAD_LETTER_EXCHANGE);return exchange;}/*** 创建数据同步队列* 配置了死信队列，当消息处理失败达到最大重试次数后，会被路由到死信队列** @return 数据同步队列*/@Beanpublic Queue dataSyncQueue() {// 设置队列参数，主要是死信相关配置Map<String, Object> args = new HashMap<>(3);// 死信交换机args.put("x-dead-letter-exchange", SyncConstant.RabbitMq.RETRY_EXCHANGE);// 死信路由键args.put("x-dead-letter-routing-key", SyncConstant.RabbitMq.RETRY_ROUTING_KEY);// 队列消息过期时间(毫秒)，防止消息一直未被处理args.put("x-message-ttl", 86400000); // 24小时// 参数说明：队列名称、是否持久化、是否排他、是否自动删除、附加参数Queue queue = QueueBuilder.durable(SyncConstant.RabbitMq.DATA_SYNC_QUEUE).withArguments(args).build();log.info("创建数据同步队列: {}", SyncConstant.RabbitMq.DATA_SYNC_QUEUE);return queue;}/*** 创建重试队列* 用于暂时存储需要重试处理的消息** @return 重试队列*/@Beanpublic Queue retryQueue() {// 设置队列参数Map<String, Object> args = new HashMap<>(3);// 死信交换机（重试后还是失败，进入死信队列）args.put("x-dead-letter-exchange", SyncConstant.RabbitMq.DEAD_LETTER_EXCHANGE);// 死信路由键args.put("x-dead-letter-routing-key", SyncConstant.RabbitMq.DEAD_LETTER_ROUTING_KEY);// 消息过期时间，即重试间隔args.put("x-message-ttl", deadLetterDelay);Queue queue = QueueBuilder.durable(SyncConstant.RabbitMq.RETRY_QUEUE).withArguments(args).build();log.info("创建重试队列: {}，重试间隔: {}ms", SyncConstant.RabbitMq.RETRY_QUEUE, deadLetterDelay);return queue;}/*** 创建死信队列* 存储最终处理失败的消息，等待人工干预** @return 死信队列*/@Beanpublic Queue deadLetterQueue() {Queue queue = QueueBuilder.durable(SyncConstant.RabbitMq.DEAD_LETTER_QUEUE).build();log.info("创建死信队列: {}", SyncConstant.RabbitMq.DEAD_LETTER_QUEUE);return queue;}/*** 绑定数据同步队列到数据同步交换机** @param dataSyncQueue 数据同步队列* @param dataSyncExchange 数据同步交换机* @return 绑定关系*/@Beanpublic Binding bindDataSyncQueue(Queue dataSyncQueue, DirectExchange dataSyncExchange) {Binding binding = BindingBuilder.bind(dataSyncQueue).to(dataSyncExchange).with(SyncConstant.RabbitMq.DATA_SYNC_ROUTING_KEY);log.info("绑定队列 {} 到交换机 {}，路由键: {}",SyncConstant.RabbitMq.DATA_SYNC_QUEUE,SyncConstant.RabbitMq.DATA_SYNC_EXCHANGE,SyncConstant.RabbitMq.DATA_SYNC_ROUTING_KEY);return binding;}/*** 绑定重试队列到重试交换机** @param retryQueue 重试队列* @param retryExchange 重试交换机* @return 绑定关系*/@Beanpublic Binding bindRetryQueue(Queue retryQueue, DirectExchange retryExchange) {Binding binding = BindingBuilder.bind(retryQueue).to(retryExchange).with(SyncConstant.RabbitMq.RETRY_ROUTING_KEY);log.info("绑定队列 {} 到交换机 {}，路由键: {}",SyncConstant.RabbitMq.RETRY_QUEUE,SyncConstant.RabbitMq.RETRY_EXCHANGE,SyncConstant.RabbitMq.RETRY_ROUTING_KEY);return binding;}/*** 绑定死信队列到死信交换机** @param deadLetterQueue 死信队列* @param deadLetterExchange 死信交换机* @return 绑定关系*/@Beanpublic Binding bindDeadLetterQueue(Queue deadLetterQueue, DirectExchange deadLetterExchange) {Binding binding = BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with(SyncConstant.RabbitMq.DEAD_LETTER_ROUTING_KEY);log.info("绑定队列 {} 到交换机 {}，路由键: {}",SyncConstant.RabbitMq.DEAD_LETTER_QUEUE,SyncConstant.RabbitMq.DEAD_LETTER_EXCHANGE,SyncConstant.RabbitMq.DEAD_LETTER_ROUTING_KEY);return binding;}/*** 配置RabbitTemplate，设置消息确认和返回回调** @param connectionFactory 连接工厂* @return RabbitTemplate实例*/@Beanpublic RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 设置消息转换器，使用Jackson2JsonMessageConverterrabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());// 设置确认回调：确认消息是否到达交换机rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {String messageId = correlationData != null ? correlationData.getId() : "unknown";if (ack) {log.info("消息 [{}] 成功发送到交换机", messageId);} else {log.error("消息 [{}] 发送到交换机失败，原因: {}", messageId, cause);// 消息发送失败，记录到数据库，后续由定时任务重试if (correlationData instanceof DataSyncCorrelationData dataSyncCorrelationData) {log.error("准备记录发送失败的消息，接收日志ID: {}", dataSyncCorrelationData.getReceiveLogId());// 调用服务记录发送失败状态dataSyncService.handleMessageSendFailed(dataSyncCorrelationData.getReceiveLogId(), messageId, cause);}}});// 设置返回回调：当消息发送到交换机但无法路由到队列时触发rabbitTemplate.setReturnsCallback(returnedMessage -> {String messageId = returnedMessage.getMessage().getMessageProperties().getMessageId();log.error("消息 [{}] 路由失败: 交换机={}, 路由键={}, 回复码={}, 回复文本={}",messageId,returnedMessage.getExchange(),returnedMessage.getRoutingKey(),returnedMessage.getReplyCode(),returnedMessage.getReplyText());// 解析消息头中的接收日志IDString receiveLogIdStr = returnedMessage.getMessage().getMessageProperties().getHeader(SyncConstant.MessageHeader.RECEIVE_LOG_ID);if (StringUtils.hasText(receiveLogIdStr)) {try {Long receiveLogId = Long.parseLong(receiveLogIdStr);// 记录路由失败状态dataSyncService.handleMessageRouteFailed(receiveLogId, messageId, returnedMessage.getReplyText());} catch (NumberFormatException e) {log.error("解析接收日志ID失败: {}", receiveLogIdStr, e);}}});return rabbitTemplate;}@Autowiredprivate DataSyncService dataSyncService;
}

3.4 自定义 CorrelationData

package com.jamguo.sync.config;import org.springframework.amqp.rabbit.connection.CorrelationData;import lombok.Getter;
import lombok.Setter;/*** 数据同步相关的CorrelationData，扩展了原生的CorrelationData* 增加了接收日志ID，方便在消息确认回调中使用** @author 果酱*/
@Getter
@Setter
public class DataSyncCorrelationData extends CorrelationData {/*** 接收日志ID*/private final Long receiveLogId;/*** 构造方法** @param id 消息ID* @param receiveLogId 接收日志ID*/public DataSyncCorrelationData(String id, Long receiveLogId) {super(id);this.receiveLogId = receiveLogId;}
}

3.5 Mapper 层

package com.jamguo.sync.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jamguo.sync.entity.DataReceiveLog;
import org.apache.ibatis.annotations.Param;import java.util.List;/*** 数据接收日志Mapper** @author 果酱*/
public interface DataReceiveLogMapper extends BaseMapper<DataReceiveLog> {/*** 根据数据ID和来源系统查询接收日志** @param dataId 数据ID* @param sourceSystem 来源系统* @return 接收日志实体*/DataReceiveLog selectByDataIdAndSourceSystem(@Param("dataId") String dataId,@Param("sourceSystem") String sourceSystem);/*** 查询需要重发的消息** @param status 状态* @param maxRetryCount 最大重试次数* @param limit 查询数量限制* @return 需要重发的消息列表*/List<DataReceiveLog> selectNeedResendMessages(@Param("status") Integer status,@Param("maxRetryCount") Integer maxRetryCount,@Param("limit") Integer limit);/*** 更新消息发送MQ状态** @param id 日志ID* @param status 状态* @param mqMessageId MQ消息ID* @param retryCount 重试次数* @param version 版本号* @return 更新数量*/int updateSendMqStatus(@Param("id") Long id,@Param("status") Integer status,@Param("mqMessageId") String mqMessageId,@Param("retryCount") Integer retryCount,@Param("version") Integer version);
}

3.6 Service 层

package com.jamguo.sync.service;import com.baomidou.mybatisplus.extension.service.IService;
import com.jamguo.sync.dto.ThirdPartyDataDTO;
import com.jamguo.sync.entity.DataReceiveLog;/*** 数据接收日志服务接口** @author 果酱*/
public interface DataReceiveLogService extends IService<DataReceiveLog> {/*** 根据数据ID和来源系统查询接收日志** @param dataId 数据ID* @param sourceSystem 来源系统* @return 接收日志实体*/DataReceiveLog getByDataIdAndSourceSystem(String dataId, String sourceSystem);/*** 保存接收日志并发送到MQ** @param data 第三方数据DTO* @return 接收日志ID*/Long saveAndSendToMq(ThirdPartyDataDTO data);/*** 处理消息发送失败** @param receiveLogId 接收日志ID* @param mqMessageId MQ消息ID* @param errorMsg 错误信息*/void handleMessageSendFailed(Long receiveLogId, String mqMessageId, String errorMsg);/*** 处理消息路由失败** @param receiveLogId 接收日志ID* @param mqMessageId MQ消息ID* @param errorMsg 错误信息*/void handleMessageRouteFailed(Long receiveLogId, String mqMessageId, String errorMsg);/*** 处理消息处理成功** @param receiveLogId 接收日志ID* @param result 处理结果*/void handleMessageProcessSuccess(Long receiveLogId, String result);/*** 处理消息处理失败** @param receiveLogId 接收日志ID* @param errorMsg 错误信息* @param retryCount 重试次数*/void handleMessageProcessFailed(Long receiveLogId, String errorMsg, int retryCount);/*** 重发未成功发送到MQ的消息** @param limit 每次处理的数量限制* @return 重发成功的数量*/int resendFailedMessages(int limit);
}

package com.jamguo.sync.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.jamguo.sync.config.DataSyncCorrelationData;
import com.jamguo.sync.constant.SyncConstant;
import com.jamguo.sync.dto.ThirdPartyDataDTO;
import com.jamguo.sync.entity.DataReceiveLog;
import com.jamguo.sync.entity.DataSyncException;
import com.jamguo.sync.mapper.DataReceiveLogMapper;
import com.jamguo.sync.service.DataReceiveLogService;
import com.jamguo.sync.service.DataSyncExceptionService;import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.UUID;import lombok.extern.slf4j.Slf4j;import javax.annotation.Resource;/*** 数据接收日志服务实现类** @author 果酱*/
@Service
@Slf4j
public class DataReceiveLogServiceImpl extends ServiceImpl<DataReceiveLogMapper, DataReceiveLog>implements DataReceiveLogService {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate DataSyncExceptionService dataSyncExceptionService;@Value("${sync.max-retry-count}")private int maxRetryCount;/*** 消息转换器*/private final Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();@Overridepublic DataReceiveLog getByDataIdAndSourceSystem(String dataId, String sourceSystem) {StringUtils.hasText(dataId, "数据ID不能为空");StringUtils.hasText(sourceSystem, "来源系统不能为空");return baseMapper.selectByDataIdAndSourceSystem(dataId, sourceSystem);}@Transactional(rollbackFor = Exception.class)@Overridepublic Long saveAndSendToMq(ThirdPartyDataDTO data) {Objects.requireNonNull(data, "第三方数据不能为空");StringUtils.hasText(data.getDataId(), "数据ID不能为空");StringUtils.hasText(data.getSourceSystem(), "来源系统不能为空");StringUtils.hasText(data.getDataType(), "数据类型不能为空");StringUtils.hasText(data.getDataContent(), "数据内容不能为空");// 检查是否已存在相同的数据DataReceiveLog existingLog = getByDataIdAndSourceSystem(data.getDataId(), data.getSourceSystem());if (existingLog != null) {log.info("数据已存在，dataId: {}, sourceSystem: {}", data.getDataId(), data.getSourceSystem());// 如果已存在且已处理成功，则直接返回if (SyncConstant.ReceiveStatus.PROCESSED_SUCCESS == existingLog.getStatus()) {return existingLog.getId();}// 如果已存在但未处理成功，则更新数据内容并尝试重新发送return updateAndResend(existingLog, data);}// 创建新的接收日志DataReceiveLog receiveLog = new DataReceiveLog();receiveLog.setDataId(data.getDataId());receiveLog.setDataType(data.getDataType());receiveLog.setDataContent(data.getDataContent());receiveLog.setSourceSystem(data.getSourceSystem());receiveLog.setReceiveTime(LocalDateTime.now());receiveLog.setStatus(SyncConstant.ReceiveStatus.INIT);receiveLog.setRetryCount(0);// 保存到数据库int insert = baseMapper.insert(receiveLog);if (insert != 1) {log.error("保存数据接收日志失败，dataId: {}", data.getDataId());throw new RuntimeException("保存数据接收日志失败");}log.info("数据接收日志保存成功，ID: {}, dataId: {}", receiveLog.getId(), data.getDataId());// 发送到MQsendToMq(receiveLog);return receiveLog.getId();}/*** 更新已有日志并重新发送** @param existingLog 已存在的日志* @param data 新的数据* @return 日志ID*/private Long updateAndResend(DataReceiveLog existingLog, ThirdPartyDataDTO data) {log.info("准备更新数据并重新发送，ID: {}, dataId: {}", existingLog.getId(), data.getDataId());// 更新数据内容和接收时间existingLog.setDataContent(data.getDataContent());existingLog.setDataType(data.getDataType());existingLog.setReceiveTime(LocalDateTime.now());existingLog.setRetryCount(existingLog.getRetryCount() + 1);// 如果之前是发送失败状态，重置为初始状态if (SyncConstant.ReceiveStatus.PROCESSED_FAILED == existingLog.getStatus() || SyncConstant.ReceiveStatus.INIT == existingLog.getStatus()) {existingLog.setStatus(SyncConstant.ReceiveStatus.INIT);}// 更新到数据库int update = baseMapper.updateById(existingLog);if (update != 1) {log.error("更新数据接收日志失败，ID: {}", existingLog.getId());throw new RuntimeException("更新数据接收日志失败");}log.info("数据接收日志更新成功，ID: {}, dataId: {}", existingLog.getId(), data.getDataId());// 发送到MQsendToMq(existingLog);return existingLog.getId();}/*** 发送消息到MQ** @param receiveLog 接收日志*/private void sendToMq(DataReceiveLog receiveLog) {try {// 创建消息内容String dataContent = receiveLog.getDataContent();// 构建消息，设置消息头Message message = MessageBuilder.withBody(dataContent.getBytes()).setContentType("application/json").setMessageId(UUID.randomUUID().toString()).setHeader(SyncConstant.MessageHeader.DATA_ID, receiveLog.getDataId()).setHeader(SyncConstant.MessageHeader.RECEIVE_LOG_ID, receiveLog.getId().toString()).setHeader(SyncConstant.MessageHeader.RETRY_COUNT, receiveLog.getRetryCount()).build();// 创建带接收日志ID的CorrelationDataDataSyncCorrelationData correlationData = new DataSyncCorrelationData(message.getMessageProperties().getMessageId(), receiveLog.getId());log.info("准备发送消息到MQ，ID: {}, dataId: {}, 消息ID: {}",receiveLog.getId(), receiveLog.getDataId(), message.getMessageProperties().getMessageId());// 发送消息rabbitTemplate.send(SyncConstant.RabbitMq.DATA_SYNC_EXCHANGE,SyncConstant.RabbitMq.DATA_SYNC_ROUTING_KEY,message,correlationData);log.info("消息发送成功（等待确认），ID: {}, dataId: {}",receiveLog.getId(), receiveLog.getDataId());} catch (Exception e) {log.error("发送消息到MQ时发生异常，ID: {}, dataId: {}",receiveLog.getId(), receiveLog.getDataId(), e);// 记录发送失败状态handleMessageSendFailed(receiveLog.getId(), null, e.getMessage());// 抛出异常，触发事务回滚throw new RuntimeException("发送消息到MQ失败", e);}}@Transactional(rollbackFor = Exception.class)@Overridepublic void handleMessageSendFailed(Long receiveLogId, String mqMessageId, String errorMsg) {Objects.requireNonNull(receiveLogId, "接收日志ID不能为空");StringUtils.hasText(errorMsg, "错误信息不能为空");log.error("处理消息发送失败，ID: {}, 错误信息: {}", receiveLogId, errorMsg);// 查询日志DataReceiveLog receiveLog = baseMapper.selectById(receiveLogId);if (receiveLog == null) {log.error("接收日志不存在，ID: {}", receiveLogId);return;}// 更新状态为处理失败receiveLog.setStatus(SyncConstant.ReceiveStatus.PROCESSED_FAILED);receiveLog.setMqMessageId(mqMessageId);receiveLog.setProcessResult("消息发送失败: " + errorMsg);receiveLog.setRetryCount(receiveLog.getRetryCount() + 1);receiveLog.setUpdatedTime(LocalDateTime.now());// 更新到数据库int update = baseMapper.updateById(receiveLog);if (update != 1) {log.error("更新消息发送失败状态失败，ID: {}", receiveLogId);} else {log.info("消息发送失败状态更新成功，ID: {}", receiveLogId);}// 记录异常信息DataSyncException exception = new DataSyncException();exception.setReceiveLogId(receiveLogId);exception.setDataId(receiveLog.getDataId());exception.setErrorType("MESSAGE_SEND_FAILED");exception.setErrorMessage(errorMsg);exception.setHandleStatus(SyncConstant.ExceptionHandleStatus.UNHANDLED);dataSyncExceptionService.save(exception);}@Transactional(rollbackFor = Exception.class)@Overridepublic void handleMessageRouteFailed(Long receiveLogId, String mqMessageId, String errorMsg) {Objects.requireNonNull(receiveLogId, "接收日志ID不能为空");StringUtils.hasText(errorMsg, "错误信息不能为空");log.error("处理消息路由失败，ID: {}, 错误信息: {}", receiveLogId, errorMsg);// 查询日志DataReceiveLog receiveLog = baseMapper.selectById(receiveLogId);if (receiveLog == null) {log.error("接收日志不存在，ID: {}", receiveLogId);return;}// 更新状态为处理失败receiveLog.setStatus(SyncConstant.ReceiveStatus.PROCESSED_FAILED);receiveLog.setMqMessageId(mqMessageId);receiveLog.setProcessResult("消息路由失败: " + errorMsg);receiveLog.setRetryCount(receiveLog.getRetryCount() + 1);receiveLog.setUpdatedTime(LocalDateTime.now());// 更新到数据库int update = baseMapper.updateById(receiveLog);if (update != 1) {log.error("更新消息路由失败状态失败，ID: {}", receiveLogId);} else {log.info("消息路由失败状态更新成功，ID: {}", receiveLogId);}// 记录异常信息DataSyncException exception = new DataSyncException();exception.setReceiveLogId(receiveLogId);exception.setDataId(receiveLog.getDataId());exception.setErrorType("MESSAGE_ROUTE_FAILED");exception.setErrorMessage(errorMsg);exception.setHandleStatus(SyncConstant.ExceptionHandleStatus.UNHANDLED);dataSyncExceptionService.save(exception);}@Transactional(rollbackFor = Exception.class)@Overridepublic void handleMessageProcessSuccess(Long receiveLogId, String result) {Objects.requireNonNull(receiveLogId, "接收日志ID不能为空");log.info("处理消息处理成功，ID: {}, 处理结果: {}", receiveLogId, result);// 查询日志DataReceiveLog receiveLog = baseMapper.selectById(receiveLogId);if (receiveLog == null) {log.error("接收日志不存在，ID: {}", receiveLogId);return;}// 更新状态为处理成功receiveLog.setStatus(SyncConstant.ReceiveStatus.PROCESSED_SUCCESS);receiveLog.setProcessTime(LocalDateTime.now());receiveLog.setProcessResult(StringUtils.defaultIfBlank(result, "处理成功"));receiveLog.setUpdatedTime(LocalDateTime.now());// 更新到数据库int update = baseMapper.updateById(receiveLog);if (update != 1) {log.error("更新消息处理成功状态失败，ID: {}", receiveLogId);} else {log.info("消息处理成功状态更新成功，ID: {}", receiveLogId);}}@Transactional(rollbackFor = Exception.class)@Overridepublic void handleMessageProcessFailed(Long receiveLogId, String errorMsg, int retryCount) {Objects.requireNonNull(receiveLogId, "接收日志ID不能为空");StringUtils.hasText(errorMsg, "错误信息不能为空");log.error("处理消息处理失败，ID: {}, 重试次数: {}, 错误信息: {}", receiveLogId, retryCount, errorMsg);// 查询日志DataReceiveLog receiveLog = baseMapper.selectById(receiveLogId);if (receiveLog == null) {log.error("接收日志不存在，ID: {}", receiveLogId);return;}// 更新状态和重试次数receiveLog.setStatus(SyncConstant.ReceiveStatus.PROCESSED_FAILED);receiveLog.setProcessResult("处理失败: " + errorMsg);receiveLog.setRetryCount(retryCount);receiveLog.setUpdatedTime(LocalDateTime.now());// 更新到数据库int update = baseMapper.updateById(receiveLog);if (update != 1) {log.error("更新消息处理失败状态失败，ID: {}", receiveLogId);} else {log.info("消息处理失败状态更新成功，ID: {}", receiveLogId);}// 记录异常信息DataSyncException exception = new DataSyncException();exception.setReceiveLogId(receiveLogId);exception.setDataId(receiveLog.getDataId());exception.setErrorType("MESSAGE_PROCESS_FAILED");exception.setErrorMessage(errorMsg);exception.setHandleStatus(SyncConstant.ExceptionHandleStatus.UNHANDLED);dataSyncExceptionService.save(exception);}@Overridepublic int resendFailedMessages(int limit) {if (limit <= 0) {limit = 100;}log.info("开始重发失败的消息，每次处理数量: {}", limit);// 查询需要重发的消息List<DataReceiveLog> needResendLogs = baseMapper.selectNeedResendMessages(SyncConstant.ReceiveStatus.PROCESSED_FAILED,maxRetryCount,limit);if (org.springframework.util.CollectionUtils.isEmpty(needResendLogs)) {log.info("没有需要重发的消息");return 0;}log.info("共查询到 {} 条需要重发的消息", needResendLogs.size());int successCount = 0;for (DataReceiveLog log : needResendLogs) {try {// 尝试发送消息sendToMq(log);successCount++;} catch (Exception e) {log.error("重发消息失败，ID: {}, dataId: {}", log.getId(), log.getDataId(), e);}}log.info("消息重发完成，共处理 {} 条，成功 {} 条", needResendLogs.size(), successCount);return successCount;}
}

3.7 数据接收控制器

package com.jamguo.sync.controller;import com.jamguo.sync.dto.ThirdPartyDataDTO;
import com.jamguo.sync.service.DataReceiveLogService;
import com.jamguo.sync.service.SignatureService;import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;import java.util.Objects;import lombok.extern.slf4j.Slf4j;/*** 第三方数据接收控制器* 负责接收第三方系统推送的数据，并进行初步验证** @author 果酱*/
@RestController
@Slf4j
@Tag(name = "第三方数据接收接口", description = "接收第三方系统推送的数据")
public class ThirdPartyDataController {private final DataReceiveLogService dataReceiveLogService;private final SignatureService signatureService;@Autowiredpublic ThirdPartyDataController(DataReceiveLogService dataReceiveLogService,SignatureService signatureService) {this.dataReceiveLogService = dataReceiveLogService;this.signatureService = signatureService;}/*** 接收第三方数据** @param data 第三方数据DTO* @return 响应结果*/@PostMapping("/api/third-party/receive")@Operation(summary = "接收第三方数据", description = "接收第三方系统推送的数据，进行签名验证后保存并处理")public ResponseEntity<String> receiveData(@Parameter(description = "第三方数据", required = true) @RequestBody ThirdPartyDataDTO data) {try {log.info("收到第三方数据，dataId: {}, sourceSystem: {}", data.getDataId(), data.getSourceSystem());// 参数校验validateData(data);// 签名验证boolean verifyResult = signatureService.verifySignature(data);if (!verifyResult) {log.error("数据签名验证失败，dataId: {}", data.getDataId());return ResponseEntity.badRequest().body("签名验证失败");}// 保存并发送到MQLong receiveLogId = dataReceiveLogService.saveAndSendToMq(data);log.info("数据接收处理完成，receiveLogId: {}, dataId: {}", receiveLogId, data.getDataId());return ResponseEntity.ok("数据接收成功，ID: " + receiveLogId);} catch (IllegalArgumentException e) {log.error("数据参数验证失败", e);return ResponseEntity.badRequest().body(e.getMessage());} catch (Exception e) {log.error("处理第三方数据时发生异常", e);return ResponseEntity.internalServerError().body("处理失败: " + e.getMessage());}}/*** 验证数据合法性** @param data 第三方数据DTO*/private void validateData(ThirdPartyDataDTO data) {Objects.requireNonNull(data, "数据不能为空");StringUtils.hasText(data.getDataId(), "dataId不能为空");StringUtils.hasText(data.getSourceSystem(), "sourceSystem不能为空");StringUtils.hasText(data.getDataType(), "dataType不能为空");StringUtils.hasText(data.getDataContent(), "dataContent不能为空");StringUtils.hasText(data.getSign(), "sign不能为空");Objects.requireNonNull(data.getTimestamp(), "timestamp不能为空");// 验证时间戳是否在有效范围内（5分钟内）long currentTime = System.currentTimeMillis();long timeDiff = Math.abs(currentTime - data.getTimestamp());if (timeDiff > 5 * 60 * 1000) {throw new IllegalArgumentException("时间戳已过期，可能存在重放攻击风险");}// 验证数据大小int dataSize = data.getDataContent().getBytes().length;if (dataSize > 10 * 1024 * 1024) { // 10MBthrow new IllegalArgumentException("数据大小超过限制，最大支持10MB");}}
}

3.8 消息消费者

package com.jamguo.sync.consumer;import com.fasterxml.jackson.databind.ObjectMapper;
import com.jamguo.sync.constant.SyncConstant;
import com.jamguo.sync.entity.BusinessData;
import com.jamguo.sync.entity.DataReceiveLog;
import com.jamguo.sync.service.BusinessDataService;
import com.jamguo.sync.service.DataReceiveLogService;
import com.rabbitmq.client.Channel;import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;import lombok.extern.slf4j.Slf4j;import javax.annotation.Resource;/*** 数据同步消费者* 负责消费MQ中的数据消息，并将其处理后保存到业务表** @author 果酱*/
@Component
@Slf4j
public class DataSyncConsumer {@Resourceprivate DataReceiveLogService dataReceiveLogService;@Resourceprivate BusinessDataService businessDataService;@Resourceprivate ObjectMapper objectMapper;@Value("${sync.max-retry-count}")private int maxRetryCount;/*** 消费数据同步消息** @param message 消息对象* @param channel 通道对象* @throws IOException IO异常*/@RabbitListener(queues = SyncConstant.RabbitMq.DATA_SYNC_QUEUE)public void consumeDataSyncMessage(Message message, Channel channel) throws IOException {long deliveryTag = message.getMessageProperties().getDeliveryTag();String messageId = message.getMessageProperties().getMessageId();try {log.info("收到数据同步消息，消息ID: {}", messageId);// 解析消息头中的接收日志IDString receiveLogIdStr = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.RECEIVE_LOG_ID);String dataId = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.DATA_ID);String retryCountStr = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.RETRY_COUNT);// 验证必要参数if (StringUtils.isBlank(receiveLogIdStr)) {log.error("消息头中未包含接收日志ID，消息ID: {}", messageId);// 拒绝消息，不重新入队channel.basicReject(deliveryTag, false);return;}Long receiveLogId;try {receiveLogId = Long.parseLong(receiveLogIdStr);} catch (NumberFormatException e) {log.error("接收日志ID格式错误: {}", receiveLogIdStr, e);channel.basicReject(deliveryTag, false);return;}// 获取当前重试次数int retryCount = 0;if (StringUtils.hasText(retryCountStr)) {try {retryCount = Integer.parseInt(retryCountStr);} catch (NumberFormatException e) {log.warn("重试次数格式错误: {}", retryCountStr, e);}}log.info("开始处理数据同步消息，消息ID: {}, receiveLogId: {}, dataId: {}, 重试次数: {}",messageId, receiveLogId, dataId, retryCount);// 查询接收日志DataReceiveLog receiveLog = dataReceiveLogService.getById(receiveLogId);if (receiveLog == null) {log.error("接收日志不存在，receiveLogId: {}, 消息ID: {}", receiveLogId, messageId);channel.basicReject(deliveryTag, false);return;}// 解析消息内容String dataContent = new String(message.getBody());log.debug("消息内容: {}", dataContent);// 处理业务逻辑String processResult = processBusinessData(receiveLog, dataContent);// 处理成功，更新状态dataReceiveLogService.handleMessageProcessSuccess(receiveLogId, processResult);// 手动确认消息channel.basicAck(deliveryTag, false);log.info("数据同步消息处理成功，消息ID: {}, receiveLogId: {}", messageId, receiveLogId);} catch (Exception e) {log.error("处理数据同步消息失败，消息ID: {}", messageId, e);// 解析接收日志ID，用于记录错误String receiveLogIdStr = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.RECEIVE_LOG_ID);if (StringUtils.hasText(receiveLogIdStr)) {try {Long receiveLogId = Long.parseLong(receiveLogIdStr);// 获取当前重试次数String retryCountStr = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.RETRY_COUNT);int retryCount = StringUtils.hasText(retryCountStr) ? Integer.parseInt(retryCountStr) : 0;retryCount++;// 更新处理失败状态dataReceiveLogService.handleMessageProcessFailed(receiveLogId, e.getMessage(), retryCount);// 判断是否超过最大重试次数if (retryCount >= maxRetryCount) {log.error("消息已达到最大重试次数，将被路由到死信队列，消息ID: {}, receiveLogId: {}",messageId, receiveLogId);// 拒绝消息，不重新入队，让其进入死信队列channel.basicReject(deliveryTag, false);} else {log.error("消息将被重新入队重试，消息ID: {}, receiveLogId: {}, 重试次数: {}",messageId, receiveLogId, retryCount);// 拒绝消息，重新入队channel.basicReject(deliveryTag, true);}} catch (NumberFormatException ex) {log.error("解析接收日志ID失败: {}", receiveLogIdStr, ex);// 无法解析接收日志ID，直接拒绝消息channel.basicReject(deliveryTag, false);}} else {// 没有接收日志ID，直接拒绝消息channel.basicReject(deliveryTag, false);}}}/*** 处理死信队列消息* 这些消息是经过多次重试后仍然处理失败的，需要人工干预** @param message 消息对象* @param channel 通道对象* @throws IOException IO异常*/@RabbitListener(queues = SyncConstant.RabbitMq.DEAD_LETTER_QUEUE)public void consumeDeadLetterMessage(Message message, Channel channel) throws IOException {long deliveryTag = message.getMessageProperties().getDeliveryTag();String messageId = message.getMessageProperties().getMessageId();try {log.error("收到死信队列消息，消息ID: {}", messageId);// 记录死信消息，触发告警String receiveLogIdStr = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.RECEIVE_LOG_ID);String dataId = message.getMessageProperties().getHeader(SyncConstant.MessageHeader.DATA_ID);log.error("死信消息详情: receiveLogId={}, dataId={}, 内容={}",receiveLogIdStr, dataId, new String(message.getBody()));// TODO: 发送告警通知，通知运维人员处理// 确认消息，从死信队列中移除channel.basicAck(deliveryTag, false);log.info("死信消息已确认处理，消息ID: {}", messageId);} catch (Exception e) {log.error("处理死信消息失败，消息ID: {}", messageId, e);// 确认消息，避免死信消息一直存在channel.basicAck(deliveryTag, false);}}/*** 处理业务数据** @param receiveLog 接收日志* @param dataContent 数据内容* @return 处理结果*/private String processBusinessData(DataReceiveLog receiveLog, String dataContent) {try {// 解析数据内容为JSON对象Map<String, Object> dataMap = objectMapper.readValue(dataContent, Map.class);// 根据数据类型进行不同的处理String dataType = receiveLog.getDataType();log.info("开始处理业务数据，dataId: {}, dataType: {}", receiveLog.getDataId(), dataType);// 创建业务数据实体BusinessData businessData = new BusinessData();businessData.setDataId(receiveLog.getDataId());businessData.setSourceSystem(receiveLog.getSourceSystem());businessData.setDataType(dataType);businessData.setDataContent(dataContent);businessData.setStatus(1); // 1-有效// 根据数据类型设置业务编号if (dataMap.containsKey("businessNo")) {businessData.setBusinessNo(dataMap.get("businessNo").toString());} else {// 生成业务编号businessData.setBusinessNo(generateBusinessNo(dataType));}// 保存业务数据businessDataService.save(businessData);log.info("业务数据处理成功，dataId: {}, businessNo: {}", receiveLog.getDataId(), businessData.getBusinessNo());return "处理成功，业务编号: " + businessData.getBusinessNo();} catch (Exception e) {log.error("处理业务数据失败，dataId: {}", receiveLog.getDataId(), e);throw new RuntimeException("处理业务数据失败: " + e.getMessage(), e);}}/*** 生成业务编号** @param dataType 数据类型* @return 业务编号*/private String generateBusinessNo(String dataType) {// 生成规则：数据类型前缀 + 时间戳 + 随机数return String.format("%s_%d_%s",dataType.toUpperCase(),System.currentTimeMillis(),UUID.randomUUID().toString().substring(0, 8));}
}

3.9 定时任务：确保数据最终一致性

package com.jamguo.sync.task;import com.jamguo.sync.service.DataReceiveLogService;
import com.jamguo.sync.service.DataSyncExceptionService;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;import lombok.extern.slf4j.Slf4j;/*** 数据同步定时任务* 负责检查和处理异常数据，确保数据最终一致性** @author 果酱*/
@Component
@Slf4j
public class DataSyncScheduleTask {@Autowiredprivate DataReceiveLogService dataReceiveLogService;@Autowiredprivate DataSyncExceptionService dataSyncExceptionService;@Value("${sync.schedule-interval}")private int scheduleInterval;/*** 定时重发失败的消息* 每5分钟执行一次*/@Scheduled(fixedRateString = "${sync.schedule-interval:5} * 60 * 1000")public void resendFailedMessages() {log.info("开始执行定时重发失败消息任务，间隔: {}分钟", scheduleInterval);try {// 每次处理100条int successCount = dataReceiveLogService.resendFailedMessages(100);log.info("定时重发失败消息任务执行完成，成功重发 {} 条消息", successCount);// 如果还有失败的消息，记录告警if (successCount > 0) {// TODO: 可以根据实际情况设置阈值，超过阈值则发送告警log.warn("仍有 {} 条消息重发成功，需要关注", successCount);}} catch (Exception e) {log.error("执行定时重发失败消息任务时发生异常", e);}}/*** 检查长时间未处理的数据* 每30分钟执行一次*/@Scheduled(cron = "0 0/30 * * * ?")public void checkUnprocessedData() {log.info("开始执行检查长时间未处理的数据任务");try {// 查询2小时内未处理成功的数据int unprocessedCount = dataReceiveLogService.countUnprocessedData(2);log.info("检查长时间未处理的数据任务执行完成，共发现 {} 条未处理数据", unprocessedCount);// 如果存在未处理的数据，发送告警if (unprocessedCount > 0) {log.error("发现 {} 条长时间未处理的数据，需要人工干预", unprocessedCount);// TODO: 发送告警通知}} catch (Exception e) {log.error("执行检查长时间未处理的数据任务时发生异常", e);}}/*** 清理过期数据* 每天凌晨2点执行*/@Scheduled(cron = "0 0 2 * * ?")public void cleanExpiredData() {log.info("开始执行清理过期数据任务");try {// 清理30天前的已处理成功的数据int deleteCount = dataReceiveLogService.cleanExpiredData(30);log.info("清理过期数据任务执行完成，共清理 {} 条数据", deleteCount);} catch (Exception e) {log.error("执行清理过期数据任务时发生异常", e);}}
}

四、可能出现的数据丢失场景及解决方案

4.1 场景一：第三方系统推送数据时网络中断

问题描述：第三方系统推送数据时，由于网络原因，数据未能到达我方系统。

解决方案：

  1. 实现请求确认机制，接收方收到数据后立即返回确认响应
  2. 第三方系统实现重试机制，未收到确认时进行有限次数重试
  3. 定期全量同步，弥补增量同步可能出现的遗漏

代码示例：第三方系统重试逻辑

/*** 发送数据到接收方，并处理重试*/
public boolean sendDataWithRetry(ThirdPartyDataDTO data, int maxRetries) {int retryCount = 0;while (retryCount <= maxRetries) {try {// 发送数据ResponseEntity<String> response = restTemplate.postForEntity(RECEIVE_URL, data, String.class);// 检查响应状态if (response.getStatusCode().is2xxSuccessful()) {log.info("数据发送成功，dataId: {}", data.getDataId());return true;}log.error("数据发送失败，响应状态: {}", response.getStatusCodeValue());} catch (Exception e) {log.error("数据发送异常，第 {} 次重试", retryCount + 1, e);}retryCount++;// 重试间隔，指数退避策略if (retryCount <= maxRetries) {long sleepTime = (long) (1000 * Math.pow(2, retryCount));try {Thread.sleep(sleepTime);} catch (InterruptedException ie) {Thread.currentThread().interrupt();return false;}}}log.error("数据发送失败，已达到最大重试次数: {}", maxRetries);// 记录到本地失败队列，等待人工处理saveFailedData(data, "达到最大重试次数");return false;
}

4.2 场景二：数据接收接口处理异常

问题描述：数据成功到达接收接口，但在处理过程中发生异常（如参数校验失败、系统错误等），导致数据未被正确保存。

解决方案：

  1. 实现接口级别的异常捕获，确保任何异常都能被捕获并处理
  2. 对于参数错误等可明确处理的异常，返回具体错误信息，便于第三方系统修正
  3. 对于系统异常，返回服务器错误，并将数据临时存储，后续手动处理
  4. 使用请求 ID 记录每一次请求的处理日志，便于问题排查

代码示例：全局异常处理器

@RestControllerAdvice
@Slf4j
public class GlobalExceptionHandler {@Autowiredprivate DataSyncExceptionService exceptionService;/*** 处理参数验证异常*/@ExceptionHandler(IllegalArgumentException.class)public ResponseEntity<String> handleIllegalArgumentException(IllegalArgumentException e) {log.warn("参数验证失败: {}", e.getMessage());return ResponseEntity.badRequest().body("参数错误: " + e.getMessage());}/*** 处理业务异常*/@ExceptionHandler(BusinessException.class)public ResponseEntity<String> handleBusinessException(BusinessException e) {log.error("业务处理异常: {}", e.getMessage());return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("业务处理失败: " + e.getMessage());}/*** 处理系统异常*/@ExceptionHandler(Exception.class)public ResponseEntity<String> handleException(Exception e, HttpServletRequest request) {log.error("系统异常", e);// 尝试提取请求数据并记录try {String requestBody = extractRequestBody(request);String requestId = request.getHeader("X-Request-ID");DataSyncException exception = new DataSyncException();exception.setDataId(requestId);exception.setErrorType("SYSTEM_EXCEPTION");exception.setErrorMessage(e.getMessage());exception.setExceptionStack(ThrowableUtils.getStackTrace(e));exception.setHandleStatus(SyncConstant.ExceptionHandleStatus.UNHANDLED);exceptionService.save(exception);log.info("系统异常信息已记录，requestId: {}", requestId);} catch (Exception ex) {log.error("记录系统异常信息失败", ex);}return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("服务器处理失败，请稍后重试");}/*** 提取请求体*/private String extractRequestBody(HttpServletRequest request) {try {return StreamUtils.copyToString(request.getInputStream(), StandardCharsets.UTF_8);} catch (IOException e) {log.error("提取请求体失败", e);return "";}}
}

4.3 场景三：数据保存到接收日志表失败

问题描述：数据接收接口处理正常，但在将数据保存到接收日志表时失败（如数据库连接异常、表空间满等）。

解决方案：

  1. 使用本地消息表模式，确保数据先落地到数据库
  2. 实现数据库操作的重试机制，处理临时的数据库连接问题
  3. 当数据库完全不可用时，将数据写入本地文件系统作为备份
  4. 监控数据库状态，及时处理数据库异常

代码示例：数据库操作重试机制

/*** 带重试机制的数据库操作模板*/
@Component
public class RetryableDbTemplate {private static final int DEFAULT_MAX_RETRIES = 3;private static final long DEFAULT_INITIAL_DELAY = 1000; // 1秒private static final double DEFAULT_MULTIPLIER = 2.0;/*** 执行带重试机制的数据库操作*/public <T> T execute(Supplier<T> dbOperation) {return execute(dbOperation, DEFAULT_MAX_RETRIES, DEFAULT_INITIAL_DELAY, DEFAULT_MULTIPLIER);}/*** 执行带重试机制的数据库操作*/public <T> T execute(Supplier<T> dbOperation, int maxRetries, long initialDelay, double multiplier) {int retryCount = 0;while (true) {try {return dbOperation.get();} catch (Exception e) {// 判断是否是可重试的异常if (!isRetryableException(e) || retryCount >= maxRetries) {log.error("数据库操作失败，已达到最大重试次数: {}", maxRetries, e);throw e;}retryCount++;long delay = (long) (initialDelay * Math.pow(multiplier, retryCount - 1));log.warn("数据库操作失败，将在 {}ms 后进行第 {} 次重试", delay, retryCount, e);try {Thread.sleep(delay);} catch (InterruptedException ie) {Thread.currentThread().interrupt();log.error("重试等待被中断", ie);throw e;}}}}/*** 判断异常是否可重试*/private boolean isRetryableException(Exception e) {// 可重试的异常包括：数据库连接异常、超时异常等return e instanceof SQLException || e instanceof TransientDataAccessException|| e.getMessage() != null && (e.getMessage().contains("timeout") || e.getMessage().contains("connection"));}
}

4.4 场景四：消息发送到 RabbitMQ 失败

问题描述：数据成功保存到接收日志表，但在发送到 RabbitMQ 时失败（如 MQ 连接异常、MQ 服务不可用等）。

解决方案：

  1. 实现消息发送的重试机制
  2. 使用 RabbitMQ 的发布确认机制（Publisher Confirm），确保消息成功到达 MQ
  3. 对于发送失败的消息，记录状态，由定时任务进行重试
  4. 部署 RabbitMQ 集群，提高 MQ 服务的可用性

代码示例：已在 3.6 节的 sendToMq 方法和 3.9 节的定时任务中实现

4.5 场景五：消息到达 RabbitMQ 但未被路由到队列

问题描述：消息成功发送到 RabbitMQ 交换机，但由于路由键错误、绑定关系不存在等原因，未能路由到目标队列，导致消息丢失。

解决方案：

  1. 使用 RabbitMQ 的发布返回机制（Publisher Return），捕获未被路由的消息
  2. 为交换机和队列设置合理的名称和路由键命名规范，避免人为错误
  3. 对未被路由的消息进行记录，并由定时任务重新发送
  4. 在 MQ 管理界面监控消息路由情况，及时发现问题

代码示例：已在 3.3 节的 RabbitTemplate 配置中实现

4.6 场景六：RabbitMQ 服务崩溃导致消息丢失

问题描述：消息成功路由到队列，但在消费者处理之前，RabbitMQ 服务崩溃，导致消息丢失。

解决方案：

  1. 配置交换机、队列和消息的持久化
  2. 部署 RabbitMQ 集群，并配置镜像队列（Mirror Queue）
  3. 定期备份 RabbitMQ 的数据
  4. 监控 RabbitMQ 服务状态，及时发现和处理服务异常

配置示例：已在 3.3 节的队列和交换机配置中实现持久化和镜像队列相关参数

4.7 场景七：消费者接收消息后处理失败

问题描述：消费者成功接收消息，但在处理过程中发生异常，导致数据未能正确保存到业务表。

解决方案：

  1. 实现消息消费的重试机制
  2. 使用死信队列（Dead Letter Queue）存储多次处理失败的消息
  3. 对处理失败的消息进行记录，便于后续分析和处理
  4. 实现幂等性处理，确保消息重试不会导致数据不一致

代码示例：已在 3.8 节的消费者代码中实现重试和死信队列机制

4.8 场景八：消费者处理成功但确认消息失败

问题描述：消费者成功处理消息，但在发送确认消息（Ack）时发生异常，导致 RabbitMQ 认为消息未被处理，从而将消息重新发送给其他消费者，造成重复处理。

解决方案：

  1. 确保消息处理和状态更新在同一个事务中
  2. 实现业务处理的幂等性，即使消息被重复处理也不会导致数据不一致
  3. 优先更新业务状态，再发送确认消息
  4. 对于关键业务，可采用 "处理日志 + 状态机" 的方式确保处理状态的准确性

代码示例：已在 3.8 节的消费者代码中实现，先更新数据库状态，再发送 Ack

4.9 场景九：服务器崩溃导致消息处理中断

问题描述：消费者正在处理消息时，服务器突然崩溃，导致消息处理中断，且未发送确认消息。

解决方案：

  1. 确保消息处理是幂等的，服务器恢复后可以重新处理
  2. 合理设置消息处理超时时间，超时未确认的消息将被重新发送
  3. 对于长耗时的任务，拆分为多个短任务，分步处理和确认
  4. 实现断点续传机制，服务器恢复后可以从断点继续处理

配置示例：在 application.yml 中设置消息处理超时时间

spring:rabbitmq:listener:simple:default-requeue-rejected: false# 设置消息处理超时时间（毫秒）acknowledge-timeout: 30000

4.10 场景十：数据成功保存但后续业务处理失败

问题描述：数据成功保存到业务表，但后续的业务处理（如通知其他系统、更新关联数据等）失败。

解决方案：

  1. 使用分布式事务或最终一致性方案（如 Saga 模式）
  2. 将后续业务处理也通过消息队列异步处理，并确保这些消息的可靠性
  3. 实现业务状态机，清晰记录每个处理步骤的状态
  4. 定时检查业务处理状态，对未完成的业务进行补偿处理

代码示例：业务状态机实现

/*** 业务状态机服务*/
@Service
@Slf4j
public class BusinessStateMachineService {@Autowiredprivate BusinessDataService businessDataService;@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 更新业务状态并触发下一步处理*/@Transactional(rollbackFor = Exception.class)public void updateStateAndProceed(Long businessId, String currentState, String nextState, String event) {Objects.requireNonNull(businessId, "业务ID不能为空");StringUtils.hasText(currentState, "当前状态不能为空");StringUtils.hasText(nextState, "下一状态不能为空");log.info("更新业务状态，ID: {}, 从 {} 到 {}, 事件: {}", businessId, currentState, nextState, event);// 查询业务数据BusinessData businessData = businessDataService.getById(businessId);Objects.requireNonNull(businessData, "业务数据不存在，ID: " + businessId);// 验证当前状态是否匹配if (!currentState.equals(businessData.getStatus())) {throw new BusinessException(String.format("状态不匹配，当前状态: %s, 期望状态: %s", businessData.getStatus(), currentState));}// 更新状态businessData.setStatus(nextState);businessDataService.updateById(businessData);log.info("业务状态更新成功，ID: {}, 新状态: {}", businessId, nextState);// 根据下一状态和事件，触发后续处理triggerNextProcess(businessData, nextState, event);}/*** 触发下一步处理*/private void triggerNextProcess(BusinessData businessData, String nextState, String event) {// 根据状态和事件，发送不同的消息String routingKey = String.format("business.%s.%s", businessData.getDataType(), nextState);BusinessEventMessage eventMessage = new BusinessEventMessage();eventMessage.setBusinessId(businessData.getId());eventMessage.setBusinessNo(businessData.getBusinessNo());eventMessage.setDataType(businessData.getDataType());eventMessage.setCurrentState(nextState);eventMessage.setEvent(event);eventMessage.setTimestamp(System.currentTimeMillis());log.info("触发业务后续处理，ID: {}, 路由键: {}", businessData.getId(), routingKey);// 发送消息到MQ，确保消息可靠传递rabbitTemplate.convertAndSend("business.event.exchange",routingKey,eventMessage,new CorrelationData(UUID.randomUUID().toString()));}
}

4.11 场景十一：数据备份和恢复机制缺失

问题描述：系统发生灾难性故障（如数据库崩溃），由于缺乏有效的备份和恢复机制，导致数据永久丢失。

解决方案：

  1. 定期备份数据库，包括全量备份和增量备份
  2. 实现跨地域的数据备份，避免单点故障
  3. 制定详细的数据恢复预案，并定期演练
  4. 使用数据库主从复制，提高数据可用性

MySQL 备份脚本示例：

#!/bin/bash
# 数据库备份脚本# 配置
DB_NAME="data_sync_db"
BACKUP_DIR="/backup/mysql"
DATE=$(date +%Y%m%d_%H%M%S)
RETENTION_DAYS=7# 创建备份目录
mkdir -p $BACKUP_DIR# 全量备份
echo "开始数据库全量备份: $DB_NAME"
mysqldump -u root -p'password' --databases $DB_NAME --single-transaction --routines --triggers > $BACKUP_DIR/${DB_NAME}_full_$DATE.sql# 压缩备份文件
gzip $BACKUP_DIR/${DB_NAME}_full_$DATE.sql
echo "数据库全量备份完成: $BACKUP_DIR/${DB_NAME}_full_$DATE.sql.gz"# 删除过期备份
echo "清理 $RETENTION_DAYS 天前的备份文件"
find $BACKUP_DIR -name "${DB_NAME}_full_*.sql.gz" -type f -mtime +$RETENTION_DAYS -delete# 备份到远程服务器
echo "同步备份到远程服务器"
rsync -avz $BACKUP_DIR/ user@remote-server:/backup/mysql/

4.12 场景十二：监控和告警机制缺失

问题描述：系统中已经发生数据丢失，但由于缺乏有效的监控和告警机制，未能及时发现，导致问题扩大。

解决方案：

  1. 实现全链路监控，包括接口调用、消息发送和接收、数据库操作等
  2. 设置关键指标的告警阈值，如消息堆积数、处理失败数等
  3. 实现多级告警机制，确保问题能够及时送达相关人员
  4. 定期检查和分析监控数据，发现潜在问题

监控指标示例：

/*** 数据同步监控服务*/
@Component
@Slf4j
public class DataSyncMonitorService {@Autowiredprivate DataReceiveLogService receiveLogService;@Autowiredprivate DataSyncExceptionService exceptionService;@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 收集监控指标*/public DataSyncMonitorDTO collectMonitorMetrics() {DataSyncMonitorDTO metrics = new DataSyncMonitorDTO();metrics.setCollectTime(LocalDateTime.now());// 统计接收数据量Map<String, Long> receiveCountByType = receiveLogService.countByDataTypeAndTimeRange(LocalDateTime.now().minusHours(24), LocalDateTime.now());metrics.setReceiveCountByType(receiveCountByType);metrics.setTotalReceiveCount(receiveCountByType.values().stream().mapToLong(v -> v).sum());// 统计处理成功和失败数量Map<Integer, Long> countByStatus = receiveLogService.countByStatus();metrics.setSuccessCount(countByStatus.getOrDefault(SyncConstant.ReceiveStatus.PROCESSED_SUCCESS, 0L));metrics.setFailedCount(countByStatus.getOrDefault(SyncConstant.ReceiveStatus.PROCESSED_FAILED, 0L));metrics.setPendingCount(countByStatus.getOrDefault(SyncConstant.ReceiveStatus.SENT_TO_MQ, 0L) +countByStatus.getOrDefault(SyncConstant.ReceiveStatus.INIT, 0L));// 统计异常数量long unhandledExceptionCount = exceptionService.countByHandleStatus(SyncConstant.ExceptionHandleStatus.UNHANDLED);metrics.setUnhandledExceptionCount(unhandledExceptionCount);// 获取MQ队列信息try {RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());QueueInformation queueInfo = rabbitAdmin.getQueueInfo(SyncConstant.RabbitMq.DATA_SYNC_QUEUE);if (queueInfo != null) {metrics.setQueueMessageCount(queueInfo.getMessageCount());metrics.setConsumerCount(queueInfo.getConsumerCount());}QueueInformation dlqInfo = rabbitAdmin.getQueueInfo(SyncConstant.RabbitMq.DEAD_LETTER_QUEUE);if (dlqInfo != null) {metrics.setDeadLetterCount(dlqInfo.getMessageCount());}} catch (Exception e) {log.error("获取MQ队列信息失败", e);}return metrics;}/*** 检查监控指标是否超过阈值*/public List<String> checkThresholds(DataSyncMonitorDTO metrics) {List<String> alerts = new ArrayList<>();// 检查死信队列数量if (metrics.getDeadLetterCount() > 100) {alerts.add(String.format("死信队列消息过多: %d条，超过阈值100条", metrics.getDeadLetterCount()));}// 检查未处理异常数量if (metrics.getUnhandledExceptionCount() > 50) {alerts.add(String.format("未处理异常过多: %d个，超过阈值50个", metrics.getUnhandledExceptionCount()));}// 检查队列消息堆积if (metrics.getQueueMessageCount() > 1000) {alerts.add(String.format("消息队列堆积严重: %d条，超过阈值1000条", metrics.getQueueMessageCount()));}// 检查失败率long totalProcessed = metrics.getSuccessCount() + metrics.getFailedCount();if (totalProcessed > 0) {double failureRate = (double) metrics.getFailedCount() / totalProcessed;if (failureRate > 0.1) { // 失败率超过10%alerts.add(String.format("消息处理失败率过高: %.2f%%，超过阈值10%%", failureRate * 100));}}return alerts;}
}

五、总结与最佳实践

通过本文的详细讲解和代码示例，我们构建了一个能够保证第三方接口数据不丢失的完整解决方案。这个方案基于 RabbitMQ 实现，通过多重保障机制确保数据从接收、传递到最终处理的全链路可靠性。

最佳实践总结

  1. 数据接收阶段：

    • 实现严格的参数校验和签名验证
    • 采用 "先落地，后处理" 的原则，确保数据首先保存到本地数据库
    • 实现接口级别的异常处理，避免因局部错误导致整个请求失败

  2. 消息传递阶段：

    • 配置 RabbitMQ 的持久化机制，包括交换机、队列和消息
    • 启用发布确认和返回机制，确保消息可靠传递
    • 实现消息发送重试机制，处理临时的网络或服务异常

  3. 消息处理阶段：

    • 使用手动确认模式，确保消息被正确处理后再确认
    • 实现消费重试机制和死信队列，处理无法正常处理的消息
    • 确保业务处理的幂等性，避免重复处理导致的数据不一致

  4. 监控和运维阶段：

    • 实现全链路监控，及时发现和解决问题
    • 建立完善的告警机制，确保异常情况能够及时通知
    • 定期备份数据，制定灾难恢复预案
    • 实现定时任务，对异常数据进行补偿处理

  5. 架构设计阶段：

    • 采用集群部署，提高系统可用性
    • 实现最终一致性方案，确保分布式环境下的数据一致性
    • 设计合理的状态机，清晰记录业务处理状态

通过遵循这些最佳实践，我们可以构建一个真正可靠的数据同步系统，确保第三方接口数据不丢失，为业务决策提供准确的数据支持。

    参考资料

    1. RabbitMQ 官方文档 - 消息确认机制：Consumer Acknowledgements and Publisher Confirms | RabbitMQ
    2. Spring AMQP 官方文档 - 消息可靠性：https://docs.spring.io/spring-amqp/docs/current/reference/html/#reliability
    http://www.xdnf.cn/news/1399393.html

    相关文章：

  • comfUI背后的技术——VAE
  • 线性代数理论——状态空间
  • 聊一聊耳机串扰-Crosstalk
  • Vue常用指令和生命周期
  • 118、【OS】【Nuttx】【周边】效果呈现方案解析：作用域？
  • 生成一份关于电脑电池使用情况、健康状况和寿命估算的详细 HTML 报告
  • 软考中级习题与解答——第一章_数据结构与算法基础（2）
  • 【Redisson 加锁源码解析】
  • VuePress添加自定义组件
  • 【MySQL数据库】索引 - 结构 学习记录
  • 加速智能经济发展：如何助力“人工智能+”战略在实时视频领域的落地
  • Swift 解法详解：LeetCode 367《有效的完全平方数》
  • Kafka入门
  • 开源 C++ QT Widget 开发（八）网络--Http文件下载
  • 《微服务架构从故障频发到自愈可控的实战突围方案》
  • CSDN博客语法（不常用但有用）
  • 谷歌 “Nano Banana“ 深度解析：AI 图像的未来是精准编辑，而非从零生成
  • ⚡ Linux find 命令参数详解
  • MySQL基础理解入门
  • 嵌入式硬件电路分析---AD采集电路
  • Spring Boot 自动配置原理深度解析：从启动流程到监听机制
  • 【Java EE进阶 --- SpringBoot】Spring Web MVC(Spring MVC)（二）
  • 设计模式之代理模式！
  • 深度学习基础：前馈网络、反向传播与梯度问题
  • 基于IEC61499开放自动化PLC数据储存方案
  • 在 WSL2-NVIDIA-Workbench 中安装Anaconda、CUDA 13.0、cuDNN 9.12 及 PyTorch（含完整环境验证）
  • 第 8 篇：量化交易之tradeUI和webserverUI 区别?
  • 系统分析师考试大纲新旧版本深度分析与备考策略
  • 捡捡java——2、基础07
  • 开发指南136-设置零值不显示