RocketMQ源码剖析

作者: 王侦 | 来源:发表于2023-02-15 19:11 被阅读0次

1.各个组件启动源码、框架结构

1.1 NameServer启动

NamesrvStartup#main

1）NamesrvController controller = createNamesrvController(args);创建controller
1-1）检测命令行参数
1-2）创建两个核心配置：NamesrvConfig和NettyServerConfig
1-3）解析-c 和 -p 参数，赋值到上面两个配置中
1-4）ROCKETMQ_HOME环境变量检测
1-5）controller = new NamesrvController(namesrvConfig, nettyServerConfig);创建controller
1-6） controller.getConfiguration().registerConfig(properties);注册一下所有的配置
2）start(controller);启动controller
2-1）controller.initialize();初始化
A）this.kvConfigManager.load();加载KV配置
B）this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);创建NettyServer网络处理对象
C） this.remotingExecutor = Executors.newFixedThreadPool()Netty服务器的工作线程池
D）this.registerProcessor();注册NameServer的Processor 注册到RemotingServer中。
E）NamesrvController.this.routeInfoManager.scanNotActiveBroker();定时任务：每间隔10S扫描一次Broker，移除不活跃的Broker
F）NamesrvController.this.kvConfigManager.printAllPeriodically();定时任务：每间隔10min打印一次KV配置
2-2）Runtime.getRuntime().addShutdownHook()服务关闭钩子，在服务正常关闭时执行。
2-3） controller.start();启动服务
A）this.remotingServer.start();启动remotingServer。
B）不为null也启动this.fileWatchService.start();

NameServer的核心作用：

一是维护Broker的服务地址并进行及时的更新。
二是给Producer和Consumer提供服务获取Broker列表。

1.2 Broker启动

BrokerStartup#main

1）createBrokerController(args)创建controller
1-1）检测命令行参数
1-2）创建四个核心配置：BrokerConfig、NettyServerConfig、NettyClientConfig和MessageStoreConfig
1-3）解析-c 参数，赋值到上面四个配置中
1-4）ROCKETMQ_HOME环境变量检测
1-5）处理NamesrcAddr
1-6）通过brokerId判断主从：ASYNC_MASTER、SYNC_MASTER、SLAVE；Dledger集群的所有Broker节点ID都是-1
1-7）解析-p和-m参数，赋值到上面四个配置中
1-8） controller = new BrokerController()，创建核心的BrokerController
1-9）controller.getConfiguration().registerConfig(properties);注册配置
1-10）controller.initialize();初始化BrokerController。
A）加载磁盘上的配置信息（json）。topicConfigManager、consumerOffsetManager、subscriptionGroupManager、consumerFilterManager
B）构建消息存储管理组件DefaultMessageStore，外层还会包装插件AbstractPluginMessageStore
C）加载磁盘文件this.messageStore.load();
D）this.remotingServer = new NettyRemotingServer()
E）this.fastRemotingServer = new NettyRemotingServer()这个fastRemotingServer与RemotingServer功能基本差不多，处理VIP端口请求
F） this.sendMessageExecutor发送消息的线程池
G） this.pullMessageExecutor处理consumer的pull请求的线程池
H） this.replyMessageExecutor回复消息的线程池
I）this.queryMessageExecutor查询消息的线程池
J）this.adminBrokerExecutor
K）this.clientManageExecutor管理客户端的线程池
L） this.heartbeatExecutor心跳请求线程池
M）this.endTransactionExecutor
N） this.consumerManageExecutor
O）this.registerProcessor();Broker注册Processor
P）BrokerController.this.getBrokerStats().record();定时进行broker统计的任务
Q）BrokerController.this.consumerOffsetManager.persist();定时进行consumer消费Offset持久化到磁盘的任务
R）BrokerController.this.consumerFilterManager.persist();对consumer的filter过滤器进行持久化的任务。这里可以看到，消费者的filter是被下推到了Broker来执行的。
S）BrokerController.this.protectBroker();定时进行broker保护
T）BrokerController.this.printWaterMark();定时打印水位线 U）BrokerController.this.getMessageStore().dispatchBehindByte
s()定时进行落后commitlog分发的任务
V）this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());或者BrokerController.this.brokerOuterAPI.fetchNameServerAddr();设置NameServer的地址列表。可以从配置加载，也可以发远程请求加载
W）initialTransaction();
X）initialAcl(); 权限控制
Y）initialRpcHooks();
1-11）Runtime.getRuntime().addShutdownHook()服务关闭钩子，在服务正常关闭时执行。
2）start()启动controller
A）this.messageStore.start();存储组件，这里启动服务主要是为了将CommitLog的写入事件分发给ComsumeQueue和IndexFile
B）this.remotingServer.start();以及this.fastRemotingServer.start();
C）this.fileWatchService.start();
D） this.brokerOuterAPI.start();Broker的brokerOuterAPI可以理解为一个Netty客户端，往外发请求的组件。例如发送心跳
E）this.pullRequestHoldService.start();长轮询请求暂存服务
F）this.clientHousekeepingService.start();
G）this.filterServerManager.start();
H）BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());Broker核心的心跳注册任务
I）this.brokerStatsManager.start();
J） this.brokerFastFailure.start();

