首页 > 资讯 > 后端开发 > Python >RocketMQ消息拉取过程详解

205

分享到

RocketMQ消息拉取过程详解

RocketMQ消息拉取 RocketMQ消息拉取 2022-12-15 18:12:01 205人浏览泡泡鱼

Python 官方文档：入门教程 => 点击学习

摘要

目录前言DefaultLitePullConsumer拉消息代码示例消息消费消息拉取入口PullTaskImpl拉消息前言在上一篇文章中，我们讲述了DefaultMQPushCon

前言

在上一篇文章中，我们讲述了DefaultMQPushConsumer拉消息的原理，它是通过重平衡触发pullRequest的创建，通过阻塞队列作为pullRequest的存储容器，另一端通过定时任务从阻塞队列中取出pullRequest来向Broker发送拉消息的请求，无论消息拉取成功还是失败，都会重新把pullRequest放回阻塞队列中，这样就能保证持续不断地向Broker拉消息了；

今天这篇文章我们继续讲述DefaultLitePullConsumer是如何实现消息拉取的；

DefaultLitePullConsumer拉消息代码示例

我们在使用DefaultLitePullConsumer时都是主动去poll消息，并不是像DefaultMQPushConsumer那样设置一个消息监听器：

DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(group);
consumer.setNamesrvAddr(nameSrv);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe(topic, subExpression);
try {
    consumer.start();
} catch (Exception e) {
    e.printStackTrace();
}
try {
    while (true) {
        List<MessageExt> messageExts = consumer.poll(5000);
        // 处理业务逻辑
        System.out.println("消息数量：" + messageExts.size());
        System.out.println("消息内容：");
        for (MessageExt messageExt : messageExts) {
            System.out.println(new String(messageExt.getBody(), StandardCharsets.UTF_8));
        }
    }
} catch (Exception e) {
    e.printStackTrace();
}

一般拿到消息后都会交给业务线程池去处理，上述代码我只简单地打印了一下消息内容；

消息消费

跟着poll()方法，我们最终定位到DefaultLitePullConsumerImpl.poll()这个方法：

    public synchronized List<MessageExt> poll(long timeout) {
        try {
            this.checkServiceState();
            if (timeout < 0L) {
                throw new IllegalArgumentException("Timeout must not be negative");
            }

            if (this.defaultLitePullConsumer.isAutoCommit()) {
                this.maybeAutoCommit();
            }

            long endTime = System.currentTimeMillis() + timeout;
            // 从阻塞队列中取ConsumeRequest
            DefaultLitePullConsumerImpl.ConsumeRequest consumeRequest = (DefaultLitePullConsumerImpl.ConsumeRequest)this.consumeRequestCache.poll(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
            if (endTime - System.currentTimeMillis() > 0L) {
                while(consumeRequest != null && consumeRequest.getProcessQueue().isDropped()) {
                    consumeRequest = (DefaultLitePullConsumerImpl.ConsumeRequest)this.consumeRequestCache.poll(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                    if (endTime - System.currentTimeMillis() <= 0L) {
                        break;
                    }
                }
            }

            if (consumeRequest != null && !consumeRequest.getProcessQueue().isDropped()) {
                List<MessageExt> messages = consumeRequest.getMessageExts();
                long offset = consumeRequest.getProcessQueue().removeMessage(messages);
              // 取到消息后直接更新消费点位
                this.assignedMessageQueue.updateConsumeOffset(consumeRequest.getMessageQueue(), offset);
                this.resetTopic(messages);
                // 下面是调用consumeMessageHook
                if (!this.consumeMessageHookList.isEmpty()) {
                    ConsumeMessageContext consumeMessageContext = new ConsumeMessageContext();
                    consumeMessageContext.setNamespace(this.defaultLitePullConsumer.getNamespace());
                    consumeMessageContext.setConsumerGroup(this.groupName());
                    consumeMessageContext.setMq(consumeRequest.getMessageQueue());
                    consumeMessageContext.setMsgList(messages);
                    consumeMessageContext.setSuccess(false);
                    this.executeHookBefore(consumeMessageContext);
                  // 默认是消费成功
                    consumeMessageContext.setStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS.toString());
                    consumeMessageContext.setSuccess(true);
                    this.executeHookAfter(consumeMessageContext);
                }

                return messages;
            }
        } catch (InterruptedException var10) {
        }

        return Collections.emptyList();
    }

1.直接从阻塞队列consumeRequestCache中取出消息对象ConsumeRequest，这里面就包含了消息内容；

2.取出来后直接更新消费点位，默认为此次消息消费成功；

这里跟DefaultMQPushConsumer不同的是，DefaultLitePullConsumerImpl.poll()默认的是消息消费一定成功，如果消费失败的话，需要开发人员自己处理，消费失败的消息不会再次发送给消费者；

那么咱们的疑问就出来了，poll()方法光顾着从consumeRequestCache中取消息，那消息是啥时候放进去的呢？

消息拉取入口

我们可以重新了解一下消费者重平衡过程，在MessageQueue分配完毕后，会对比被分配的MessageQueue是否和分配前的不一致，大部分情况下是会发生改变的，那么就会触发messageQueueChanged()方法的调用：

    @Override
    public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
        // 取出所有的MessageQueueListener
        MessageQueueListener messageQueueListener = this.litePullConsumerImpl.getDefaultLitePullConsumer().getMessageQueueListener();
        if (messageQueueListener != null) {
            try {
                // 依次调用messageQueueChanged方法
                messageQueueListener.messageQueueChanged(topic, mqAll, mqDivided);
            } catch (Throwable e) {
                log.error("messageQueueChanged exception", e);
            }
        }
    }

