RabbitMQ生产方式和解决消息可靠性投递及其他问题

RabbitMQ基本结构

3种Exchange(交换机)与对应生产方式

  1. direct
    将消息发送到与交换机绑定且对应routingkey的队列
    生产方式有:Work queues(工作模式)Routing(路由模式)
  2. fanout
    将消息发送到与交换机绑定的队列,不需要routingkey
    生产方式有:Publish/Subscribe(订阅模式)
  3. topic
    将消息发送到与交换机绑定的队列,routingkey模糊匹配
    生产方式有:Topics(主题模式)

Queue队列重载参数

durable:是否持久化,true 交换机或者队列会存到本地文件数据库,当mq重启,依然在,false的话,重启或者断电,就没了;默认true
autoDelete:是否自动删除,true当没有Consumer消费者时候,自动删除掉;默认false
exclusive:是否独占:true的话只有一个消费者监听这个队列。默认false
    /**
     * 定义一个direct队列
     *
     * @return
     */
    @Bean
    public Queue directQueue() {
        return new Queue(DIRECT_QUEUE);
    }

    /**
     * 定义一个带参direct队列
     *
     * boolean durable 是否持久化
     * boolean autoDelete 是否自动删除
     * boolean exclusive 是否独占
     * @return
     */
    @Bean
    public Queue directQueue2() {
        return new Queue(DIRECT_QUEUE, true, false, false);
    }

RabbitMQ生产方式

  • Work queues(工作模式)

    一个或者多个消费者共同消费一个队列中的消息(队列中的每一个消息只可能被其中一个消费者消费)
    

    在这里插入图片描述

  • Routing(路由模式)

    跟订阅模式类似,只不过在订阅模式的基础上加上了类型,订阅模式是分发到所有绑定到交换机的队列,路由模式
    

    在这里插入图片描述

  • Publish/Subscribe(订阅模式)

    生产者生产的消息,所有订阅过的消费者都能够接收到消息(队列中的每一个消息可以被多个消费者消费)
    

    在这里插入图片描述

  • Topics(主题模式)

    主题模式和路由模式很像,路由模式是精确匹配,主题模式是模糊匹配,* 匹配一个单词,# 匹配零个或者多个单词
    

    在这里插入图片描述

消息投递流程

Producer(生产者)=>Channel(网络信道)=>Exchange(交换机)=>Queue(队列,到此会进行存储,即完成投递)=>Channel=>Consumer(消费者)

考虑以下问题

