阅读全文

⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7575182522411991078 「我怎么想不到」欢迎转载，保留摘要，谢谢！

• 一、HTTP客户端方案
  • 方案一：RestTemplate（同步基础款，适合简单场景）
  • 方案二：Feign（声明式调用，适配微服务）
  • 方案三：WebClient（响应式非阻塞，适配高并发）
• 二、数据同步方案
  • 方案一：定时全量同步
  • 方案二：定时增量同步
  • 方案三：实时同步 （Webhook）

---

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
4. 新的源码解析文章实时收到通知。每周更新一篇左右。
5. 认真的源码交流微信群。

---

根据实际场景需求去选择需要的解决方案。

HTTP客户端选择方案：RestTemplate、Feign、WebClient。

同步方案：全量同步、增量同步、实时同步 三种核心方案。

一、HTTP客户端方案

Spring Boot 对接第三方接口有多种常用方案，适配不同场景，比如简单场景用RestTemplate，微服务架构用Feign，高并发场景用响应式的WebClient。以下是每种方案的详细教程，包含依赖配置、代码实现和核心说明。

Spring Boot 官方在文档中推荐使用 RestTemplate（传统项目）或 WebClient（响应式项目），而 Feign 作为 Spring Cloud 的一部分，也是微服务场景的首选。

方案一：RestTemplate（同步基础款，适合简单场景）

RestTemplate是 Spring 框架提供的同步 HTTP 客户端，适配大多数简单的第三方接口调用场景，Spring Boot 2.x 中可直接集成使用。

步骤1： 添加依赖Spring Boot 2.x 的spring-boot-starter-web已内置RestTemplate，在pom.xml中添加 web 依赖即可：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

步骤2： 配置 RestTemplate Bean 创建配置类，将RestTemplate注入 Spring 容器，可配置超时时间等参数：

@Configuration
public class RestTemplateConfig {
    @Bean
    public RestTemplate restTemplate() {
        HttpComponentsClientHttpRequestFactory factory = new HttpComponentsClientHttpRequestFactory();
        factory.setConnectTimeout(5000); // 连接超时5秒
        factory.setReadTimeout(5000);    // 读取超时5秒
        return new RestTemplate(factory);
    }
}

步骤3： 调用第三方接口在 Service 层注入RestTemplate，分别实现 GET 和 POST 请求调用。这里以调用模拟的用户接口为例：

@Service
publicclass ThirdPartyService {
    @Resource
    private RestTemplate restTemplate;

    // GET请求：根据ID查询用户
    public UserDTO getUserById(Long userId) {
        String url = "https://api.example.com/users/{id}";
        // 占位符替换，返回结果自动转为UserDTO
        return restTemplate.getForObject(url, UserDTO.class, userId);
    }

    // POST请求：创建用户
    public UserDTO createUser(UserRequest request) {
        String url = "https://api.example.com/users";
        // 发送POST请求，携带JSON请求体，返回UserDTO
        return restTemplate.postForObject(url, request, UserDTO.class);
    }
}

步骤4： 定义实体类创建与接口请求 / 响应对应的实体类UserRequest和UserDTO：

// 请求实体
publicclass UserRequest {
    private String username;
    private String email;
    // getter和setter
}

// 响应实体
publicclass UserDTO {
    private Long id;
    private String username;
    private String email;
    // getter和setter
}

方案二：Feign（声明式调用，适配微服务）

Feign 是声明式 HTTP 客户端，通过注解简化请求代码，且能与 Spring Cloud 集成实现负载均衡，适合微服务架构下的第三方接口调用。

步骤1： 添加依赖在pom.xml中添加 OpenFeign 依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-openfeign</artifactId>
        <version>3.1.3</version>
    </dependency>
</dependencies>

步骤2： 启用 Feign 客户端在 Spring Boot 启动类添加@EnableFeignClients注解：

@SpringBootApplication
@EnableFeignClients // 启用Feign客户端
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

步骤3： 定义 Feign 接口创建 Feign 接口，通过注解声明第三方接口的请求规则：

// name为客户端名称，url为第三方接口基地址
@FeignClient(name = "user-api", url = "https://api.example.com")
public interface UserFeignClient {

    @GetMapping("/users/{id}")
    UserDTO getUserById(@PathVariable("id") Long userId);