MessageQueueListener是什么时候被放进去的呢？可以看一下subscribe()方法：

    public synchronized void subscribe(String topic, String subExpression) throws MQClientException {
        try {
            if (topic == null || "".equals(topic)) {
                throw new IllegalArgumentException("Topic can not be null or empty.");
            }
            setSubscriptionType(SubscriptionType.SUBSCRIBE);
            SubscriptionData subscriptionData = Filterapi.buildSubscriptionData(topic, subExpression);
            this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
            // 每个subscribe()都会设置MessageQueueListener
            this.defaultLitePullConsumer.setMessageQueueListener(new MessageQueueListenerImpl());
            assignedMessageQueue.setRebalanceImpl(this.rebalanceImpl);
            if (serviceState == ServiceState.RUNNING) {
                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
                updateTopicSubscribeInfoWhenSubscriptionChanged();
            }
        } catch (Exception e) {
            throw new MQClientException("subscribe exception", e);
        }
    }

可以发现，每个subscribe()都会设置MessageQueueListener，MessageQueueListenerImpl里面只干了一件事情：更新MessageQueue并且创建pullTask；

    class MessageQueueListenerImpl implements MessageQueueListener {
        @Override
        public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
            updateAssignQueueAndStartPullTask(topic, mqAll, mqDivided);
        }
    }

    public void updateAssignQueueAndStartPullTask(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
        MessageModel messageModel = defaultLitePullConsumer.getMessageModel();
        switch (messageModel) {
            case BROADCASTING:
                updateAssignedMessageQueue(topic, mqAll);
                // 更新拉消息任务
                updatePullTask(topic, mqAll);
                break;
            case CLUSTERING:
                updateAssignedMessageQueue(topic, mqDivided);
                // 更新拉消息任务
                updatePullTask(topic, mqDivided);
                break;
            default:
                break;
        }
    }

现在终于快找到这个消息拉取的入口了：

    private void startPullTask(Collection<MessageQueue> mqSet) {
        for (MessageQueue messageQueue : mqSet) {
            if (!this.taskTable.containsKey(messageQueue)) {
                // 创建消息拉取任务
                PullTaskImpl pullTask = new PullTaskImpl(messageQueue);
                this.taskTable.put(messageQueue, pullTask);
                // 这个就是任务执行的入口
                this.scheduledThreadPoolExecutor.schedule(pullTask, 0, TimeUnit.MILLISECONDS);
            }
        }
    }

消息拉取的入口寻找起来还是有点困难的，但是主要思路还是从【重平衡】开始，另外就是触发了MessageQueueListener，此时才会创建pullTask；

虽然DefaultMQPushConsumer也是【重平衡】触发pullRequest的创建，但是它是将pullRequest放进阻塞队列，另一端由消息拉取任务去取pullRequest向Broker发送请求；而DefaultLitePullConsumer是直接创建pullTask去拉消息；

