RabbitMQ消费端代码注意事项

落日映苍穹つ 2021-07-24 23:34 470阅读 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);
                    }
                }
            });
    
    
            // 模拟发送大量消息，观测消费端使用QueueingConsumer的性能
            for (int i = 1; i < 10000*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: QueueingConsumer
    * Author:   cakin
    * Date:     2020/5/12
    */
    package com.rabbitmq.basicqos;
    
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    
    import static com.rabbitmq.common.Constant.NUM2;
    
    
    /**
    * @ClassName: QueueingConsumer
    * @Description: QueueingConsumer测试
    * @Date: 2020/5/12
    * @Author: cakin
    */
    
    
    @Slf4j
    public class QueueingConsumerTest {
        /**
         * 消费队列
         */
        private static final String QUEUEING_QUEUE_NAME = "queue.basicqos";
    
    
        public static void main( String[] args ) throws Exception {
    
    
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
    
            // 测试basicQos对QueueingConsumer性能的影响，通过开启和屏蔽下面这行来观测
            // channel.basicQos(3);
            // 创建一个QueueingConsumer类型的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(QUEUEING_QUEUE_NAME, false, consumer);
            log.info(" [x] Awaiting requests");
    
    
            while (true) {
                // 如果没收到消息，会阻塞
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    
    
                String message = new String(delivery.getBody());
                log.info(message);
                // 手动发送确认消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

**三 最后看测试方法**

1 打开jvisualvm工具

2 运行生产者

3 运行消息者，通过开启和屏蔽 channel.basicQos(3); 这行代码，来观察它对QueueingConsumer消费者的性能影响。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70]: /images/20210724/1e4cbe80c9bb44518b3e4711983c457e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70 1]: /images/20210724/025158a621cf459abcabc97cc1a1f344.png