本文着重讲解消息可靠性投递

  1. 如何保证消息没有重复消费

    保证消费者的幂等性
    
  2. 消息可靠性投递

    生产端

    confirm确认模式和return退回模式(一般是一起使用)
    

    confirm确认模式(保证消息从生产者投递到交换机)

    生产者消息投递后,如果Broker收到消息,则会给生产者一个应答(ack),生产者进行接收应答,用来确定这条消息是否正常的发送到Broker
    

    实现如下:
    1.配置文件:spring.rabbitmq.publisher-confirm-type=correlated

    server:
      port: 80
      
    spring:
      rabbitmq:
        host: 192.168.31.6
        port: 5672
        username: linhy
        password: 123456
        virtual-host: /
        # confirm 确认模式
        publisher-confirm-type: correlated
    

    2.消息发送类实现RabbitTemplate.ConfirmCallback接口,实现confirm方法
    3.初始化RabbitTemplate,rabbitTemplate.setConfirmCallback(this);
    4.发送消息new一个CorrelationData对象将主键ID放入,将该对象一起发送(confirm方法参数是一个消息唯一标识,即CorrelationData对象,所以需要将主键ID放入)
    5.判断confirm方法ack参数进行补发操作
    消息发送类具体代码示例

    public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate.ConfirmCallback {
    
        @Resource
        private AmqpTemplate amqpTemplate;
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 初始化confirm 确认模式
         */
        @PostConstruct
        public void init() {
            rabbitTemplate.setConfirmCallback(this);
        }
    
        /**
         * String exchange 交换机名称
         * String routingKey 路由Key
         * Object object 具体发送的消息
         *
         * @param message
         */
        @Override
        public void sendMessage(String message) {
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId("12050");
            rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return
        }
    
        /**
         * confirm只能保证消息从生产者投递到交换机后,进入队列前
         *
         * @param correlationData 消息唯一标识
         * @param ack             交换机是否成功收到消息 true成功  false失败
         * @param cause           失败原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("消息已被推送=========>" + correlationData);
            if (ack) {
                System.out.println("交换机接受消息成功" + cause);
            } else {
                System.out.println("交换机接受消息失败" + cause);
                // 消息补发操作
                System.out.println("失败id:" + correlationData.getId());
            }
        }
    
    }
    

    return退回模式(保证消息从交换机投递到队列)

    用于处理交换机不存在或指定的路由key路由不到的情况
    

    实现如下:
    1.配置文件:publisher-returns: true(默认为false,即消息路由不可达时消息丢弃;true消息回退)

    server:
      port: 80
      
    spring:
      rabbitmq:
        host: 192.168.31.6
        port: 5672
        username: linhy
        password: 123456
        virtual-host: /
        # confirm 确认模式
        publisher-confirm-type: correlated
        # return 退回模式
        publisher-returns: true
    

    2.消息发送类实现RabbitTemplate.ReturnCallback接口,实现returnedMessage方法
    3.初始化RabbitTemplate,rabbitTemplate.setReturnCallback(this);
    4.获取主键ID进行日志纪录或其他操作(returnedMessage方法参数是消息主体,生产者发送时可将主键ID作为消息主体发送)
    消息发送类具体代码示例

    public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    
        @Resource
        private AmqpTemplate amqpTemplate;
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 初始化confirm 确认模式和return 退回模式
         */
        @PostConstruct
        public void init() {
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
        }
    
        /**
         * String exchange 交换机名称
         * String routingKey 路由Key
         * Object object 具体发送的消息
         *
         * @param message
         */
        @Override
        public void sendMessage(String message) {
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId("12050");
            rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return
        }
    
        /**
         * confirm只能保证消息从生产者投递到交换机后,进入队列前
         *
         * @param correlationData 消息唯一标识
         * @param ack             交换机是否成功收到消息 true成功  false失败
         * @param cause           失败原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("消息已被推送=========>" + correlationData);
            if (ack) {
                System.out.println("交换机接受消息成功" + cause);
            } else {
                System.out.println("交换机接受消息失败" + cause);
                // 消息补发操作
                System.out.println("失败id:" + correlationData.getId());
            }
        }
    
        /**
         * returnedMessage能保证消息从交换机投递到队列
         *
         * @param message    消息主体
         * @param replyCode  返回code
         * @param replyText  返回信息
         * @param exchange   交换机
         * @param routingKey 路由key
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("消息推送失败,return方法被执行...");
            System.out.println("消息主体:" + new String(message.getBody()));
            System.out.println("replyCode:" + replyCode);
            System.out.println("replyText:" + replyText);
            System.out.println("exchange:" + exchange);
            System.out.println("routingKey:" + routingKey);
        }
    
    }
    

    消费端

    消费端手动ACK确认
    

    实现如下:
    1.确认方法改为手动确认,配置文件:acknowledge-mode: manual

    server:
      port: 81
      
    spring:
      rabbitmq:
        host: 192.168.31.6
        port: 5672
        username: admin
        password: admin
        virtual-host: /
        # 消费端手动ack确认队列消息
        listener:
          simple:
            acknowledge-mode: manual
    

    2.消费类增加Channel对象参数和deliveryTag参数
    3.channel.basicAck(deliveryTag, true)确认队列消息,channel.拒签,是否回到队列
    消息接收类具体代码示例
    Service

    public interface RabbitMqService {
    
        /**
         * 消费端手动ack确认队列消息
         *
         * @param message     消息
         * @param channel     管道
         * @param deliveryTag 标号
         */
        public void receiveSubMessageAck(String message, Channel channel, long deliveryTag);
    
    }
    

    Service实现

    public class RabbitMqServiceImpl implements RabbitMqService {
    
        @Resource
        private AmqpTemplate amqpTemplate;
    
        @Override
        @RabbitListener(queues = {RabbitMQConfig.ACK_QUEUE})
        public void receiveSubMessageAck(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
            try {
                System.out.println("接收到的mq消息:" + message);
                // 处理业务
                System.out.println("开始处理业务=========>");
    
                // 消费者手动确认队列消息(批量确认)
                if (deliveryTag == 5) {
                    channel.basicAck(deliveryTag, true);// long deliveryTag 标号(当前消息在队列中的的索引), boolean multiple 是否批量处理
                }
                // 消费者手动确认队列消息(单个确认,低效)
                // channel.basicAck(deliveryTag, true);
                
            } catch (Exception e) {
            	// 报错,消费者进行拒签
                e.printStackTrace();
                try {
                    channel.basicNack(deliveryTag, false, false);//long deliveryTag 标号, boolean multiple 是否批量处理, boolean requeue 是否回到队列(false直接丢弃)
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
        }
    }
    
  3. 怎么保证消息传递的顺序

    开启多个队列,消息投递到同一个队列,一个消费者消费
    
  4. 一致性问题

    一致性问题即事务问题,可使用nacos注册中心+seata解决分布式事务问题,下次更新
    
本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>