RabbitMQ如何保证幂等性_rabbitmq如何保证消息的幂等性

一、简介

幂等性是分布式中比较重要的一个概念，是指在多作业操作时候避免造成重复影响，其实就是保证同一个消息不被消费者重复消费两次，但是可能存在网络波动等问题，生产者无法接受消费者发送的ack信息，因此这条消息将会被重复发送给其他消费者进行消费，实际上这条消息已经被消费过了，这就是重复消费的问题。

如何避免重复消费的问题
1.消息全局唯一ID
2.通过redis中的setnx命令，给消息分配一个全局ID，当消费了这个消息的时候写入redis

----下面通过第一种方式示例来避免重复消费

二、消息全局ID

【1】 将队列和交换机绑定

@Configuration
public class DirectRabbitConfig {

    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false

        return new Queue("TestDirectQueue",true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        //  return new DirectExchange("TestDirectExchange",true,true);
        return new DirectExchange("TestDirectExchange",true,false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

【2】写一个接口来发送消息

@RestController
public class SendMessageController {
    @Resource
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法

    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        MessageProperties messageProperties = new MessageProperties();

        messageProperties.setMessageId(UUID.randomUUID().toString());
        messageProperties.setContentType("text/plain");
        messageProperties.setContentEncoding("utf-8");
        Message message = new Message("hello,message idempotent!".getBytes(), messageProperties);

        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting",message);
        return "ok";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

【3】消费者消费消息

注：这里是手动确认模式通过实现ChannelAwareMessageListener方法的方式重写了onMessage方法，可以实现ack

相关配置代码

@Configuration
public class MessageListenerConfig {

    @Resource
    private CachingConnectionFactory connectionFactory;
    @Resource
    private MyAckReceiver myAckReceiver;//消息接收处理类

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
        container.setQueueNames("TestDirectQueue");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        container.setMessageListener(myAckReceiver);
        return container;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

【4】 消息消费者

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){

                /*
                 * 消费消息之前，根据message中设置的全局ID查看消息是否已经被消费过
                 */
                String messageId = message.getMessageProperties().getMessageId();

                if (StringUtils.isBlank(messageId)){
                    System.out.println("获取ID为空");
                    return;
                }
                String str = "----数据库业务操作-------";
                if (StringUtils.isNotBlank(str)){
                    // 消费当前消息，并获取message内容
                    String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                    System.out.println("---------生产者发送消息内容--------------messageId："+ messageId + ",内容："+ msg +"");

                    // 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息（成功消费，消息从队列中删除 ）
                    channel.basicAck(deliveryTag, false);
                }else {

                    // ack返回false，requeue-true并重新回到队列
                    channel.basicNack(deliveryTag, false, true);
                }
            }
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

实现效果

三、总结

可以看到消费者已经拿到了全局ID，消费完之后成功把队列的消息删除，这样就可以避免消息被重复消费