1.3 Netty网络框架

Broker端的NettyRemotingServer：

NettyRemotingServer#start

    public void start() {
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyServerConfig.getServerWorkerThreads(),
            new ThreadFactory() {

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                }
            });

        prepareSharableHandlers();
        //K1 Netty服务启动的核心流程
        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        //Netty的核心服务流程，encoder和decoder，二进制传输协议。
                        //RocketMQ中的二进制传输协议比较复杂，是否能按照JSON自定义二进制协议？
                        //serverHandler负责最关键的网络请求处理。
                        ch.pipeline()
                            .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
                            .addLast(defaultEventExecutorGroup,
                                encoder,
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                connectionManageHandler,
                                serverHandler
                            );
                    }
                });

        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }
        //开始Socket监听
        try {
            ChannelFuture sync = this.serverBootstrap.bind().sync();
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            this.port = addr.getPort();
        } catch (InterruptedException e1) {
            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
        }

        if (this.channelEventListener != null) {
            this.nettyEventExecutor.start();
        }
        //每秒清理过期的异步请求暂存结果。
        this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
    }

核心是NettyServerHandler

    @ChannelHandler.Sharable
    class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
            processMessageReceived(ctx, msg);
        }
    }

    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
        final RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());

核心的业务处理是在processorTable，其初始化是在BrokerController#registerProcessor

    public void registerProcessor() {
        /**
         * SendMessageProcessor
         */
        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
        sendProcessor.registerSendMessageHook(sendMessageHookList);
        sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        /**
         * PullMessageProcessor
         */
        this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
        this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);

所以关键的业务处理，后面只要去processorTable查找类型的对应Processor就行。

RocketMQ的同步结果推送与异步结果推送
RocketMQ的RemotingServer服务端，会维护一个responseTable，这是一个线程同步的Map结构。 key为请求的ID，value是异步的消息结果。ConcurrentMap<Integer , ResponseFuture> 。

