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

news2025/11/1 2:06:30

文章目录

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

🍃前言

本次开发任务

实现消费者订阅消息

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

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

在这里插入图片描述

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

在这里插入图片描述

回调函数实现代码如下：

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

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

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

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

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

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

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

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

在这里插入图片描述

🎍消费者类

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

/*
 * 表示一个消费者(完整的执行环境)
 */
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;
    }
}