Spring Boot整合RabbitMQ消息重复消费检测与防重机制

各位朋友，大家好！今天我们来聊聊Spring Boot整合RabbitMQ时，如何应对消息重复消费的问题。在分布式系统中，消息队列是常用的组件，但由于网络抖动、服务器故障等原因，消息重复消费的情况难以避免。如果不加以处理，可能导致数据错误、业务流程混乱等问题。因此，设计有效的消息重复消费检测和防重机制至关重要。

一、消息重复消费的场景分析

在讨论解决方案之前，我们先来分析一下消息重复消费可能发生的场景：

生产者发送消息后未收到ACK： 生产者发送消息到RabbitMQ后，如果网络中断或RabbitMQ服务器故障，生产者可能无法收到ACK确认。为了保证消息不丢失，生产者可能会重试发送，从而导致消息重复。
消费者消费消息后未发送ACK： 消费者从RabbitMQ接收到消息并处理后，如果网络中断或消费者自身故障，可能无法及时发送ACK确认。RabbitMQ会认为消息未被成功消费，将其重新放入队列，等待下次消费。
消费者处理消息超时： 消费者处理消息的时间超过RabbitMQ的配置的超时时间，RabbitMQ会认为消息未被成功消费，将其重新放入队列。

二、常见的消息防重方案

针对以上场景，我们可以采用多种防重方案，以下列出几种常用的方法：

幂等性设计： 这是最根本的解决方案。将消费者端的业务逻辑设计成幂等的，即多次执行的结果与一次执行的结果相同。比如，更新操作可以基于版本号进行，插入操作可以先判断数据是否存在。
唯一ID机制： 为每条消息生成一个全局唯一的ID，消费者在处理消息前，先判断该ID是否已被处理过。可以使用Redis、数据库等存储已处理的ID。
本地事务表： 消费者在本地数据库中创建一个事务表，用于记录已处理的消息ID。在消费消息前，先查询事务表，如果存在，则直接返回成功；否则，开启本地事务，处理消息并将消息ID写入事务表，提交事务。
乐观锁机制： 对于更新操作，可以引入乐观锁。在更新数据时，带上版本号，如果版本号不匹配，则更新失败，防止重复更新。

三、基于唯一ID机制的Spring Boot + RabbitMQ防重实现

接下来，我们以唯一ID机制为例，演示如何在Spring Boot项目中实现消息防重。

1. 项目依赖

首先，在pom.xml文件中添加RabbitMQ和Redis的依赖：

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

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2. 配置文件 (application.properties 或 application.yml)

配置RabbitMQ和Redis的连接信息：

# RabbitMQ configuration
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

# Redis configuration
spring.redis.host=localhost
spring.redis.port=6379

3. 消息实体类

定义消息实体类，包含消息ID和消息内容：

import java.io.Serializable;

public class Message implements Serializable {

    private String messageId;
    private String content;

    public Message() {
    }

    public Message(String messageId, String content) {
        this.messageId = messageId;
        this.content = content;
    }

    public String getMessageId() {
        return messageId;
    }

    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    @Override
    public String toString() {
        return "Message{" +
                "messageId='" + messageId + ''' +
                ", content='" + content + ''' +
                '}';
    }
}

4. Redis工具类

创建一个Redis工具类，用于判断消息ID是否存在，并存储已处理的消息ID：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class RedisUtil {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static final String MESSAGE_ID_PREFIX = "message_id:";
    private static final long MESSAGE_ID_EXPIRE_TIME = 60 * 60 * 24; // 24 hours

    public boolean isMessageIdProcessed(String messageId) {
        return stringRedisTemplate.hasKey(MESSAGE_ID_PREFIX + messageId);
    }

    public void saveMessageId(String messageId) {
        stringRedisTemplate.opsForValue().set(MESSAGE_ID_PREFIX + messageId, "processed", MESSAGE_ID_EXPIRE_TIME, TimeUnit.SECONDS);
    }
}

5. 消息生产者

创建一个消息生产者，用于发送消息到RabbitMQ：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class MessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static final String EXCHANGE_NAME = "my_exchange";
    private static final String ROUTING_KEY = "my_routing_key";

    public void sendMessage(String content) {
        String messageId = UUID.randomUUID().toString();
        Message message = new Message(messageId, content);
        System.out.println("Sending message: " + message);
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message);
    }
}

6. 消息消费者

创建一个消息消费者，用于接收并处理消息，并使用Redis进行防重：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {

    @Autowired
    private RedisUtil redisUtil;

    @RabbitListener(queues = "my_queue")
    public void receiveMessage(Message message) {
        String messageId = message.getMessageId();

        if (redisUtil.isMessageIdProcessed(messageId)) {
            System.out.println("Message with ID " + messageId + " already processed. Ignoring.");
            return;
        }

        try {
            // 模拟消息处理过程
            System.out.println("Received message: " + message);
            Thread.sleep(1000); // Simulate processing time
            System.out.println("Message processed successfully: " + messageId);

            // 保存消息ID到Redis
            redisUtil.saveMessageId(messageId);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Error processing message: " + messageId);
        } catch (Exception e) {
            System.err.println("Error processing message: " + messageId);
        }
    }
}

7. RabbitMQ配置

创建一个RabbitMQ配置类，用于声明交换机、队列和绑定关系：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    private static final String EXCHANGE_NAME = "my_exchange";
    private static final String QUEUE_NAME = "my_queue";
    private static final String ROUTING_KEY = "my_routing_key";

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, true); // durable = true
    }

    @Bean
    public Binding binding(DirectExchange directExchange, Queue queue) {
        return BindingBuilder.bind(queue).to(directExchange).with(ROUTING_KEY);
    }
}