处理同步请求(NettyRemotingAbstract#invokeSyncImpl)时，处理的结果会存入responseTable，通过ResponseFuture提供一定的服务端异步处理支持，提升服务端的吞吐量。请求返回后，立即从responseTable中移除请求记录。

处理异步请求(NettyRemotingAbstract#invokeAsyncImpl)时，处理的结果依然会存入responsTable，等待客户端后续再来请求结果。但是他保存的依然是一个ResponseFuture，也就是在客户端请求结果时再去获取真正的结果。另外，在RemotingServer启动时，会启动一个定时的线程任务，不断扫描responseTable，将其中过期的response清除掉。

1.4 Broker心跳注册过程

BrokerController#start

        //K2 Broker核心的心跳注册任务，需要深入解读下。
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
                } catch (Throwable e) {
                    log.error("registerBrokerAll Exception", e);
                }
            }
        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

    public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
        ...
        //这里才是比较关键的地方。先判断是否需要注册，然后调用doRegisterBrokerAll方法真正去注册。
        if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
            this.getBrokerAddr(),
            this.brokerConfig.getBrokerName(),
            this.brokerConfig.getBrokerId(),
            this.brokerConfig.getRegisterBrokerTimeoutMills())) {
            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
        }
    }

    //K2 Broker注册最核心的部分
    private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
        TopicConfigSerializeWrapper topicConfigWrapper) {
        List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
            this.brokerConfig.getBrokerClusterName(),
            this.getBrokerAddr(),
            this.brokerConfig.getBrokerName(),
            this.brokerConfig.getBrokerId(),
            this.getHAServerAddr(),
            topicConfigWrapper,
            this.filterServerManager.buildNewFilterServerList(),
            oneway,
            this.brokerConfig.getRegisterBrokerTimeoutMills(),
            this.brokerConfig.isCompressedRegister());

        if (registerBrokerResultList.size() > 0) {
            RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
            if (registerBrokerResult != null) {
                //注册完保存主从节点的地址
                if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
                    this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
                }

                this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());

                if (checkOrderConfig) {
                    this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
                }
            }
        }
    }

会向所有NameServer进行注册：

    public List<RegisterBrokerResult> registerBrokerAll(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId,
        final String haServerAddr,
        final TopicConfigSerializeWrapper topicConfigWrapper,
        final List<String> filterServerList,
        final boolean oneway,
        final int timeoutMills,
        final boolean compressed) {
        //使用CopyOnWriteArrayList提升并发安全性
        final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();
        List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
        if (nameServerAddressList != null && nameServerAddressList.size() > 0) {

            final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
            requestHeader.setBrokerAddr(brokerAddr);
            requestHeader.setBrokerId(brokerId);
            requestHeader.setBrokerName(brokerName);
            requestHeader.setClusterName(clusterName);
            requestHeader.setHaServerAddr(haServerAddr);
            requestHeader.setCompressed(compressed);

            RegisterBrokerBody requestBody = new RegisterBrokerBody();
            requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
            requestBody.setFilterServerList(filterServerList);
            final byte[] body = requestBody.encode(compressed);
            final int bodyCrc32 = UtilAll.crc32(body);
            requestHeader.setBodyCrc32(bodyCrc32);
            //通过CountDownLatch，保证在所有NameServer上完成注册后再一起结束。
            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
            for (final String namesrvAddr : nameServerAddressList) {
                brokerOuterExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            RegisterBrokerResult result = registerBroker(namesrvAddr, oneway, timeoutMills, requestHeader, body);
                            if (result != null) {
                                registerBrokerResultList.add(result);
                            }

                            log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                        } catch (Exception e) {
                            log.warn("registerBroker Exception, {}", namesrvAddr, e);
                        } finally {
                            countDownLatch.countDown();
                        }
                    }
                });
            }

            try {
                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
            }
        }

        return registerBrokerResultList;
    }