    @PostMapping("/users")
    UserDTO createUser(@RequestBody UserRequest request);
}

步骤4： 调用 Feign 接口在 Service 层注入 Feign 接口直接调用，无需手动构建请求：

@Service
public class UserService {
    @Resource
    private UserFeignClient userFeignClient;

    public UserDTO getUser(Long userId) {
        return userFeignClient.getUserById(userId);
    }

    public UserDTO addUser(UserRequest request) {
        return userFeignClient.createUser(request);
    }
}

方案三：WebClient（响应式非阻塞，适配高并发）

WebClient是 Spring WebFlux 提供的响应式 HTTP 客户端，非阻塞 IO，适合高并发场景，Spring Boot 2.x 及以上版本支持。

步骤1： 添加依赖在pom.xml中添加 WebFlux 依赖（内置 WebClient）：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
</dependencies>

步骤2： 配置 WebClient Bean创建配置类，统一配置基础 URL 和请求头：

@Configuration
public class WebClientConfig {
    @Bean
    public WebClient webClient() {
        return WebClient.builder()
                .baseUrl("https://api.example.com") // 第三方接口基地址
                .defaultHeader("Content-Type", "application/json")
                .build();
    }
}

步骤3： 调用第三方接口WebClient返回Mono（单结果）或Flux（多结果），通过响应式编程处理结果：

@Service
publicclass ReactiveThirdPartyService {
    @Resource
    private WebClient webClient;

    // GET请求：查询用户
    public Mono<UserDTO> getUserById(Long userId) {
        return webClient.get()
                .uri("/users/{id}", userId)
                .retrieve() // 发送请求并接收响应
                .bodyToMono(UserDTO.class); // 响应体转为UserDTO的Mono对象
    }

    // POST请求：创建用户
    public Mono<UserDTO> createUser(UserRequest request) {
        return webClient.post()
                .uri("/users")
                .bodyValue(request) // 设置请求体
                .retrieve()
                .bodyToMono(UserDTO.class);
    }
}

步骤4： 控制器层调用响应式接口需返回Mono或Flux对象：

@RestController
@RequestMapping("/api/users")
publicclass UserController {
    @Resource
    private ReactiveThirdPartyService service;

    @GetMapping("/{id}")
    public Mono<UserDTO> getUser(@PathVariable Long id) {
        return service.getUserById(id);
    }

    @PostMapping
    public Mono<UserDTO> addUser(@RequestBody UserRequest request) {
        return service.createUser(request);
    }
}

二、数据同步方案

方案一：定时全量同步

适用于数据量小、对实时性要求不高的场景。

实现思路：

每天凌晨2点执行一次全量拉取 删除旧数据，插入新数据（或软删除 + 更新） 使用事务保证一致性

1、全量删除 + 批量插入

@Slf4j
@Component
@RequiredArgsConstructor
publicclass FullSyncScheduler {

    privatefinal RestTemplate restTemplate = new RestTemplate();
    privatefinal DepartmentService departmentService;
    privatefinal UserService userService;

    @Value("${third-party.api-base-url}")
    private String apiBaseUrl;

    // 每天凌晨2点执行
    @Scheduled(cron = "0 0 2 * * *")
    public void performFullSync() {
        log.info("--- 开始执行全量同步 ---");
        Instant startTime = Instant.now();
        try {
            // 步骤 1: 删除本地所有数据
            departmentService.remove(new QueryWrapper<>()); 
            UserService.remove(new QueryWrapper<>());
            
            // 步骤 2: 从第三方拉取全量数据
            syncDepartments(); // 1. 同步部门
            syncUsers(); // 2. 同步用户

            log.info("--- 全量同步成功完成，总耗时: {} ms ---",
                    Duration.between(startTime, Instant.now()).toMillis());
        } catch (Exception e) {
            log.error("全量同步失败", e);
        }
    }

    // 同步部门逻辑
    private void syncDepartments() {
        log.info("同步部门数据...");
        Instant depStartTime = Instant.now();
        // 通过第三方接口获取数据
        String url = apiBaseUrl + "/api/departments";
        Department[] remoteDepartments = restTemplate.getForObject(url, Department[].class);
        if (remoteDepartments == null || remoteDepartments.length == 0) {
            log.warn("从第三方API获取部门数据为空");
            return;
        }
        List<Department> deptList = Arrays.asList(remoteDepartments);

        // 批量插入到本地数据库
        departmentService.saveBatch(deptList);
        
        log.info("部门同步完成，共 {} 个部门，耗时：{}", 
            remoteDepartments.length,Duration.between(depStartTime, Instant.now()).toMillis());
    }

    // 同步用户逻辑
    private void syncUsers() {
        log.info("同步用户数据...");
        // 通过第三方接口获取数据
        String url = apiBaseUrl + "/api/users";
        User[] remoteUsers = restTemplate.getForObject(url, User[].class);
        if (remoteUsers == null || remoteUsers.length == 0) {
            log.warn("从第三方API获取用户数据为空");
            return;
        }
        List<User> userList = Arrays.asList(remoteUsers);
        
        // 批量插入到本地数据库
        userService.saveBatch(userList);

        log.info("用户同步完成，共 {} 个用户。", remoteUsers.length);
    }
}

