ROCKETMQ-消息发送与消费（二）

站长

2024年04月14日 10:47 · 阅读数 10

本文是ROCKETMQ的第二篇，主要将介绍ROCKETMQ消息发送和消费相关关注点

一. 消息发送

核心关注点：如何可靠发送消息、如何将消息发送到broker（负载）

1.1 核心类

ROCKETMQ-消息发送与消费（二）

MQAdmin：MQ 基本的管理接口，提供对 MQ 提供基础的管理能力
MQProducer：消息发送者接口
ClientConfig:客户端配置相关
DefaultMQProducer:消息发送者默认实现类
TransactionMQProducer:事务消息发送者默认实现类

1.2 消息发送类型

同步发送

/**
同步发送，参数列表说明如下：
-   Message msg：待发送的消息对象
-   long timeout：超时时间，默认为 3s
**/
SendResult send(Message msg, long timeout)

异步发送

/**
异步消息发送，参数列表说明如下：

-   Message msg：待发送的消息
-   SendCallback sendCallback：异步发送回调接口
-   long timeout：发送超时时间，默认为 3s
**/
void send(Message msg, SendCallback sendCallback, long timeout)

Oneway 消息发送

void sendOneway(Message msg)

批量发送消息（同步）

SendResult send(Collection<Message> msgs, MessageQueue mq, long timeout)

1.3 队列选择

1.3.1 消息发送轮询策略

RoundRobin模式
- 使用范围:对于非顺序消息（普通消息、定时/延时消息、事务消息）场景，默认且只能使用RoundRobin模式的负载均衡策略。
- 策略原理：轮询方式

ROCKETMQ-消息发送与消费（二）

MessageGroupHash模式
- 使用范围:顺序消息，默认且只能使用MessageGroupHash模式的负载均衡策略
- 策略原理：Hash算法，生产者发送消息时，以消息组为粒度，按照内置的Hash算法，将相同消息组的消息分配到同一队列中，保证同一消息组的消息按照发送的先后顺序存储。

ROCKETMQ-消息发送与消费（二）

1.3.2 消息发送高可用设计与故障规避机制

针对非顺序消息保证消息发送的高可用性，在内部引入了重试机制，默认重试 2 次。RocketMQ 消息发送端采取的队列负载均衡默认采用轮循。

ROCKETMQ-消息发送与消费（二） topicA 在 broker-a、broker-b 上分别创建了 4 个队列，例如一个线程使用 Producer 发送消息时，通过对 sendWhichQueue getAndIncrement() 方法获取下一个队列。

例如在发送之前 sendWhichQueue 该值为 broker-a 的 q1，如果由于此时 broker-a 的突发流量异常大导致消息发送失败，会触发重试，按照轮循机制，下一个选择的队列为 broker-a 的 q2 队列，此次消息发送大概率还是会失败。为此引入故障规避机制：在消息重试的时候，会尽量规避上一次发送的 Broker，回到上述示例，当消息发往 broker-a q1 队列时返回发送失败，那重试的时候，会先排除 broker-a 中所有队列，即这次会选择 broker-b q1 队列，增大消息发送的成功率。

提供两种规避策略，该参数由 sendLatencyFaultEnable 控制

1.3 事务消息

消息发送与数据库事务的不一致性带来的业务出错

ROCKETMQ-消息发送与消费（二）

简单的登录实践，原文

ROCKETMQ-消息发送与消费（二）

其他思路：如果一定需要发送可靠消息，也可采用本地事务表方式：

本地事务发送MQ消息前，先记录事务表，与本地事务一起
发送MQ，如发送失败不处理，发送成功，删除本地事务表（或者修改状态）
定时任务轮询本地事务表，将未发送的消息捞取出来发送

上述可保证消息发送的最终一致性

二. 消息消费

2.1 核心类

ROCKETMQ-消息发送与消费（二）

MQConsumer：MQ消费者
MQPushConsumer：推模式消费者
MQPullConsumer：拉模式消费者
DefaultMQPushConsumer：推模式默认实现类
DefaultMQPullConsumer：取模式默认实现类

2.2 推拉模式使用示例