NameServer端的处理：
DefaultRequestProcessor#processRequest

    //K2 NameServer处理请求的核心代码
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {

        if (ctx != null) {
            log.debug("receive request, {} {} {}",
                request.getCode(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                request);
        }


        switch (request.getCode()) {
            case RequestCode.PUT_KV_CONFIG:
                return this.putKVConfig(ctx, request);
            case RequestCode.GET_KV_CONFIG:
                return this.getKVConfig(ctx, request);
            case RequestCode.DELETE_KV_CONFIG:
                return this.deleteKVConfig(ctx, request);
            case RequestCode.QUERY_DATA_VERSION:
                return queryBrokerTopicConfig(ctx, request);
            case RequestCode.REGISTER_BROKER: //Broker注册请求处理。版本默认是当前框架版本
                Version brokerVersion = MQVersion.value2Version(request.getVersion());
                if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                    return this.registerBrokerWithFilterServer(ctx, request); //当前版本
                } else {
                    return this.registerBroker(ctx, request);
                }

最终会调用RouteInfoManager对Broker信息进行注册：

    //K2 NameServer 实际处理Broker注册的地方
    public RemotingCommand registerBrokerWithFilterServer(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
        final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
        final RegisterBrokerRequestHeader requestHeader =
            (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

        if (!checksum(ctx, request, requestHeader)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("crc32 not match");
            return response;
        }

        RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();

        if (request.getBody() != null) {
            try {
                registerBrokerBody = RegisterBrokerBody.decode(request.getBody(), requestHeader.isCompressed());
            } catch (Exception e) {
                throw new RemotingCommandException("Failed to decode RegisterBrokerBody", e);
            }
        } else {
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setCounter(new AtomicLong(0));
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0);
        }
        //routeInfoManager就是管理路由信息的核心组件。
        RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId(),
            requestHeader.getHaServerAddr(),
            registerBrokerBody.getTopicConfigSerializeWrapper(),
            registerBrokerBody.getFilterServerList(),
            ctx.channel());

        responseHeader.setHaServerAddr(result.getHaServerAddr());
        responseHeader.setMasterAddr(result.getMasterAddr());

        byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
        response.setBody(jsonValue);

        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

看看RouteInfoManager 管理的路由信息：

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    //几个关键的Table
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

2.发送消息

2.1 普通消息发送DefaultMQProducer

DefaultMQProducer#start

    public void start() throws MQClientException {
        this.setProducerGroup(withNamespace(this.producerGroup));
        this.defaultMQProducerImpl.start();
        if (null != traceDispatcher) {
            try {
                traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
            } catch (MQClientException e) {
                log.warn("trace dispatcher start failed ", e);
            }
        }
    }

DefaultMQProducerImpl#start()

    public void start() throws MQClientException {
        this.start(true);
    }
    //K2 消息生产者的启动方法
    public void start(final boolean startFactory) throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                this.serviceState = ServiceState.START_FAILED;

                this.checkConfig();
                //修改当前的instanceName为当前进程ID
                if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                    this.defaultMQProducer.changeInstanceNameToPID();
                }
                //客户端核心的MQ客户端工厂 对于事务消息发送者，在这里面会完成事务消息的发送者的服务注册
                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
                //注册MQ客户端工厂示例
                boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
                if (!registerOK) {
                    this.serviceState = ServiceState.CREATE_JUST;
                    throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                        null);
                }

                this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
                //启动示例 --所有客户端组件都交由mQClientFactory启动
                if (startFactory) {
                    mQClientFactory.start();
                }

                log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
                    this.defaultMQProducer.isSendMessageWithVIPChannel());
                this.serviceState = ServiceState.RUNNING;
                break;
            case RUNNING:
            case START_FAILED:
            case SHUTDOWN_ALREADY:
                throw new MQClientException("The producer service state not OK, maybe started once, "
                    + this.serviceState
                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                    null);
            default:
                break;
        }

        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

        this.startScheduledTask();

    }

总结：

1）RocketMQ的所有客户端实例，包括生产者和消费者，都是统一交由mQClientFactory组件来启动，也就是说，所有客户端的启动流程是固定的，不同客户端的区别只是在于他们在启动前注册的一些信息不同。例如生产者注册到producerTable，消费者注册到consumerTable，管理控制端注册到adminExtTable
2）MQClientInstance#start启动了很多服务
this.mQClientAPIImpl.fetchNameServerAddr();
this.mQClientAPIImpl.start();
this.startScheduledTask();
this.pullMessageService.start();
this.rebalanceService.start();
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

DefaultMQProducer#send(Message)
-> DefaultMQProducerImpl#send(Message)
-> DefaultMQProducerImpl#sendDefaultImpl

1）TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());获取Topic信息
2）MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);Producer根据发送者负载均衡策略，计算把消息发到哪个MessageQueue中（Producer在获取路由信息后，会选出一个MessageQueue去发送消息。这个选MessageQueue的方法就是一个索引自增然后取模的方式。.sendLatencyFaultEnable默认是关闭的，Broker故障延迟机制，表示一种发送消息失败后一定时间内不再往同一个Queue重复发送的机制）
3）sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);实际发送消息的方法（根据MessageQueue 获取对应的Broker地址）

