RabbitMQ消费端消息分发问题研究

迈不过友情╰ 2021-07-24 17:23 502阅读 0赞

**一 先上思维导图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70][]

**二 再看代码**

1 生产者

package com.rabbitmq.basicqos;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.ConfirmListener;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.MessageProperties;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    
    import java.io.IOException;
    import java.util.Collections;
    import java.util.SortedSet;
    import java.util.TreeSet;
    import java.util.concurrent.TimeoutException;
    
    
    import static com.rabbitmq.common.Constant.NUM64;
    
    
    /**
    * @className: QosProducer
    * @description: Qos生产者
    * @date: 2020/5/14
    * @author: cakin
    */
    @Slf4j
    public class QosProducer {
        public static void main( String[] args ) throws IOException, TimeoutException {
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
    
            channel.exchangeDeclare("exchange.confirm", "direct", true, false, null);
            channel.queueDeclare("queue.basicqos", true, false, false, null);
            channel.queueBind("queue.basicqos", "exchange.confirm", "routingKey2");
    
    
            final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());
            // 启动confirm模式
            channel.confirmSelect();
    
    
            // 开启监听
            channel.addConfirmListener(new ConfirmListener() {
                // 处理成功
                public void handleAck( long deliveryTag, boolean multiple ) throws IOException {
                    log.info("消息发送成功: " + deliveryTag
                            + ", multiple : " + multiple);
                    // 收到消息，从消息序号队列中删除deliveryTag
                    if (multiple) {
                        confirmSet.headSet(deliveryTag - 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
    
    
                // 处理失败
                public void handleNack( long deliveryTag, boolean multiple ) throws IOException {
                    if (multiple) {
                        confirmSet.headSet(deliveryTag - 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
            });
    
    
            // 循环发送消息
            for (int i = 1; i < NUM64; i++) {
                String msg = "我是confirm模式消息.异步[" + i + "]";
                long tag = channel.getNextPublishSeqNo();
                // 发送消息，将deliveryTag加入到消息序号队列
                channel.basicPublish("exchange.confirm", "routingKey2", MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
                log.info("tag:" + tag);
                confirmSet.add(tag);
            }
        }
    }

2 消费者

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: RabbitConsumer.java
    * Author:   cakin
    * Date:     2020/5/2
    * Description: 消费端
    */
    package com.rabbitmq.basicqos;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    
    import static com.rabbitmq.common.Constant.NUM5;
    
    
    /**
    * Qos测试的消费端
    */
    @Slf4j
    public class QosConsumer {
        /**
         * 消息队列
         */
        private static final String QUEUE_NAME = "queue.basicqos";
    
    
        /**
         * 控制ack消息
         */
        private static int num = 0;
    
    
        public static void main( String[] args ) {
            try {
                Connection connection = ConnectionUtils.getConnection();
                Channel channel = connection.createChannel();
                // 客户端最多未确认消息数为5，一旦超过5，客户端将无法接收到消息
                channel.basicQos(NUM5);
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery( String consumerTag,
                                                Envelope envelope,
                                                AMQP.BasicProperties properties,
                                                byte[] body )
                            throws IOException {
                        log.info(" recv message: " + new String(body));
                        // 控制是接收到的第几个消息
                        num++;
                        // 模拟业务处理耗时1秒
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            log.info("catch InterruptedException");
                        }
    
    
                        // 只让是5的倍数的消息发ack消息
                        if (num % NUM5 == 0) {
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        } else {
                            log.trace("模拟不发ack情况");
                            // channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                channel.basicConsume(QUEUE_NAME, consumer);
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }

**三 运行说明**

1 先运行生产者。

2 再运行消费者，消费者可以通过屏蔽和开启 channel.basicAck(envelope.getDeliveryTag(), false); 代码来观察运行结果，以此达到理解 channel.basicQos(NUM5); 目的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70]: /images/20210724/a8693f3846ff4125b86b7e73e9d51e9b.png