PullTaskImpl拉消息

很显然，PullTaskImpl就是一个Runnable，那么最重要的就是它的run()方法，这个方法就是负责从Broker拉消息并放进consumeRequestCache阻塞队列中，这样poll()方法才能从consumeRequestCache阻塞队列中取到消息；

messageQueue暂停

if (DefaultLitePullConsumerImpl.this.assignedMessageQueue.isPaused(this.messageQueue)) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 1000L, TimeUnit.MILLISECONDS);
    DefaultLitePullConsumerImpl.this.log.debug("Message Queue: {} has been paused!", this.messageQueue);
                    return;
}

如果messageQueue处于暂停状态，那么延迟1秒重新执行这个任务；

ProcessQueue被移除

ProcessQueue processQueue = DefaultLitePullConsumerImpl.this.assignedMessageQueue.getProcessQueue(this.messageQueue);
if (null == processQueue || processQueue.isDropped()) {
    DefaultLitePullConsumerImpl.this.log.info("The message queue not be able to poll, because it's dropped. group={}, messageQueue={}", DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getConsumerGroup(), this.messageQueue);
    return;
}

如果processQueue不存在或者已经被移除了，那么这个任务也不用执行了；

流量控制

if ((long)DefaultLitePullConsumerImpl.this.consumeRequestCache.size() * (long)DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullBatchSize() > DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullThresholdForAll()) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 50L, TimeUnit.MILLISECONDS);
    if (DefaultLitePullConsumerImpl.this.consumeRequestFlowControlTimes++ % 1000L == 0L) {
        DefaultLitePullConsumerImpl.this.log.warn("The consume request count exceeds threshold {}, so do flow control, consume request count={}, flowControlTimes={}", DefaultLitePullConsumerImpl.this.consumeRequestCache.size(), DefaultLitePullConsumerImpl.this.consumeRequestFlowControlTimes);
    }
    return;
}

如果consumeRequestCache中的消息数量超过了PullThresholdForAll阈值，那么触发限流机制，当前任务将不会继续拉消息，并且50毫秒后才会重新执行该任务；

long cachedMessageCount = processQueue.getMsGCount().get();
long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / 1048576L;
// 单个processQueue上面消息数量限制
if (cachedMessageCount > (long)DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullThresholdForQueue()) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 50L, TimeUnit.MILLISECONDS);
    if (DefaultLitePullConsumerImpl.this.queueFlowControlTimes++ % 1000L == 0L) {
        DefaultLitePullConsumerImpl.this.log.warn("The cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, flowControlTimes={}", new Object[]{DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firsTKEy(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, DefaultLitePullConsumerImpl.this.queueFlowControlTimes});
    }
    return;
}
// 单个processQueue中消息总大小限制
if (cachedMessageSizeInMiB > (long)DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullThresholdSizeForQueue()) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 50L, TimeUnit.MILLISECONDS);
    if (DefaultLitePullConsumerImpl.this.queueFlowControlTimes++ % 1000L == 0L) {
        DefaultLitePullConsumerImpl.this.log.warn("The cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, flowControlTimes={}", new Object[]{DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, DefaultLitePullConsumerImpl.this.queueFlowControlTimes});
    }
    return;
}

如果当前processQueue中消息的数量大于PullThresholdForQueue阈值，也同样触发限流机制，当前任务不再执行，50毫秒后重新执行该任务；
如果当前processQueue中消息的总大小超过PullThresholdSizeForQueue（单位：MB）阈值，将触发限流机制，当前任务不再执行，50毫秒后重新执行该任务；

if (processQueue.getMaxSpan() > (long)DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getConsumeMaxSpan()) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 50L, TimeUnit.MILLISECONDS);
    if (DefaultLitePullConsumerImpl.this.queueMaxSpanFlowControlTimes++ % 1000L == 0L) {
        DefaultLitePullConsumerImpl.this.log.warn("The queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, flowControlTimes={}", new Object[]{processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(), DefaultLitePullConsumerImpl.this.queueMaxSpanFlowControlTimes});
    }
    return;
}

如果processQueue中的maxSpan大于消费者的ConsumeMaxSpan，也就是第一个消息与最后一个消息的点位偏差大于ConsumeMaxSpan（默认是2000），将触发限流机制，当前任务不执行，50毫秒后重新执行该任务；

