下面我将详细介绍如何使用SpringBoot整合RabbitMQ实现数据同步，包括配置步骤和代码实现。

一、项目配置

1. 添加依赖

首先在pom.xml中添加RabbitMQ和SpringBoot相关依赖：

<dependencies>
    <!-- Spring Boot Starter for RabbitMQ -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    
    <!-- Spring Boot Starter Web (可选，如果需要REST接口) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <!-- Lombok (可选，简化代码) -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

2. 配置文件

在application.yml或application.properties中添加RabbitMQ配置：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    
    # 生产者配置
    publisher-confirm-type: correlated  # 发布确认
    publisher-returns: true            # 发布返回
    
    # 消费者配置
    listener:
      simple:
        acknowledge-mode: manual       # 手动确认
        prefetch: 10                  # 每次从broker拉取的消息数量

二、代码实现

1. 创建RabbitMQ配置类

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
​
@Configuration
@EnableRabbit
public class RabbitMQConfig {
​
    // 定义数据同步的交换机和队列名称
    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";
​
    // 创建直连交换机
    @Bean
    public DirectExchange dataSyncExchange() {
        return new DirectExchange(DATA_SYNC_EXCHANGE, true, false);
    }
​
    // 创建队列
    @Bean
    public Queue dataSyncQueue() {
        return QueueBuilder.durable(DATA_SYNC_QUEUE)
                .withArgument("x-queue-mode", "lazy") // 惰性队列，消息直接持久化到磁盘
                .build();
    }
​
    // 绑定队列到交换机
    @Bean
    public Binding dataSyncBinding() {
        return BindingBuilder.bind(dataSyncQueue())
                .to(dataSyncExchange())
                .with(DATA_SYNC_ROUTING_KEY);
    }
​
    // 配置JSON消息转换器
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
​
    // 配置RabbitTemplate
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        
        // 消息发送到Exchange确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("消息发送到Exchange成功: " + correlationData);
            } else {
                System.out.println("消息发送到Exchange失败: " + cause);
            }
        });
        
        // 消息从Exchange路由到Queue失败回调
        rabbitTemplate.setReturnsCallback(returned -> {
            System.out.println("消息从Exchange路由到Queue失败: " + returned.getMessage());
        });
        
        return rabbitTemplate;
    }
}

2. 创建消息生产者

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
​
import java.nio.charset.StandardCharsets;
import java.util.UUID;
​
@Slf4j
@Service
public class DataSyncProducer {
​
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private ObjectMapper objectMapper;
​
    /**
     * 发送数据同步消息
     * @param data 要同步的数据
     */
    public void sendDataSyncMessage(Object data) {
        try {
            // 将数据转换为JSON字符串
            String jsonData = objectMapper.writeValueAsString(data);
            
            // 构建消息
            Message message = MessageBuilder.withBody(jsonData.getBytes(StandardCharsets.UTF_8))
                    .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                    .setMessageId(UUID.randomUUID().toString())
                    .build();
            
            // 发送消息
            rabbitTemplate.convertAndSend(
                    RabbitMQConfig.DATA_SYNC_EXCHANGE,
                    RabbitMQConfig.DATA_SYNC_ROUTING_KEY,
                    message,
                    correlationData -> {
                        correlationData.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        return correlationData;
                    }
            );
            
            log.info("数据同步消息发送成功: {}", jsonData);
        } catch (JsonProcessingException e) {
            log.error("数据序列化失败: {}", e.getMessage(), e);
        }
    }
}

