【消息队列开发】实现消费者订阅消息

拼搏现实的明天。 2024-04-17 10:55 90阅读 0赞

#### 文章目录 ####

*  ?前言
 *  ?关于订阅消息方法参数解析
 *  ?如何实现将消息推送给消费者
 *  ?消费者类
 *  ?消费消息的流程
 *  ?如何实现消息确认呢？
 *  ⭕总结

## ?前言 ##

本次开发任务

*  实现消费者订阅消息

## ?关于订阅消息方法参数解析 ##

我们关于订阅消息的方法如下：

![在这里插入图片描述][d6f59e2605d144f7a7168f33f3d7bfc5.png]

*  consumerTag: 消费者的身份标识
 *  autoAck: 消息被消费完成后, 应答的方式. 为 true 自动应答. 为 false 手动应答.
 *  consumer: 是一个回调函数，此处类型设定成函数式接口. 这样后续调用 basicConsume 并且传实参的时候, 就可以写作 lambda 样子了
 *  queueName：需要订阅消息的队列

![在这里插入图片描述][36a22a51cb3646a583d9a56818151a65.png]

回调函数实现代码如下：

/*
     * 只是一个单纯的函数式接口(回调函数). 收到消息之后要处理消息时调用的方法.
     */
    @FunctionalInterface
    public interface Consumer {
        
        // Delivery 的意思是 "投递", 这个方法预期是在每次服务器收到消息之后, 来调用.
        // 通过这个方法把消息推送给对应的消费者.
        void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException;
    }

## ?如何实现将消息推送给消费者 ##

消费者订阅消息后，如何将消息推送给消费者呢？

首先我们需要知道一个队列都有那些消费者进行消费，且一个队列的消费者肯定不止一个。

这里博主的做法是，给最初的队列类，加一个集合类的属性，用于储存当前都有那些消费者进行消费

这么多的消费者，那他们应该怎样消费这些消息呢？

这里博主采用轮询的方式进行消费，轮着进行消费就好

以此我们再创建一个属性为 consumerSeq 记录当前取到了第几个消费者. 方便实现轮询策略.

![在这里插入图片描述][7a8a50e42fec4ab5a3269fe63e4fc18b.png]

## ?消费者类 ##

对于消费者，我们创建一个类，并给出基础属性，实现如下：

/*
     * 表示一个消费者(完整的执行环境)
     */
    public class ConsumerEnv {
        
        private String consumerTag;
        private String queueName;
        private boolean autoAck;
        // 通过这个回调来处理收到的消息.
        private Consumer consumer;
    
        public ConsumerEnv(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        
            this.consumerTag = consumerTag;
            this.queueName = queueName;
            this.autoAck = autoAck;
            this.consumer = consumer;
        }
    
        public String getConsumerTag() {
        
            return consumerTag;
        }
    
        public void setConsumerTag(String consumerTag) {
        
            this.consumerTag = consumerTag;
        }
    
        public String getQueueName() {
        
            return queueName;
        }
    
        public void setQueueName(String queueName) {
        
            this.queueName = queueName;
        }
    
        public boolean isAutoAck() {
        
            return autoAck;
        }
    
        public void setAutoAck(boolean autoAck) {
        
            this.autoAck = autoAck;
        }
    
        public Consumer getConsumer() {
        
            return consumer;
        }
    
        public void setConsumer(Consumer consumer) {
        
            this.consumer = consumer;
        }
    }

## ?消费消息的流程 ##

如果有消息需要进行消费，我们就将该队列放入 一个阻塞队列中，并用一个扫描线程对给该队列进行扫描。若有需要消费的消息队列，就将该队列取出来，交给线程池进行执行信息的回调函数。  
![在这里插入图片描述][285baa740dbd43dd810047537c5deb5f.png]  
该流程，我们创建一个类ConsumerManager进行实现这些操作：

![在这里插入图片描述][2d5a255526034ea7ac673b04a13be89c.png]  
该部分代码后面博主再进行书写

## ?如何实现消息确认呢？ ##

我们在执行回调函数之前，先将需要消费的该消息放入待确认的集合中，若执行回调函数没有发生异常与错误，我们就认为消息消费成功。实现了消息确认。

## ⭕总结 ##

关于《【消息队列开发】 实现消费者订阅消息》就讲解到这儿，感谢大家的支持，欢迎各位留言交流以及批评指正，如果文章对您有帮助或者觉得作者写的还不错可以点一下关注，点赞，收藏支持一下

[d6f59e2605d144f7a7168f33f3d7bfc5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/47dbfa7edfe74c4fbab6473a2aae5dc5.png
[36a22a51cb3646a583d9a56818151a65.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/7cf2838f569648a68602b29b091fedff.png
[7a8a50e42fec4ab5a3269fe63e4fc18b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/2d73347c06bb45e5a3ce6abe3176160b.png
[285baa740dbd43dd810047537c5deb5f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/090be57286a644c682508e63352e68ab.png
[2d5a255526034ea7ac673b04a13be89c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/e8c000c403714e3a82fa54ce9a30650a.png