8. 测试

创建一个测试类，用于发送消息：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class MessageProducerTest {

    @Autowired
    private MessageProducer messageProducer;

    @Test
    public void testSendMessage() {
        messageProducer.sendMessage("Hello, RabbitMQ!");
        messageProducer.sendMessage("Hello, RabbitMQ!"); // Send the same message again
    }
}

运行测试类，可以看到控制台输出：

Sending message: Message{messageId='...', content='Hello, RabbitMQ!'}
Sending message: Message{messageId='...', content='Hello, RabbitMQ!'}
Received message: Message{messageId='...', content='Hello, RabbitMQ!'}
Message processed successfully: ...
Message with ID ... already processed. Ignoring.

可以看到，第二次发送的消息被消费者检测到已经处理过，因此被忽略，从而实现了消息防重。

四、基于本地事务表的Spring Boot + RabbitMQ防重实现

除了使用Redis，我们还可以使用本地事务表来实现消息防重。

1. 创建消息处理记录表

首先，需要在本地数据库中创建一个消息处理记录表，用于记录已处理的消息ID。

CREATE TABLE `message_processed` (
  `message_id` varchar(255) NOT NULL,
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2. 创建实体类

创建对应于message_processed表的实体类：

import java.util.Date;

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;

@Entity
@Table(name = "message_processed")
public class MessageProcessed {

    @Id
    @Column(name = "message_id")
    private String messageId;

    @Column(name = "create_time")
    private Date createTime;

    public String getMessageId() {
        return messageId;
    }

    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }

    public Date getCreateTime() {
        return createTime;
    }

    public void setCreateTime(Date createTime) {
        this.createTime = createTime;
    }
}

3. 创建Repository

创建访问message_processed表的Repository接口：

import org.springframework.data.jpa.repository.JpaRepository;

public interface MessageProcessedRepository extends JpaRepository<MessageProcessed, String> {
}

4. 修改消息消费者

修改消息消费者，使用本地事务表进行防重：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
public class MessageConsumer {

    @Autowired
    private MessageProcessedRepository messageProcessedRepository;

    @RabbitListener(queues = "my_queue")
    @Transactional
    public void receiveMessage(Message message) {
        String messageId = message.getMessageId();

        if (messageProcessedRepository.existsById(messageId)) {
            System.out.println("Message with ID " + messageId + " already processed. Ignoring.");
            return;
        }

        try {
            // 模拟消息处理过程
            System.out.println("Received message: " + message);
            Thread.sleep(1000); // Simulate processing time
            System.out.println("Message processed successfully: " + messageId);

            // 保存消息ID到本地事务表
            MessageProcessed messageProcessed = new MessageProcessed();
            messageProcessed.setMessageId(messageId);
            messageProcessedRepository.save(messageProcessed);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Error processing message: " + messageId);
        } catch (Exception e) {
            System.err.println("Error processing message: " + messageId);
        }
    }
}

注意：

需要在receiveMessage方法上添加@Transactional注解，开启本地事务。
使用messageProcessedRepository.existsById(messageId)判断消息ID是否已存在。
使用messageProcessedRepository.save(messageProcessed)将消息ID保存到本地事务表。

五、一些补充说明

消息ID的生成： 消息ID可以使用UUID、雪花算法等方式生成，需要保证全局唯一性。
消息ID的存储： 可以使用Redis、数据库等存储已处理的消息ID。Redis的优点是性能高，但数据可能会丢失；数据库的优点是数据可靠性高，但性能相对较低。
消息ID的过期时间： 需要设置消息ID的过期时间，避免存储空间被耗尽。过期时间可以根据业务需求进行设置。
消费失败重试机制： 如果消息消费失败，可以进行重试。重试次数和重试间隔需要根据业务需求进行设置。
死信队列： 如果消息重试多次仍然失败，可以将消息放入死信队列，由人工进行处理。

六、各种方案的对比

方案	优点	缺点	适用场景
幂等性设计	从根本上解决问题，无需额外存储和判断。	需要对业务逻辑进行改造，可能比较复杂。	任何场景，但需要业务逻辑支持。
唯一ID机制	实现简单，易于理解。	需要额外的存储空间（Redis、数据库等），性能会有所影响。	适用于对数据一致性要求较高的场景，但允许一定程度的性能损失。
本地事务表	数据可靠性高，与本地事务绑定，保证数据一致性。	性能相对较低，会增加数据库的负担。	适用于对数据一致性要求非常高的场景，对性能要求不高的场景。
乐观锁机制	适用于更新操作，可以避免并发更新导致的数据错误。	需要在数据表中添加版本号字段，如果并发更新冲突较多，可能会导致更新失败率较高。	适用于更新操作，且对数据一致性要求较高的场景。

总而言之，选择哪种方案需要根据具体的业务场景和需求进行权衡。没有一种方案是万能的，需要选择最适合自己的方案。

最后：选择合适的防重策略并确保实现正确

消息重复消费是分布式系统中常见的问题，需要根据实际情况选择合适的防重方案，并确保实现正确。幂等性设计是最根本的解决方案，但需要对业务逻辑进行改造。唯一ID机制和本地事务表是常用的防重方案，但需要额外的存储空间和性能开销。乐观锁机制适用于更新操作，可以避免并发更新导致的数据错误。选择合适的方案并确保实现正确，才能有效地解决消息重复消费的问题，保证系统的稳定性和可靠性。

Spring Boot整合RabbitMQ消息重复消费检测与防重机制

发表回复 取消回复

发表回复取消回复