Producer如何管理Borker路由信息？

Producer需要拉取Broker列表，然后跟Broker建立连接等等很多核心的流程，其实都是在发送消息时建立的。Send方法中，首先需要获得Topic的路由信息。这会从本地缓存中获取，如果本地缓存中没有，就从NameServer中去申请。

    //找路由表的过程都是先从本地缓存找，本地缓存没有，就去NameServer上申请
    private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
            //Producer向NameServer获取更新Topic的路由信息
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
            //还是从本地缓存中寻找Topic的路由信息
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
        }

        if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
            return topicPublishInfo;
        } else {
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
            return topicPublishInfo;
        }
    }

2.2 事务消息TransactionMQProducer

    //事务消息的启动过程。启动过程中会完成Processor的注册
    @Override
    public void start() throws MQClientException {
        this.defaultMQProducerImpl.initTransactionEnv();
        super.start();
    }

    public void initTransactionEnv() {
        TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
        if (producer.getExecutorService() != null) {
            this.checkExecutor = producer.getExecutorService();
        } else {
            this.checkRequestQueue = new LinkedBlockingQueue<Runnable>(producer.getCheckRequestHoldMax());
            this.checkExecutor = new ThreadPoolExecutor(
                producer.getCheckThreadPoolMinSize(),
                producer.getCheckThreadPoolMaxSize(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.checkRequestQueue);
        }
    }

这里唯一区别就是多了一个线程池checkExecutor。

    public TransactionSendResult sendMessageInTransaction(final Message msg,
        final Object arg) throws MQClientException {
        if (null == this.transactionListener) {
            throw new MQClientException("TransactionListener is null", null);
        }

        msg.setTopic(NamespaceUtil.wrapNamespace(this.getNamespace(), msg.getTopic()));
        return this.defaultMQProducerImpl.sendMessageInTransaction(msg, null, arg);
    }

DefaultMQProducerImpl#sendMessageInTransaction

1）TransactionListener transactionListener = getCheckListener();获取监听器
2）设置消息属性
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
3）发送消息sendResult = this.send(msg);
4）发送成功，执行本地事务localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
5）发送失败（各种错误原因），设置为回滚状态
6）根据状态进行处理this.endTransaction(msg, sendResult, localTransactionState, localException);这里会向broker发起提交或者回滚请求

客户端处理Broker回查请求：
ClientRemotingProcessor#processRequest
-> RequestCode.CHECK_TRANSACTION_STATE
ClientRemotingProcessor#checkTransactionState
-> DefaultMQProducerImpl#checkTransactionState
-> this.checkExecutor.submit(request);
-> localTransactionState = transactionListener.checkLocalTransaction(message);

3.消费消息

DefaultMQPushConsumer#start

    public void start() throws MQClientException {
        setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
        this.defaultMQPushConsumerImpl.start();
        if (null != traceDispatcher) {
            try {
                traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
            } catch (MQClientException e) {
                log.warn("trace dispatcher start failed ", e);
            }
        }
    }

DefaultMQPushConsumerImpl#start

1）CLUSTERING模式，this.defaultMQPushConsumer.changeInstanceNameToPID();
2）this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);客户端示例工厂，生产者也是交由这个工厂启动的。
3）this.rebalanceImpl.setAllocateMessageQueueStrategy(
this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());负载均衡策略
4）this.pullAPIWrapper = new PullAPIWrapper();
5）广播模式与集群模式的最本质区别就是offset存储的地方不一样。
广播模式是在消费者本地存储offset：this.offsetStore = new LocalFileOffsetStore();
集群模式是在Broker远端存储offset：this.offsetStore = new RemoteBrokerOffsetStore();
6）消费者服务
顺序消费监听创建ConsumeMessageOrderlyService；
并发消费监听创建ConsumeMessageConcurrentlyService；
this.consumeMessageService.start();
7）注册消费者。与生产者类似，客户端只要按要求注册即可，后续会随mQClientFactory一起启动。
mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
8） mQClientFactory.start();
9）this.updateTopicSubscribeInfoWhenSubscriptionChanged();
10）this.mQClientFactory.checkClientInBroker();
11）this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
12）this.mQClientFactory.rebalanceImmediately();