2.2.1 PULL拉模式

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public class PullConsumerTest {
    public static void main(String[] args) throws Exception {
        Semaphore semaphore = new Semaphore();
        Thread t = new Thread(new Task(semaphore));
        t.start();
        CountDownLatch cdh = new CountDownLatch(1);
        try {
            //程序运行 120s　后介绍
            cdh.await(120 * 1000, TimeUnit.MILLISECONDS);
        } finally {
            semaphore.running = false;
        }
    }
    /**
     * 消息拉取核心实现逻辑
     */
    static class Task implements Runnable {
        Semaphore s = new Semaphore();
        public Task(Semaphore s ) {
            this.s = s;
        }
        public void run() {
            try {
                DefaultMQPullConsumer consumer = new 
                    DefaultMQPullConsumer("dw_pull_consumer");
                consumer.setNamesrvAddr("127.0.01:9876");
                consumer.start();
                Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();
                Set<MessageQueue> msgQueueList = consumer.
                    fetchSubscribeMessageQueues("TOPIC_TEST"); // 获取该 Topic 的所有队列
                if(msgQueueList != null && !msgQueueList.isEmpty()) {
                    boolean noFoundFlag = false;
                    while(this.s.running) {
                        if(noFoundFlag) { //　没有找到消息，暂停一下消费
                            Thread.sleep(1000);
                        }
                        for( MessageQueue q : msgQueueList ) {
                            PullResult pullResult = consumer.pull(q, "*",                                          decivedPulloffset(offsetTable
                             , q, consumer) , 3000);
                            System.out.println("pullStatus:" + 
                                               pullResult.getPullStatus());
                            switch (pullResult.getPullStatus()) {
                                case FOUND:
                                    doSomething(pullResult.getMsgFoundList());
                                    break;
                                case NO_MATCHED_MSG:
                                    break;
                                case NO_NEW_MSG:
                                case OFFSET_ILLEGAL:
                                    noFoundFlag = true;
                                    break;
                                default:
                                    continue ;
                            }
                            //提交位点
                            consumer.updateConsumeOffset(q, 
                                 pullResult.getNextBeginOffset());
                        }
                        System.out.println("balacne queue is empty: " + consumer.
                              fetchMessageQueuesInBalance("TOPIC_TEST").isEmpty());
                    }
                } else {
                    System.out.println("end,because queue is enmpty");
                }
                consumer.shutdown();
                System.out.println("consumer shutdown");
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }
    }
    /** 拉取到消息后具体的处理逻辑　*/
    private static void doSomething(List<MessageExt> msgs) {
        System.out.println("本次拉取到的消息条数:" + msgs.size());
    }
    public static long decivedPulloffset(Map<MessageQueue, Long> offsetTable, 
             MessageQueue queue, DefaultMQPullConsumer consumer) throws Exception {
        long offset = consumer.fetchConsumeOffset(queue, false);
        if(offset < 0 ) {
            offset = 0;
        }
        System.out.println("offset:" + offset);
        return offset;
    }
    static class Semaphore {
        public volatile boolean running = true;
    }
}

上述针对单消费者场景，多消费者针对PULL模式就比较复杂

2.2.2 PUSH推模式

public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new 
            DefaultMQPushConsumer("dw_test_consumer_6");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TOPIC_TEST", "*");
        consumer.setAllocateMessageQueueStrategy(new 
               AllocateMessageQueueAveragelyByCircle());
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                try {
                    System.out.printf("%s Receive New Messages: %s %n", 
                          Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Throwable e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

2.3 DefaultMQPushConsumer核心概念

PUSH 模式是对 PULL 模式的封装，类似于一个高级 API，用户使用起来将非常简单，基本将消息消费所需要解决的问题都封装好了，故使用起来将变得简单。

2.3.1 `setConsumeFromWhere`

包含三个枚举类：

CONSUME_FROM_LAST_OFFSET：
- 当一个新的订阅组（Consumer Group）第一次启动时，从这个队列的最后一个偏移量（Offset）开始消费。
- 在后续的启动中，消费者会继续从上次消费的进度（即上次消费的偏移量）开始消费。
- 这种模式适用于那些只关心最新消息，而不需要处理历史消息的场景。
CONSUME_FROM_FIRST_OFFSET：
- 当一个新的订阅组第一次启动时，从这个队列的初始位置（即第一个偏移量）开始消费。
- 在后续的启动中，消费者会继续从上次消费的进度开始消费。
- 这种模式通常用于那些需要处理所有消息，包括历史消息的场景。
CONSUME_FROM_TIMESTAMP：
- 当一个新的订阅组第一次启动时，从指定的时间戳位置开始消费。
- RocketMQ 会根据这个时间戳找到对应的消息位置，并从那个位置开始消费。
- 在后续的启动中，消费者会继续从上次消费的进度开始消费。
- 这种模式允许消费者根据特定的时间条件来开始消费，适用于那些需要基于时间进行消息回溯的场景。

ConsumeFromWhere 这个参数的含义是，初次启动从何处开始消费。更准确的表述是，如果查询不到消息消费进度时，从什么地方开始消费。

2.3.2 AllocateMessageQueueStrategy 消息队列负载算法

RocketMQ 默认提供了如下负载均衡算法：

AllocateMessageQueueAveragely：平均连续分配算法。
AllocateMessageQueueAveragelyByCircle：平均轮流分配算法。
AllocateMachineRoomNearby：机房内优先就近分配。
AllocateMessageQueueByConfig：手动指定，这个通常需要配合配置中心，在消费者启动时，首先先创建 AllocateMessageQueueByConfig 对象，然后根据配置中心的配置，再根据当前的队列信息，进行分配，即该方法不具备队列的自动负载，在 Broker 端进行队列扩容时，无法自动感知，需要手动变更配置。
AllocateMessageQueueByMachineRoom：消费指定机房中的队列，该分配算法首先需要调用该策略的 setConsumeridcs(Set<String> consumerIdCs) 方法，用于设置需要消费的机房，将刷选出来的消息按平均连续分配算法进行队列负载。

2.3.3 OffsetStore 消息进度存储管理器

RocketMQ 在广播消息、集群消费两种模式下消息消费进度的存储策略会有所不同。

集群模式：RocketMQ 会将消息消费进度存储在 Broker 服务器，存储路径为 ${ROCKET_HOME}/store/config/ consumerOffset.json 文件中。
广播模式：RocketMQ 会将消息消费进存储在消费端所在的机器上，存储路径为 ${user.home}/.rocketmq_offsets 中。

2.3.4 其他参数

consumeThreadMin：消费者每一个消费组线程池中最小的线程数量，默认为 20。
consumeThreadMax：最大线程数，因为队列是无界队列，该参数没啥意义
consumeConcurrentlyMaxSpan：并发消息消费时处理队列中最大偏移量与最小偏移量的差值的阔值，如差值超过该值，触发消费端限流。默认值2000
pullThresholdForQueue：消费端允许消费端端单队列积压的消息数量，如果处理队列中超过该值，会触发消息消费端的限流。默认值1000
pullThresholdSizeForQueue：消费端允许消费端但队列中挤压的消息体大小，默认为 100MB
pullThresholdForTopic：按 Topic 级别进行消息数量限流，默认不开启，为 -1
pullThresholdSizeForTopic：按 Topic 级别进行消息消息体大小进行限流，默认不开启
pullInterval：消息拉取的间隔，默认 0 表示，消息客户端在拉取一批消息提交到线程池后立即向服务端拉取下一批
pullBatchSize：一次消息拉取请求最多从 Broker 返回的消息条数，默认为 32
consumeMessageBatchMaxSize：消息消费一次最大消费的消息条数
maxReconsumeTimes：消息消费重试次数，并发消费模式下默认重试 16 次后进入到死信队列，如果是顺序消费，重试次数为 Integer.MAX_VALUE。
suspendCurrentQueueTimeMillis：消费模式为顺序消费时设置每一次重试的间隔时间，提高重试成功率
consumeTimeout：消息消费超时时间

2.3.5 消息消费进度提交

ROCKETMQ-消息发送与消费（二）向broker提交消息进度时，提交的是取 ProceeQueue 中最小的偏移量为消息消费进度，那样可能会导致重复消费

原因是 msg 1~5，msg3 msg4和msg1消费完，msg2 msg5没有消费完，此时往broker提交msg2，如果此时client挂掉，重启下次从msg2开始消费，msg3 msg4会重复消费

2.4 DefaultLitePullConsumer 核心概念

DefaultMQPullConsumer（PULL 模式）的 API 太底层，使用起来及其不方便，RocketMQ 官方设计者也注意到这个问题，为此在 RocketMQ 4.6.0 版本中引入了 PULL 模式的另外一个实现类 DefaultLitePullConsumer

2.4.1 核心UML图

ROCKETMQ-消息发送与消费（二）

2.4.2 使用举例

public class LitePullConsumerSubscribe02 {
    public static volatile boolean running = true;
    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer litePullConsumer = new 
            DefaultLitePullConsumer("dw_lite_pull_consumer_test");
        litePullConsumer.setNamesrvAddr("192.168.3.166:9876");
        litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        litePullConsumer.subscribe("TopicTest", "*");
        litePullConsumer.setAutoCommit(true); //该值默认为 true
        litePullConsumer.start();
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                doConsumeSomething(messageExts);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
    private static void doConsumeSomething(List<MessageExt> messageExts) {
        // 真正的业务处理
        System.out.printf("%s%n", messageExts);
    }
}

注意参数pullThreadNums,消息拉取线程数量，默认为 20 个，注意这个是每一个消费者默认 20 个线程往 Broker 拉取消息。这个应该是 Lite PULL 模式对比 PUSH 模式一个非常大的优势。

Lite Pull流程如下：

ROCKETMQ-消息发送与消费（二）

使用场景参考：12 结合实际场景再聊 DefaultLitePullConsumer 的使用

三. 顺序消息

3.1 发送消息发往同一个队列

ROCKETMQ-消息发送与消费（二）

3.2 消费消息，同一队列需要顺序消费

ROCKETMQ-消息发送与消费（二）

3.3. 出现问题

保证消费端对单队列中的消息顺序处理，故多线程处理，需要按照消息消费队列进行加锁。

消费端的横向扩容或 Broker 端队列个数的变更都会触发消息消费队列的重新负载，在并发消息时在队列负载的时候一个消费队列有可能被多个消费者同时消息，但顺序消费时并不会出现这种情况，因为顺序消息不仅仅在消费消息时会锁定消息消费队列，在分配到消息队列时，能从该队列拉取消息还需要在 Broker 端申请该消费队列的锁，即同一个时间只有一个消费者能拉取该队列中的消息，确保顺序消费的语义。

从前面的文章中也介绍到并发消费模式在消费失败时有重试机制，默认重试 16 次，而且重试时是先将消息发送到 Broker，然后再次拉取到消息，这种机制就会丧失其消费的顺序性，故如果是顺序消费模式，消息重试时在消费端不停的重试，重试次数为 Integer.MAX_VALUE，即如果一条消息如果一直不能消费成功，其消息消费进度就会一直无法向前推进，即会造成消息积压现象。

参考资料生产者负载均衡

转载自:https://juejin.cn/post/7347125955649896499

ROCKETMQ-消息发送与消费（二）

一. 消息发送

1.1 核心类

1.2 消息发送类型

1.3 队列选择

1.3.1 消息发送轮询策略

1.3.2 消息发送高可用设计与故障规避机制

1.3 事务消息

二. 消息消费

2.1 核心类

2.2 推拉模式使用示例

2.2.1 PULL拉模式

2.2.2 PUSH推模式

2.3 DefaultMQPushConsumer核心概念

2.3.1 setConsumeFromWhere

2.3.2 AllocateMessageQueueStrategy 消息队列负载算法

2.3.3 OffsetStore 消息进度存储管理器

2.3.4 其他参数

2.3.5 消息消费进度提交

2.4 DefaultLitePullConsumer 核心概念

2.4.1 核心UML图

2.4.2 使用举例

三. 顺序消息

3.1 发送消息发往同一个队列

3.2 消费消息，同一队列需要顺序消费

3.3. 出现问题

2.3.1 `setConsumeFromWhere`