2、UPSERT + 删除多余 (SaveOrUpdateBatch + Delete Not In)

@Slf4j
@Component
@RequiredArgsConstructor
publicclass FullSyncScheduler {

    privatefinal RestTemplate restTemplate = new RestTemplate();
    privatefinal DepartmentService departmentService;
    privatefinal UserService userService;

    @Value("${third-party.api-base-url}")
    private String apiBaseUrl;

    // 每天凌晨2点执行
    @Scheduled(cron = "0 0 2 * * *")
    public void performFullSync() {
        log.info("--- 开始执行全量同步 ---");
        Instant startTime = Instant.now();
        try {
            // 1. 同步部门
            syncDepartments();
            // 2. 同步用户
            syncUsers();

            log.info("--- 全量同步成功完成，总耗时: {} ms ---",
                    Duration.between(startTime, Instant.now()).toMillis());
        } catch (Exception e) {
            log.error("全量同步失败", e);
        }
    }

    // 同步部门逻辑
    private void syncDepartments() {
        log.info("同步部门数据...");
        Instant depStartTime = Instant.now();
        // 步骤 1: 从第三方拉取全量数据
        String url = apiBaseUrl + "/api/departments";
        Department[] remoteDepartments = restTemplate.getForObject(url, Department[].class);
        if (remoteDepartments == null || remoteDepartments.length == 0) {
            log.warn("从第三方API获取部门数据为空");
            return;
        }

        List<Department> deptList = Arrays.asList(remoteDepartments);
        List<String> remoteIds = deptList.stream()
                .map(Department::getExternalId)
                .collect(Collectors.toList());

        // 步骤 2: 执行 UPSERT (更新或插入)
        departmentService.saveOrUpdateBatch(deptList);
        
        // 收集 externalId
        List<String> remoteIds = deptList.stream()
                .map(Department::getExternalId)
                .collect(Collectors.toList());
        // 步骤3：找出并删除本地存在但远程不存在的数据
        departmentService.removeByExternalIdNotIn(remoteIds);

        log.info("部门同步完成，共 {} 个部门，耗时：{}", 
            remoteDepartments.length,Duration.between(depStartTime, Instant.now()).toMillis());
    }

    // 同步用户逻辑
    private void syncUsers() {
        log.info("同步用户数据...");
        // 步骤 1: 从第三方拉取全量数据
        String url = apiBaseUrl + "/api/users";
        User[] remoteUsers = restTemplate.getForObject(url, User[].class);
        if (remoteUsers == null || remoteUsers.length == 0) {
            log.warn("从第三方API获取用户数据为空");
            return;
        }

        List<User> userList = Arrays.asList(remoteUsers);
        
        // 步骤 2: 执行 UPSERT (更新或插入)
        userService.saveOrUpdateBatch(userList);
        
        // 收集 externalId
        List<String> remoteIds = userList.stream()
                .map(User::getExternalId)
                .collect(Collectors.toList());
        // 步骤3：找出并删除本地存在但远程不存在的数据
        userService.removeByExternalIdNotIn(remoteIds);

        log.info("用户同步完成，共 {} 个用户。", remoteUsers.length);
    }
}

几乎在所有其他情况下，方案二都是更优、更安全的选择。