消费端负载均衡：

AllocateMachineRoomNearby：将同机房的Consumer和Broker优先分配在一起。
AllocateMessageQueueAveragely：平均分配。将所有MessageQueue平均分给每一个消费者
AllocateMessageQueueAveragelyByCircle：轮询分配。轮流的给一个消费者分配一个MessageQueue。
AllocateMessageQueueByConfig：不分配，直接指定一个messageQueue列表。类似于广播模式，直接指定所有队列。
AllocateMessageQueueByMachineRoom：按逻辑机房的概念进行分配。又是对BrokerName和ConsumerIdc有定制化的配置。
AllocateMessageQueueConsistentHash。源码中有测试代码AllocateMessageQueueConsitentHashTest。这个一致性哈希策略只需要指定一个虚拟节点数，是用的一个哈希环的算法，虚拟节点是为了让Hash数据在换上分布更为均匀。

消费者服务的串联：

DefaultMQPushConsumerImpl#start
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
instance = new MQClientInstance()
this.pullMessageService = new PullMessageService(this);
PullMessageService#run
PullMessageService#pullMessage
DefaultMQPushConsumerImpl#pullMessage
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest()消费者消息服务处理消费到的消息
ConsumeMessageConcurrentlyService#submitConsumeRequest
ConsumeMessageOrderlyService#submitConsumeRequest
this.consumeExecutor.submit(consumeRequest);并发和顺序消费的线程池线程数都为20
ConsumeMessageConcurrentlyService.ConsumeRequest#run
ConsumeMessageOrderlyService.ConsumeRequest#run 需要加锁
status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);这个listener就是自定义的业务逻辑。

4.Broker端

4.1 文件存储

DefaultMessageStore#putMessage

    //K1 Broker典型的消息存储处理
    //当前版本将默认的写入方式更改成了异步写入机制。
    @Override
    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
        try {
            return asyncPutMessage(msg).get();
        } catch (InterruptedException | ExecutionException e) {
            return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null);
        }
    }

DefaultMessageStore#asyncPutMessage
-> CommitLog#asyncPutMessage

1）迟消息的实现方式，就要修改一下msg的topic和queueID，改为系统默认创建的延迟队列。topic是固定的SCHEDULE_TOPIC_XXXX，queueId是根据延迟级别选择的。
2）加锁putMessageLock.lock();
3）找到最后一个CommitLog文件。最后一个就是当前写的文件
4）mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext); 以零拷贝的方式实现消息顺序写。 ByteBuffer.allocateDirect(fileSize)
5）putMessageLock.unlock();
6）submitFlushRequest(result, msg);提交刷盘请求
同步刷盘机制：向this.flushCommitLogService提交GroupCommitRequest请求，每10ms执行一次flush；
异步刷盘机制：flushCommitLogService.wakeup();或者commitLogService.wakeup();
7）submitReplicaRequest(result, msg);提交主从同步请求。
-> HAService#putRequest
-> this.groupTransferService.putRequest(request);
HAService有三个核心组件，与Master相关的是acceptSocketService和groupTransferService。其中acceptSocketService主要负责维护Master与Slave之间的TCP连接。groupTransferService主要与主从同步复制有关。而slave相关的则是haClient。

接下来看看ConsumeQueue和IndexFile的写入：

DefaultMessageStore#start
this.reputMessageService.start();
DefaultMessageStore.ReputMessageService#run，Commit日志分发服务，每隔1毫秒，会检查是否需要（就是看有没有新数据）向ConsumeQueue和IndexFile中转发一次CommitLog写入的消息。
DefaultMessageStore.ReputMessageService#doReput
DefaultMessageStore.this.doDispatch(dispatchRequest);
CommitLogDispatcherBuildConsumeQueue#doDispatch
CommitLogDispatcherBuildIndex#doDispatch
长轮询：如果有消息到了主节点，并且开启了长轮询。就要通过长轮询机制通知消费者，新消息已经到了，可以消费了。DefaultMessageStore.this.messageArrivingListener.arriving()，实例是NotifyMessageArrivingListener

