Java初识RabbitMQ一消费端自定义监听

水深无声 2023-07-04 15:42 17阅读 0赞

推荐：[消息中间件汇总][Link 1]

## Java初识RabbitMQ一消费端自定义监听 ##

之前的例子中，消费端实现监听绑定的Queue是否有准备好了的消息时，是通过一个`while`死循环来实现的，看起来很不优雅，如下所示：

// 接收消息
            while (true){ 
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String msg = new String(delivery.getBody());
                System.out.println(msg);
            }

首先我们来看一下`basicConsume()`函数，之前我们都是把`consumer`（不是自定义的消费者）传给`callback`，所以实现消费端消费消息，都是在`while`死循环里面，通过`QueueingConsumer.Delivery`来接收消息。如果想实现消费端自定义监听，我们需要自己实现`Consumer`接口，并且将该实现类的实例传给`callback`，但是由于`Consumer`接口的方法太多了，实现起来太繁琐，虽然需要实现的方法不多，但是代码看起来也不简洁，所以我们换一种方法，通过继承`Consumer`接口的实现类`DefaultConsumer`，并且只要重写`handleDelivery()`方法即可。

/** * Start a non-nolocal, non-exclusive consumer, with * a server-generated consumerTag. * @param queue the name of the queue * @param autoAck true if the server should consider messages * acknowledged once delivered; false if the server should expect * explicit acknowledgements * @param callback an interface to the consumer object * @return the consumerTag generated by the server * @throws java.io.IOException if an error is encountered * @see com.rabbitmq.client.AMQP.Basic.Consume * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer) */
        String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

`DefaultConsumer`类源码如下：

// Copyright (c) 2007-Present Pivotal Software, Inc. All rights reserved.
    //
    // This software, the RabbitMQ Java client library, is triple-licensed under the
    // Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
    // ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
    // LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL,
    // please see LICENSE-APACHE2.
    //
    // This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
    // either express or implied. See the LICENSE file for specific language governing
    // rights and limitations of this software.
    //
    // If you have any questions regarding licensing, please contact us at
    // info@rabbitmq.com.
    
    
    package com.rabbitmq.client;
    
    import java.io.IOException;
    
    /** * Convenience class providing a default implementation of {@link Consumer}. * We anticipate that most Consumer implementations will subclass this class. */
    public class DefaultConsumer implements Consumer { 
        /** Channel that this consumer is associated with. */
        private final Channel _channel;
        /** Consumer tag for this consumer. */
        private volatile String _consumerTag;
    
        /** * Constructs a new instance and records its association to the passed-in channel. * @param channel the channel to which this consumer is attached */
        public DefaultConsumer(Channel channel) { 
            _channel = channel;
        }
    
        /** * Stores the most recently passed-in consumerTag - semantically, there should be only one. * @see Consumer#handleConsumeOk */
        public void handleConsumeOk(String consumerTag) { 
            this._consumerTag = consumerTag;
        }
    
        /** * No-op implementation of {@link Consumer#handleCancelOk}. * @param consumerTag the defined consumer tag (client- or server-generated) */
        public void handleCancelOk(String consumerTag) { 
            // no work to do
        }
    
        /** * No-op implementation of {@link Consumer#handleCancel(String)} * @param consumerTag the defined consumer tag (client- or server-generated) */
        public void handleCancel(String consumerTag) throws IOException { 
            // no work to do
        }
    
        /** * No-op implementation of {@link Consumer#handleShutdownSignal}. */
        public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { 
            // no work to do
        }
    
         /** * No-op implementation of {@link Consumer#handleRecoverOk}. */
        public void handleRecoverOk(String consumerTag) { 
            // no work to do
        }
    
        /** * No-op implementation of {@link Consumer#handleDelivery}. */
        public void handleDelivery(String consumerTag,
                                   Envelope envelope,
                                   AMQP.BasicProperties properties,
                                   byte[] body)
            throws IOException
        { 
                // no work to do
        }
    
        /** * Retrieve the channel. * @return the channel this consumer is attached to. */
        public Channel getChannel() { 
            return _channel;
        }
    
        /** * Retrieve the consumer tag. * @return the most recently notified consumer tag. */
        public String getConsumerTag() { 
            return _consumerTag;
        }
    }

**生产端**

package com.kaven.rabbitmq.api.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer { 
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // default exchange
        private static String exchange = "";
        // default exchange 的路由规则： routingKey（test） 将匹配同名的 queue(test)
        private static String routingKey = "test";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 1 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建Connection
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
    
            // 4 发送消息
            for (int i = 0; i < 5; i++) { 
                String msg = "RabbitMQ: consumer message" + i;
                channel.basicPublish(exchange , routingKey , null , msg.getBytes());
            }
    
            // 5 关闭连接
            channel.close();
            connection.close();
        }
    }

**消费端**

重点是`MyConsumer`类，它来继承`DefaultConsumer`类，实现消费端自定义监听的逻辑。

package com.kaven.rabbitmq.api.consumer;
    
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    import java.io.IOException;
    
    public class MyConsumer extends DefaultConsumer { 
    
        public MyConsumer(Channel channel) { 
            super(channel);
        }
    
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                                   AMQP.BasicProperties properties, byte[] body) throws IOException { 
            System.out.println("------------ consumer message -----------");
            System.out.println("consumerTag：" + consumerTag);
            System.out.println("envelope：" + envelope);
            System.out.println("properties：" + properties);
            System.out.println("body：" + new String(body));
        }
    }

package com.kaven.rabbitmq.api.consumer;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer { 
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // default exchange
        private static String exchange = "";
        // 队列名
        private static String queueName = "test";
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            // 1 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建Connection
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
            // 4 创建Queue
            channel.queueDeclare(queueName , true , false , false , null);
    
            // 5 消费端开始消费信息
            channel.basicConsume(queueName , true , new MyConsumer(channel));
        }
    }

**测试**

启动生产端和消费端，消费端可以接收到消息，消息在`body`属性里面，如下所示：

------------ consumer message -----------
    consumerTag：amq.ctag-CebIvWwveIxajYqFBqoZmA
    envelope：Envelope(deliveryTag=1, redeliver=false, exchange=, routingKey=test)
    properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    body：RabbitMQ: consumer message0
    ------------ consumer message -----------
    consumerTag：amq.ctag-CebIvWwveIxajYqFBqoZmA
    envelope：Envelope(deliveryTag=2, redeliver=false, exchange=, routingKey=test)
    properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    body：RabbitMQ: consumer message1
    ------------ consumer message -----------
    consumerTag：amq.ctag-CebIvWwveIxajYqFBqoZmA
    envelope：Envelope(deliveryTag=3, redeliver=false, exchange=, routingKey=test)
    properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    body：RabbitMQ: consumer message2
    ------------ consumer message -----------
    consumerTag：amq.ctag-CebIvWwveIxajYqFBqoZmA
    envelope：Envelope(deliveryTag=4, redeliver=false, exchange=, routingKey=test)
    properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    body：RabbitMQ: consumer message3
    ------------ consumer message -----------
    consumerTag：amq.ctag-CebIvWwveIxajYqFBqoZmA
    envelope：Envelope(deliveryTag=5, redeliver=false, exchange=, routingKey=test)
    properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    body：RabbitMQ: consumer message4

`RabbitMQ Management`也有反应。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]

[Link 1]: https://kaven.blog.csdn.net/article/details/104223534
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200212152909779.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70