它能最大限度地保证数据的一致性和业务的连续性，虽然在性能上可能比方案一略逊一筹，但在绝大多数企业级应用中，数据一致性和系统稳定性远比同步快几秒更为重要。

因此，在组织架构同步场景中，强烈推荐使用方案二（saveOrUpdateBatch + delete not in）。它能确保在同步过程中，业务系统总能查询到有效的部门和用户信息，避免了因同步失败或数据真空期导致的业务异常。

方案二：定时增量同步

1、基于时间戳的增量同步（最常用）记录上次同步的时间戳（如 last_sync_time），每次同步时只拉取第三方系统中 update_time > last_sync_time 的数据。

步骤1： 记录同步时间戳

在本地数据库中维护一张同步记录表（如 sync_checkpoint），存储每个同步任务的上次成功时间戳。

CREATE TABLE sync_checkpoint (
    idINT PRIMARY KEY AUTO_INCREMENT,
    task_name VARCHAR(50) NOTNULLCOMMENT'任务名称（如部门同步、用户同步）',
    last_sync_time DATETIME NOTNULLCOMMENT'上次同步时间戳',
    create_time DATETIME DEFAULTCURRENT_TIMESTAMP,
    update_time DATETIME DEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP
);

步骤2： 拉取增量数据 每次同步时，从 sync_checkpoint 读取 last_sync_time，调用第三方 API 时传入该时间戳，只获取更新时间晚于该值的数据。

GET /api/departments?since=2024-05-20T10:00:00Z
GET /api/users?since=2024-05-20T10:00:00Z

步骤3： 更新时间戳

同步成功后，将 last_sync_time 更新为当前时间（或第三方返回的最新数据时间戳）。

优点

• 实现简单，第三方 API 通常自带 since/update_time 筛选参数。
• 资源消耗低，只处理变化的数据。

缺点

• 依赖第三方系统的 update_time 字段准确性（若第三方未正确更新该字段，会导致数据漏同步）。
• 若同步失败，需手动处理时间戳回滚，否则会丢失中间数据。

适用场景

• 第三方 API 支持按时间戳筛选（如钉钉、企业微信的增量接口）。
• 数据变更频率适中，对漏同步可通过后续全量同步兜底。

@Component
@Slf4j
@RequiredArgsConstructor
publicclass IncrementalSyncScheduler {

    privatefinal RestTemplate restTemplate = new RestTemplate();
    privatefinal DepartmentService departmentService;
    privatefinal UserService userService;
    privatefinal SyncCheckpointService checkpointService;

    @Value("${third-party.api-base-url}")
    private String apiBaseUrl;

    privatestaticfinal String TASK_NAME = "DEPT_USER_SYNC_MP";
    privatestaticfinal DateTimeFormatter dtf = DateTimeFormatter.ISO_LOCAL_DATE_TIME;

    // 每10分钟执行一次
    @Scheduled(cron = "0 */10 * * * *")
    public void performIncrementalSync() {
        log.info("--- 开始执行增量同步 ---");
        try {
            // 1. 获取上次同步时间
            LocalDateTime lastSyncTime = getLastSyncTime();
            
            // 2. 拉取增量数据
            String url = apiBaseUrl + "/api/changes?since=" + dtf.format(lastSyncTime);
            ChangeEventWrapper changes = restTemplate.getForObject(url, ChangeEventWrapper.class);

            if (changes == null || (changes.getDepartments().isEmpty() && changes.getUsers().isEmpty())) {
                log.info("没有增量数据。");
                updateCheckpoint();
                return;
            }

            // 3. 应用变更
            applyChanges(changes);

            // 4. 更新检查点
            updateCheckpoint();

            log.info("--- [MyBatis-Plus] 增量同步成功完成 ---");
        } catch (Exception e) {
            log.error("[MyBatis-Plus] 增量同步失败", e);
        }
    }

    private void applyChanges(ChangeEventWrapper changes) {
        if (!changes.getDepartments().isEmpty()) {
            log.info("处理 {} 条部门变更...", changes.getDepartments().size());
            for (ChangeEvent<Department> event : changes.getDepartments()) {
                Department data = event.getData();
                switch (event.getType()) {
                    case CREATE:
                    case UPDATE:
                        departmentService.saveOrUpdate(data);
                        break;
                    case DELETE:
                        departmentService.removeById(data.getId());
                        break;
                }
            }
        }
        if (!changes.getUsers().isEmpty()) {
            log.info("处理 {} 条用户变更...", changes.getUsers().size());
            for (ChangeEvent<User> event : changes.getUsers()) {
                User data = event.getData();
                switch (event.getType()) {
                    case CREATE:
                    case UPDATE:
                        userService.saveOrUpdate(data);
                        break;
                    case DELETE:
                        userService.removeById(data.getId());
                        break;
                }
            }
        }
    }

    private LocalDateTime getLastSyncTime() {
        SyncCheckpoint checkpoint = checkpointService.getByTaskName(TASK_NAME);
        return checkpoint != null ? checkpoint.getLastSyncTimestamp() : LocalDateTime.of(2000, 1, 1, 0, 0);
    }

    private void updateCheckpoint() {
        SyncCheckpoint checkpoint = checkpointService.getByTaskName(TASK_NAME);
        if (checkpoint == null) {
            checkpoint = new SyncCheckpoint();
            checkpoint.setTaskName(TASK_NAME);
        }
        checkpoint.setLastSyncTimestamp(LocalDateTime.now());
        checkpointService.saveOrUpdate(checkpoint);
    }

    // 辅助类
    publicstaticclass ChangeEventWrapper {
        private java.util.List<ChangeEvent<Department>> departments;
        private java.util.List<ChangeEvent<User>> users;
        // getters and setters
    }
    publicstaticclass ChangeEvent<T> {
        private String type;
        private T data;
        // getters and setters
    }
}

