springboot 整合 rabbitMQ（2）

springboot 整合 rabbitMQ（1）-CSDN博客

上期说了rabbitMQ的基础用法（普通队列模式）

这期学习一下如何防止消息重复消费和进阶用法（订阅者模式）

重复消费问题

导致 RabbitMQ 重复消费问题的原因：

1. 网络问题： 在分布式系统中，网络通信是不稳定的因素之一。如果生产者发送一条消息到 RabbitMQ 但尚未收到确认（acknowledgment），可能会导致 RabbitMQ 认为消息未被正确处理并重新发送。
2. 消费者故障： 消费者在处理消息时可能会发生故障，例如应用程序崩溃或因某种原因终止。如果 RabbitMQ 未收到消费者的确认消息，它可能会认为消息未被消费并重新发送。
3. 网络分区： 当分布式系统中的网络发生分区（网络隔离）时，可能会导致消息在不同部分之间重复传递。这是因为每个分区可能都会独立处理消息。
4. 消息重复传递策略： RabbitMQ 提供了不同的消息传递策略，例如“至少一次传递”和“最多一次传递”。这些策略可能会导致消息的重复传递，尤其在异常情况下。
5. 消费者超时设置不当： 如果消费者设置了较长的超时时间，在消费者未确认消息的情况下，RabbitMQ 可能会认为消息未被处理并重新发送。

解决思路

• 生产者发送消息时携带一个唯一的id
• 消费者每次消费前先判断一下在redis中是否在id，不存在就消费，消费完之后就把id存储到redis中

代码实现（这里在上一期的代码上进行修改）

生产者：

消费者：

    @Autowired
    private RedisTemplate redisTemplate;

    private static final String QUEUE_NAME="login_queue";

    @RabbitListener(queuesToDeclare = @Queue(QUEUE_NAME))
    public void test01(User getUser, Message message, Channel channel) throws IOException {
        //获取消息唯一id
        String messageId = message.getMessageProperties().getMessageId();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //判断消息是否存在
            if(redisTemplate.opsForHash().hasKey(QUEUE_NAME,messageId)){
                //存在表示该消息已被消费，return
                return;
            }
            //将该消息id存入redis 第三个参数随便字符串，不要填对象，没法序列化
            redisTemplate.opsForHash().put(QUEUE_NAME,messageId,"a");

            //发送消息到邮箱
            sendMail(getUser);
            System.out.println("消费成功");
            //手动确认消息  multiple 是一个布尔值，它决定了是否确认多个消息
            //  false：只确认指定的单个消息  true：确认所有未确认的消息，这些消息的deliveryTag 小于或等于指定的 deliveryTag
            channel.basicAck(deliveryTag,false);
        } catch (Exception e) {
            //消息回滚  deliveryTag：指定要拒绝的消息 false：表示只拒绝指定的这条消息 true：表示将消息重新入队，让其他消费者有机会再次处理这条消息。
            channel.basicNack(deliveryTag,false,true);
            System.out.println("消息消费异常");
        }
    }

订阅者模式：

是什么：

简单解释就是，可以将消息发送给不同类型的消费者。做到发布一次，消费多个。下图取自于官方网站（RabbitMQ）的发布/订阅模式的图例

  那么要使交换机接受到消息后转发给队列，就需要将队列绑定给交换机，可以写一个配置类，项目启动，自动将队列绑定给交换机

代码实现：

/**
 * @ClassName QueueConfig
 * @Description 声明队列和交换机 并将队列和交换机绑定
 * @Author
 * @Date 2024/10/9 16:28
 */
@Configuration
public class QueueConfig {
    //订阅模式交换机名称
    public static String EXCHANGENAME="fanout_exchange";
    public static String TESTQUEUE01="test_01";
    public static String TESTQUEUE02="test_02";

    @Bean
    public Queue queue01(){
        return new Queue(TESTQUEUE01);
    }

    @Bean
    public Queue queue02(){
        return new Queue(TESTQUEUE02);
    }

    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(EXCHANGENAME);
    }

    @Bean
    public Binding bingingQueue01(Queue queue01,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(queue01).to(fanoutExchange);
    }

    @Bean
    public Binding bingingQueue02(Queue queue02,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(queue02).to(fanoutExchange);
    }
}

生产者：

    只需要交换机的名字和传入的数据，队列名不需要填，因为交换机已经绑定队列![](https://i-blog.csdnimg.cn/direct/b5b38414e29843409c6bd2666590e7f4.png)

消费者：

效果：

    发送一条数据，两个消费者消费