3. 创建消息消费者

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
​
import java.io.IOException;
​
@Slf4j
@Service
public class DataSyncConsumer {
​
    @Autowired
    private ObjectMapper objectMapper;
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
​
    /**
     * 接收并处理数据同步消息
     */
    @RabbitListener(queues = RabbitMQConfig.DATA_SYNC_QUEUE)
    public void handleDataSyncMessage(@Payload Message message) throws IOException {
        try {
            // 获取消息内容
            String jsonData = new String(message.getBody(), StandardCharsets.UTF_8);
            
            // 这里根据实际业务需求解析数据
            // 示例：假设同步的是用户数据
            User user = objectMapper.readValue(jsonData, new TypeReference<User>() {});
            
            log.info("接收到数据同步消息: {}", user);
            
            // 处理数据同步逻辑
            processDataSync(user);
            
            // 手动确认消息已处理
            rabbitTemplate.execute(channel -> {
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                channel.basicAck(deliveryTag, false);
                return null;
            });
            
        } catch (Exception e) {
            log.error("处理数据同步消息失败: {}", e.getMessage(), e);
            
            // 处理失败，可以选择重试或放入死信队列
            rabbitTemplate.execute(channel -> {
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                channel.basicNack(deliveryTag, false, false);
                return null;
            });
        }
    }
    
    private void processDataSync(User user) {
        // 这里实现具体的数据同步逻辑
        // 例如：将数据保存到数据库、更新缓存等
        log.info("正在同步用户数据: {}", user);
        // 模拟业务处理
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        log.info("用户数据同步完成: {}", user);
    }
}

4. 创建实体类（示例）

import lombok.Data;
​
@Data
public class User {
    private Long id;
    private String username;
    private String email;
    private String phone;
}

5. 创建测试Controller（可选）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
​
@RestController
public class DataSyncController {
​
    @Autowired
    private DataSyncProducer dataSyncProducer;
​
    @PostMapping("/sync/user")
    public String syncUserData(@RequestBody User user) {
        dataSyncProducer.sendDataSyncMessage(user);
        return "用户数据同步请求已发送";
    }
}

三、高级配置（可选）

1. 死信队列配置

在RabbitMQConfig中添加死信队列配置：

// 死信交换机
@Bean
public DirectExchange dataSyncDlxExchange() {
    return new DirectExchange("data.sync.dlx.exchange", true, false);
}
​
// 死信队列
@Bean
public Queue dataSyncDlxQueue() {
    return QueueBuilder.durable("data.sync.dlx.queue").build();
}
​
// 绑定死信队列
@Bean
public Binding dataSyncDlxBinding() {
    return BindingBuilder.bind(dataSyncDlxQueue())
            .to(dataSyncDlxExchange())
            .with("data.sync.dlx.routing.key");
}
​
// 修改主队列配置，添加死信设置
@Bean
public Queue dataSyncQueue() {
    return QueueBuilder.durable(DATA_SYNC_QUEUE)
            .withArgument("x-queue-mode", "lazy")
            .withArgument("x-dead-letter-exchange", "data.sync.dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "data.sync.dlx.routing.key")
            .withArgument("x-message-ttl", 10000) // 消息存活时间10秒
            .withArgument("x-max-length", 1000)   // 队列最大长度
            .build();
}

2. 消费者并发配置

在application.yml中增加消费者并发配置：

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 5   # 最小消费者数量
        max-concurrency: 10 # 最大消费者数量

四、测试流程

1. 启动RabbitMQ服务

2. 启动SpringBoot应用

3. 使用Postman或curl发送POST请求到/sync/user接口

4. 观察控制台日志，确认消息发送和消费过程

五、关键点说明

1. 消息确认机制：

   • 生产者通过publisher-confirm-type和publisher-returns确保消息正确发送到Exchange和Queue

   • 消费者通过手动确认(acknowledge-mode: manual)确保消息正确处理

2. 消息持久化：

   • 队列设置为持久化(durable)

   • 消息设置为持久化(setDeliveryMode(MessageDeliveryMode.PERSISTENT))

3. 异常处理：

   • 消费者处理失败时可以选择重试或进入死信队列

4. 性能考虑：

   • 使用惰性队列(x-queue-mode: lazy)减少内存使用

   • 合理设置预取数量(prefetch)和并发消费者数量

通过以上配置和代码，可以实现一个健壮的基于SpringBoot和RabbitMQ的数据同步系统。