定时删除过期文件：

DefaultMessageStore#start
DefaultMessageStore#addScheduleTask
定时任务：DefaultMessageStore.this.cleanFilesPeriodically();
this.cleanCommitLogService.run();定时删除过期commitlog
this.cleanConsumeQueueService.run();定时删除过期的consumequeue

4.2 长轮询机制

RocketMQ对消息消费者提供了Push推模式和Pull拉模式两种消费模式。但是这两种消费模式的本质其实都是Pull拉模式，Push模式可以认为是一种定时的Pull机制。但是这时有一个问题，当使用Push模式时，如果RocketMQ中没有对应的数据，那难道一直进行空轮询吗？如果是这样的话，那显然会极大的浪费网络带宽以及服务器的性能，并且，当有新的消息进来时，RocketMQ也没有办法尽快通知客户端，而只能等客户端下一次来拉取消息了。针对这个问题，RocketMQ实现了一种长轮询机制 long polling。

长轮询机制简单来说，就是当Broker接收到Consumer的Pull请求时，判断如果没有对应的消息，不用直接给Consumer响应(给响应也是个空的，没意义)，而是就将这个Pull请求给缓存起来。当Producer发送消息过来时，增加一个步骤去检查是否有对应的已缓存的Pull请求，如果有，就及时将请求从缓存中拉取出来，并将消息通知给Consumer。

消费者拉取消息：
BrokerController#registerProcessor
-> this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
-> PullMessageProcessor#processRequest()
-> ResponseCode.PULL_NOT_FOUND，消息长轮询1：消费者消费时，没有消息就会被缓存起来。brokerAllowSuspend 客户端初次请求消息时是指定的true。重新唤醒时指定为false，hasSuspendFlag默认都是true。
this.brokerController.getPullRequestHoldService().suspendPullRequest( topic, queueId, pullRequest);请求是PullRequest

生产者发送消息：
BrokerController#registerProcessor

    public void registerProcessor() {
        /**
         * SendMessageProcessor
         */
        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
        sendProcessor.registerSendMessageHook(sendMessageHookList);
        sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        /**
         * PullMessageProcessor
         */
        this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
        this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);

SendMessageProcessor#processRequest

SendMessageProcessor#asyncProcessRequest()
SendMessageProcessor#asyncSendMessage
DefaultMessageStore#asyncPutMessage（参见4.1）

接下来看看关联的ReputMessageService

DefaultMessageStore#start
this.reputMessageService.start();
DefaultMessageStore.ReputMessageService#run，Commit日志分发服务，每隔1毫秒，会检查是否需要（就是看有没有新数据）向ConsumeQueue和IndexFile中转发一次CommitLog写入的消息。
DefaultMessageStore.ReputMessageService#doReput
DefaultMessageStore.this.doDispatch(dispatchRequest);
CommitLogDispatcherBuildConsumeQueue#doDispatch
CommitLogDispatcherBuildIndex#doDispatch
长轮询：如果有消息到了主节点，并且开启了长轮询。就要通过长轮询机制通知消费者，新消息已经到了，可以消费了。DefaultMessageStore.this.messageArrivingListener.arriving()，实例是NotifyMessageArrivingListener
NotifyMessageArrivingListener#arriving 长轮询：生产者发送消息后的监听事件
this.pullRequestHoldService.notifyMessageArriving()
PullRequestHoldService#notifyMessageArriving() 会去this.pullRequestTable.get(key);查找

RocketMQ源码剖析

1.各个组件启动源码、框架结构

1.1 NameServer启动

1.2 Broker启动

1.3 Netty网络框架

1.4 Broker心跳注册过程

2.发送消息

2.1 普通消息发送DefaultMQProducer

2.2 事务消息TransactionMQProducer

3.消费消息

4.Broker端

4.1 文件存储

4.2 长轮询机制

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读