2、方案二：基于变更 ID 的增量同步（高可靠性）第三方系统为每条数据分配唯一的变更 ID（如 change_id），每次同步时只拉取 change_id > last_change_id 的数据。

变更 ID 通常按时间递增生成。

步骤1： 记录上次变更 ID

在 sync_checkpoint 表中增加 last_change_id 字段，存储上次同步的最大变更 ID。

ALTER TABLE sync_checkpoint ADD COLUMN last_change_id BIGINT DEFAULT 0 COMMENT '上次同步的最大变更ID';

步骤2： 拉取增量数据

调用第三方 API 时传入 last_change_id，只获取变更 ID 更大的数据。示例 API 请求：

GET /api/changes?last_change_id=12345

步骤3： 更新变更 ID

同步成功后，将 last_change_id 更新为本次同步到的最大变更 ID。

总结：

• 可靠性高，变更 ID 唯一且递增，不会漏同步或重复同步。
• 无需依赖时间戳，避免因时间偏差导致的问题。
• 第三方系统需支持变更 ID 筛选（并非所有 API 都提供）。

对数据一致性要求极高的场景（如金融、支付数据同步）。其实这种方案实现思路和时间戳类似，只是手动维护了一个自增的变更ID，用来规避时间戳未设值之类的情况。

方案三：实时同步 （Webhook）

实时同步的核心目标是 “数据变更后立即同步” ，实现 “准实时” 或 “实时” 的数据一致性。与定时同步不同，实时同步无需依赖定时任务触发，而是由 “事件驱动” （数据变更事件触发同步）。

以下是几种常见的实时同步实现方案，从简单到复杂，覆盖不同技术栈和场景：

1、Webhook 回调（最常用）

第三方系统（如钉钉、企业微信、CRM 系统）在数据发生变更时（如新增用户、修改部门），主动调用你的系统提供的 回调接口（Webhook Endpoint） ，将变更数据推送到你的系统，你的系统接收并处理这些数据。

步骤1： 提供 Webhook 接口

• 在你的系统中开发一个公开的接口（如 /api/webhook/sync），用于接收第三方推送的变更事件。
• 接口需支持 POST 请求，通常接收 JSON 格式的事件数据。

步骤2： 配置第三方 Webhook

• 在第三方系统的管理后台（如钉钉开放平台），配置你的 Webhook 接口地址，并选择需要监听的事件类型（如用户新增、部门删除）。

步骤3： 接收并处理事件

你的系统接收事件数据后，解析数据内容（如变更类型、变更数据、时间戳），并执行同步操作（插入、更新、删除本地数据库）。

关键注意点：