计算拉取点位

long offset = 0L;
try {
    offset = DefaultLitePullConsumerImpl.this.nextPullOffset(this.messageQueue);
} catch (Exception var17) {
    DefaultLitePullConsumerImpl.this.log.error("Failed to get next pull offset", var17);
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, 3000L, TimeUnit.MILLISECONDS);
    return;
}

计算消息拉取的点位，如果产生异常，那么简隔3秒后再来重新开始任务；

拉消息

PullResult pullResult = DefaultLitePullConsumerImpl.this.pull(this.messageQueue, subscriptionData, offset, DefaultLitePullConsumerImpl.this.defaultLitePullConsumer.getPullBatchSize());

这个就是发请求给Broker拉消息；

放进消息缓存区

switch(pullResult.getPullStatus()) {
    case FOUND:
        Object objLock = DefaultLitePullConsumerImpl.this.messageQueueLock.fetchLockObject(this.messageQueue);
        synchronized(objLock) {
            if (pullResult.getMsgFoundList() != null && !pullResult.getMsgFoundList().isEmpty() && DefaultLitePullConsumerImpl.this.assignedMessageQueue.getSeekOffset(this.messageQueue) == -1L) {
                processQueue.putMessage(pullResult.getMsgFoundList());
                  DefaultLitePullConsumerImpl.this.submitConsumeRequest(DefaultLitePullConsumerImpl.this.new ConsumeRequest(pullResult.getMsgFoundList(), this.messageQueue, processQueue));
     }
     break;
}

找到消息的情况下，将调用submitConsumeRequest()方法把消息放进阻塞队列中，等待poll()方法来消费；

重新开启拉取任务

if (!this.isCancelled()) {
    DefaultLitePullConsumerImpl.this.scheduledThreadPoolExecutor.schedule(this, pullDelayTimeMills, TimeUnit.MILLISECONDS);
} else {
    DefaultLitePullConsumerImpl.this.log.warn("The Pull Task is cancelled after doPullTask, {}", this.messageQueue);
}

如果当前任务还没有被取消的话，那么重新开启下一个轮回，准备下一次消息拉取；

以上就是RocketMQ消息拉取过程详解的详细内容，更多关于RocketMQ消息拉取过程的资料请关注编程网其它相关文章！

您可能感兴趣的文档:

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: RocketMQ消息拉取过程详解

本文链接: https://www.lsjlt.com/news/174991.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

RocketMQ消息拉取过程详解

目录前言DefaultLitePullConsumer拉消息代码示例消息消费消息拉取入口PullTaskImpl拉消息前言在上一篇文章中，我们讲述了DefaultMQPushCon...

99+

2022-12-15

RocketMQ消息拉取 RocketMQ消息拉取
RocketMQ 事务消息详解

🍊 Java学习：Java从入门到精通总结 🍊 深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想 🍊 绝对不一样的职场干货：大厂最佳实践经验指南 📆 最近更新：2023年4月9日 🍊 个人简介：通信工程本...

99+

2023-08-17

java-rocketmq rocketmq java
RocketMQ生产消息与消费消息超详细讲解

目录1 RocketMQ简介2 MQ的常见产品3 环境搭建4 单生产者单消费者模式5 单生产者多消费者模式5.1默认模式（负载均衡）5.2广播模式6 多生产者多消费者模式1 Rock...

99+

2022-12-27

RocketMQ生产消息 RocketMQ消费消息
Springboot详解RocketMQ实现广播消息流程

RocketMQ消息模式主要有两种：广播模式、集群模式（负载均衡模式）广播模式是每个消费者，都会消费消息；负载均衡模式是每一个消费只会被某一个消费者消费一次；我们业务上一般用的...

99+

2022-11-13
详解RocketMQ消费端如何监听消息

目录前言流程地图源码跟踪核心模块（消息拉取）拉取流程拉取消息处理当pullStatus为FOUND，消息进行提交消费的请求消息消费进度提交总结前言上一篇文章中我们主要来看Rock...

99+

2022-12-15

RocketMQ 消费端监听消息 RocketMQ 消息监听
RocketMQ延迟消息超详细讲解

目录一、什么是延时消息二、延时消息等级三、延时消息使用场景四、延时消息示例五、延时消息实现原理一、什么是延时消息当消息写入到Broker后，不能立刻被消费者消费，需要等待指定的时长...

99+

2023-02-14

RocketMQ延迟消息 RocketMQ延迟消息实现原理
RocketMQ普通消息实战演练详解

目录引言普通消息同步发送普通消息异步发送普通消息单向发送集群消费模式广播消费模式引言之前研究了RocketMQ的源码，在这里将各种消息发送与消费的demo进行举例，方便以后使用的时...

99+

2022-11-13
分布式消息队列RocketMQ概念详解

目录1.MQ概述1.1 RocketMQ简介1.2 MQ用途1.3 常见MQ产品2.RocketMQ 基本概念2.1 消息2.2 主题2.3 标签2.4 队列2.5 Producer...

99+

2023-05-19

分布式消息队列RocketMQ概念消息队列RocketMQ RocketMQ概念
RocketMQ事务消息原理与使用详解

目录一、RocketMQ事务消息概要二、RocketMQ事务消息使用案例(1)、定义消息监听器(2)、定义消息生产者(3)、定义消息消费者(4)、观察生产者控制台输出(5)、观察消费...

99+

2023-02-13

RocketMQ事务消息 RocketMQ事务消息原理 RocketMQ事务消息使用
Springboot详解RocketMQ实现消息发送与接收流程

springboot+rockermq 实现简单的消息发送与接收普通消息的发送方式有3种：单向发送、同步发送和异步发送。下面来介绍下 springboot+rockermq 整合...

99+

2022-11-13
Springboot 整合 RocketMQ 收发消息的配置过程

Springboot 整合 RocketMQ 收发消息创建springboot项目 pom.xml添加rocketmq-spring-boot-starter依赖。 <d...

99+

2022-11-12
Springboot详细讲解RocketMQ实现顺序消息的发送与消费流程

目录一、创建Springboot项目添加rockermq依赖二、配置rocketmq三、新建一个controller来做消息发送四、创建消费端监听消息消费消息五、启动服务测试顺序消息...

99+

2022-11-13
RocketMq消息重试机制及死信队列详解

目录生产者消息重试消费者消息重试并发消费顺序消费并发消费和顺序消费区别死信队列实践出真知公共部分创建测试并发消费并发消费状态测试顺序消费顺序消费状态测试死信队列死信队列特性生产者消息...

99+

2022-11-13
java开发RocketMQ消息中间件原理基础详解

RocketMQ 是什么 Github 上关于 RocketMQ 的介绍： RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：支持发布/...

99+

2022-11-12
RocketMQ Broker消息处理流程剩余源码解析

🍊 Java学习：Java从入门到精通总结 🍊 深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想 🍊 绝对不一样的职场干货：大厂最佳实践经验指南 📆 最近更新：2023...

99+

2023-09-03

java-rocketmq rocketmq java
Android消息机制Handler的工作过程详解

综述　　在Android系统中，出于对性能优化的考虑，对于Android的UI操作并不是线程安全的。也就是说若是有多个线程来操作UI组件，就会有可能导致线程安全问题。所以在A...

99+

2022-06-06

handler Android
Asp.net core 使用SignalR推送消息过程详解

1）、SignalR简介 ASP.NET Core SignalR 是为 ASP.NET 开发人员提供的一个库，可以简化开发人员将实时 Web 功能添加到应用程序的过程。实时 We...

99+

2022-11-13
RocketMQ broker 消息投递流程处理PULL_MESSAGE请求解析

目录RocketMq消息处理1. 处理PULL_MESSAGE请求2. 获取消息3. 挂起请求：PullRequestHoldService#suspendPullRequest3....

99+

2023-05-14

RocketMQ broker 消息投递 PULL_MESSAGE 请求处理
RocketMQ Broker消息处理流程及部分源码解析

🍊 Java学习：Java从入门到精通总结 🍊 深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想 🍊 绝对不一样的职场干货：大厂最佳实践经验指南 📆 最近更新：2023...

99+

2023-09-06

java-rocketmq rocketmq java
详解RocketMQ中的消费者启动与消费流程分析

目录一、简介1.1 RocketMQ 简介1.2 工作流程二、消费者启动流程2.1 实例化消费者2.2 设置NameServer和订阅topic过程2.2.1 添加tag2.2.2 ...

99+

2022-11-13