• 签名验证： 第三方会在请求头中携带签名（如 X-Signature），你需要验证签名的合法性，防止恶意请求。
• 幂等性处理： 由于网络重试等原因，可能会收到重复事件，需确保同步逻辑幂等（如通过 event_id 去重）。
• 异步处理： 接收到事件后，应立即返回响应（如 HTTP 200），再通过线程池或消息队列异步处理同步逻辑，避免阻塞第三方的回调请求。

@RestController
@RequestMapping("/api/webhook")
@Slf4j
publicclass WebhookController {

    @Autowired
    private SyncService syncService;
    @Autowired
    private WebhookSignatureService signatureService;

    @PostMapping("/dingtalk")
    public ResponseEntity<?> handleDingTalkWebhook(
            @RequestBody String requestBody,
            @RequestHeader("X-Signature") String signature,
            @RequestHeader("X-Timestamp") String timestamp) {
        
        // 1. 验证签名
        if (!signatureService.validateSignature(requestBody, timestamp, signature)) {
            log.warn("Webhook签名验证失败");
            return ResponseEntity.badRequest().body("Invalid signature");
        }

        // 2. 解析事件数据
        DingTalkWebhookEvent event = JsonUtils.parseObject(requestBody, DingTalkWebhookEvent.class);
        log.info("收到钉钉Webhook事件：{}", event.getEventType());

        // 3. 异步处理同步逻辑（避免阻塞）
        syncService.asyncProcessEvent(event);

        // 4. 立即返回响应
        return ResponseEntity.ok().body("{"errcode":0,"errmsg":"success"}");
    }
}

总结：

• 第三方系统支持 Webhook（如钉钉、企业微信、GitHub、PayPal 等）。
• 对实时性要求中等（秒级延迟可接受），且不希望引入复杂中间件的场景。

2、消息队列（MQ）异步同步（高可靠）

通过 消息队列（如 RabbitMQ、Kafka、RocketMQ）解耦数据变更源和同步目标：

• 数据变更源（如业务系统、第三方 API）将变更事件写入消息队列。
• 你的系统作为消费者，监听消息队列，读取事件并执行同步操作。

步骤1： 选择并部署消息队列

根据场景选择 MQ（如 RabbitMQ 适合可靠性优先，Kafka 适合高吞吐）。

步骤2： 生产端写入消息

数据变更时（如用户更新），生产端（如业务系统的服务）将变更事件（如用户 ID、变更字段、操作类型）序列化为消息，发送到 MQ 的指定主题 / 队列。

示例（Java + RabbitMQ）：

@Service
public class EventProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendUserUpdateEvent(User user) {
        UserUpdateEvent event = new UserUpdateEvent(user.getId(), user.getName(), LocalDateTime.now());
        rabbitTemplate.convertAndSend("user.sync.exchange", "user.update", event);
        log.info("发送用户更新事件：{}", user.getId());
    }
}

步骤3： 消费端处理消息

你的系统作为消费者，订阅 MQ 的主题 / 队列，接收消息后解析并同步到本地数据库。

示例（Java + RabbitMQ）：

@Service
publicclass EventConsumer {
    @Autowired
    private UserRepository userRepository;

    @RabbitListener(queues = "user.update.queue")
    public void handleUserUpdateEvent(UserUpdateEvent event) {
        log.info("接收用户更新事件：{}", event.getUserId());
        // 执行同步操作
        User user = userRepository.findByExternalId(event.getUserId())
                .orElseThrow(() -> new RuntimeException("用户不存在"));
        user.setName(event.getUserName());
        userRepository.save(user);
    }
}

步骤4： 保障可靠性

• 消息持久化： 将消息和队列设置为持久化，避免 MQ 重启后消息丢失。
• 消费者确认（Ack）： 消费者处理完消息后手动发送 Ack，确保消息被成功处理。
• 死信队列（DLQ）： 处理失败的消息（如数据库异常）转入死信队列，避免阻塞正常消息，后续可人工重试。

总结：

• 对数据可靠性要求高（如金融交易、订单同步），不允许消息丢失。
• 高并发场景（如每秒数千条数据变更），需要 MQ 削峰填谷。
• 多系统间数据同步（如业务系统 → 数据仓库 → 报表系统）。