顺藤摸瓜RocketMQ之整体架构以及执行流程

柔情只为你懂 2022-12-26 09:13 111阅读 0赞

### 文章目录 ###

*   *  代码结构
     *  执行流程
     *   *  1.启动namesrv
         *  2.启动broker
         *  3.发送消息
         *  4.消费消息

## 代码结构 ##

映入眼帘，还是可以很清晰的分辨出各个模块都是负责哪块的

*  acl 访问控制列表（Access Control Lists，ACL），RocketMQ权限管理模块
 *  broker `接收客户端请求，处理消息分发，存储消息数据等`
 *  client MQ客户端，包括Producer和Consumer
 *  common 用于存储RocketMQ项目的通用代码和Model等
 *  dev merge\_rocketmq\_pr.py脚本，用于处理RP
 *  distribution Client、Namesrv、Broker等启动脚本及打包脚本
 *  docs 文档
 *  example RocketMQ示例代码，源码分析可以从这里入手
 *  filter 过滤器模块，包含SQL过滤
 *  logappender 日志appender模块
 *  logging 日志实现模块
 *  namesrv `Namesrv实现模块`
 *  openmessaging openmessaging模块
 *  remoting `基于Netty实现的网络通信模块，RocketMQ各组件之间的通信都依赖它`
 *  srvutil 工具包模块
 *  store `数据存储模块，例如Broker数据`
 *  style 代码风格XML文件
 *  test RocketMQ案例测试模块
 *  tools RocketMQ对外命令行接口、管理类接口等

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]

## 执行流程 ##

如下图：  
启动的时候namesvc、broker、store会向NettyRemotingServer注册自己的处理器以及操作的线程池，供NettyServerHandler进行调度，而NettyRemotingServer启动的时候则会使用Netty开启一个监听，监听到有请求或者是响应进来，就最终交给NettyServerHandler进行处理，NettyServerHandler会根据请求中的code值进行匹配，去processorTable中找到这个请求所对应的processer，以及线程池，然后创建出Task，将Task submit到processer。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]

### 1.启动namesrv ###

启动主要是执行一下这个方法：org.apache.rocketmq.namesrv.NamesrvController\#initialize

以下是代码清单：

public boolean initialize() { 
    		//将文件中的配置加载到kvConfigManager中，之后在kvConfigManager里面拿到配置，进行一些参数设置
            this.kvConfigManager.load();
    		//初始化remotingServer
            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    		//初始化执行线程池（process线程池）
            this.remotingExecutor =
                Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    		
            this.registerProcessor();
    		//开启一个定时任务，扫描挂掉的Broker
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    NamesrvController.this.routeInfoManager.scanNotActiveBroker();
                }
            }, 5, 10, TimeUnit.SECONDS);
    		//开启一个定时任务，定期打印所有config
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    NamesrvController.this.kvConfigManager.printAllPeriodically();
                }
            }, 1, 10, TimeUnit.MINUTES);
    		//配置ssl一些信息
            if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) { 
                // Register a listener to reload SslContext
                try { 
                    fileWatchService = new FileWatchService(
                        new String[] { 
                            TlsSystemConfig.tlsServerCertPath,
                            TlsSystemConfig.tlsServerKeyPath,
                            TlsSystemConfig.tlsServerTrustCertPath
                        },
                        new FileWatchService.Listener() { 
                            boolean certChanged, keyChanged = false;
                            @Override
                            public void onChanged(String path) { 
                                if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) { 
                                    log.info("The trust certificate changed, reload the ssl context");
                                    reloadServerSslContext();
                                }
                                if (path.equals(TlsSystemConfig.tlsServerCertPath)) { 
                                    certChanged = true;
                                }
                                if (path.equals(TlsSystemConfig.tlsServerKeyPath)) { 
                                    keyChanged = true;
                                }
                                if (certChanged && keyChanged) { 
                                    log.info("The certificate and private key changed, reload the ssl context");
                                    certChanged = keyChanged = false;
                                    reloadServerSslContext();
                                }
                            }
                            private void reloadServerSslContext() { 
                                ((NettyRemotingServer) remotingServer).loadSslContext();
                            }
                        });
                } catch (Exception e) { 
                    log.warn("FileWatchService created error, can't load the certificate dynamically");
                }
            }
    
            return true;
        }

this.registerProcessor();这个方法是将执行器注册到remoting模块中，如下代码：  
clusterTest默认为false，所以就会执行this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);  
可以看到remotingServer对象是NettyRemotingServer（传输服务），所以registerDefaultProcessor这个方法是将namesvc自己的DefaultRequestProcessor以及一个线程池注册到NettyRemotingServer中，以供NettyRemotingServer在接受到消息之后进行调用，这也就是`观察者模式`，比如broker发送一个注册消息到NettyRemotingServer中，这个时候NettyRemotingServer就会找到注册所对应的registerDefaultProcessor，然后构建出一个task，用注入进去的remotingExecutor来执行这个task。

private void registerProcessor() { 
            if (namesrvConfig.isClusterTest()) { 
    
                this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
                    this.remotingExecutor);
            } else { 
    
                this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
            }
        }

### 2.启动broker ###

createBrokerController方法主要是对broker中的各个配置进行初始化，包括需要注入到broker中的组件，以及broker中自己的一些时间配置，端口号设置，主从配置，优雅关闭。

接下来就来到了controller.start();方法

public void start() throws Exception { 
    		//启动store模块
            if (this.messageStore != null) { 
                this.messageStore.start();
            }
    		//启动remoting模块
            if (this.remotingServer != null) { 
                this.remotingServer.start();
            }
    		//启动remoting模块，端口号-2
            if (this.fastRemotingServer != null) { 
                this.fastRemotingServer.start();
            }
    		//开启文件扫描更新配置
            if (this.fileWatchService != null) { 
                this.fileWatchService.start();
            }
    		//启动RemotingClient-> 初始化Netty客户端线程组 以及 开启一个任务扫描响应
            if (this.brokerOuterAPI != null) { 
                this.brokerOuterAPI.start();
            }
    		//长轮询服务开启
            if (this.pullRequestHoldService != null) { 
                this.pullRequestHoldService.start();
            }
    		//开启定时任务 扫描异常的channel（超时的channel）
            if (this.clientHousekeepingService != null) { 
                this.clientHousekeepingService.start();
            }
    		//定时管理filtersvc
            if (this.filterServerManager != null) { 
                this.filterServerManager.start();
            }
    		//处理broker主从节点关系，将broker注册到所有namesvc上
            if (!messageStoreConfig.isEnableDLegerCommitLog()) { 
                startProcessorByHa(messageStoreConfig.getBrokerRole());
                handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
                this.registerBrokerAll(true, false, true);
            }
    		//开启定时任务向namesvc注册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);
    		//空的
            if (this.brokerStatsManager != null) { 
                this.brokerStatsManager.start();
            }
    		//将队列中过期请求快速失败掉
            if (this.brokerFastFailure != null) { 
                this.brokerFastFailure.start();
            }
        }

### 3.发送消息 ###

从org.apache.rocketmq.example.simple.Producer 这个类进入，

Message msg = new Message("TopicTest",
                            "TagA",
                            "OrderID188",
                            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                        SendResult sendResult = producer.send(msg);
                        System.out.printf("%s%n", sendResult);

进入到send方法中

@Override
        public SendResult send(
            Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
            Validators.checkMessage(msg, this);
            msg.setTopic(withNamespace(msg.getTopic()));
            return this.defaultMQProducerImpl.send(msg);
        }

再次进入到send方法中，

public SendResult send(
            Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
            return send(msg, this.defaultMQProducer.getSendMsgTimeout());
        }
        public SendResult send(Message msg,
            long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
            return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
        }

进入到sendDefaultImpl中，在代码的584行，

beginTimestampPrev = System.currentTimeMillis();
                            if (times > 0) { 
                                //Reset topic with namespace during resend.
                                msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
                            }
                            long costTime = beginTimestampPrev - beginTimestampFirst;
                            if (timeout < costTime) { 
                                callTimeout = true;
                                break;
                            }
    
                            sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
                            endTimestamp = System.currentTimeMillis();

在这个方法中可以看到，执行了发送消息前钩子，构建了请求头，最终在834行调用client中的sendMessage方法，

sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
                                brokerAddr,
                                mq.getBrokerName(),
                                tmpMessage,
                                requestHeader,
                                timeout - costTimeAsync,
                                communicationMode,
                                sendCallback,
                                topicPublishInfo,
                                this.mQClientFactory,
                                this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
                                context,
                                this);

进入sendMessage方法，然后来到487行，同步发送消息，

switch (communicationMode) { 
                case ONEWAY:
                    this.remotingClient.invokeOneway(addr, request, timeoutMillis);
                    return null;
                case ASYNC:
                    final AtomicInteger times = new AtomicInteger();
                    long costTimeAsync = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTimeAsync) { 
                        throw new RemotingTooMuchRequestException("sendMessage call timeout");
                    }
                    this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
                        retryTimesWhenSendFailed, times, context, producer);
                    return null;
                case SYNC:
                    long costTimeSync = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTimeSync) { 
                        throw new RemotingTooMuchRequestException("sendMessage call timeout");
                    }
                    return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
                default:
                    assert false;
                    break;
            }

private SendResult sendMessageSync(
            final String addr,
            final String brokerName,
            final Message msg,
            final long timeoutMillis,
            final RemotingCommand request
        ) throws RemotingException, MQBrokerException, InterruptedException { 
            RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
            assert response != null;
            return this.processSendResponse(brokerName, msg, response);
        }

来到NettyRemotingClient中的377行，这里调用了this.invokeSyncImpl

try { 
                    doBeforeRpcHooks(addr, request);
                    long costTime = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTime) { 
                        throw new RemotingTimeoutException("invokeSync call timeout");
                    }
                    RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
                    doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                    return response;
                } catch (RemotingSendRequestException e) { 
                    log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                    this.closeChannel(addr, channel);
                    throw e;
                } catch (RemotingTimeoutException e) { 
                    if (nettyClientConfig.isClientCloseSocketIfTimeout()) { 
                        this.closeChannel(addr, channel);
                        log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
                    }
                    log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                    throw e;
                }

然后invokeSyncImpl的默认实现在NettyRemotingAbstract这个类中406行，

public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
            final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException { 
            final int opaque = request.getOpaque();
    
            try { 
                final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
                this.responseTable.put(opaque, responseFuture);
                final SocketAddress addr = channel.remoteAddress();
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception { 
                        if (f.isSuccess()) { 
                            responseFuture.setSendRequestOK(true);
                            return;
                        } else { 
                            responseFuture.setSendRequestOK(false);
                        }
    
                        responseTable.remove(opaque);
                        responseFuture.setCause(f.cause());
                        responseFuture.putResponse(null);
                        log.warn("send a request command to channel <" + addr + "> failed.");
                    }
                });
    
                RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
                if (null == responseCommand) { 
                    if (responseFuture.isSendRequestOK()) { 
                        throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                            responseFuture.getCause());
                    } else { 
                        throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                    }
                }
    
                return responseCommand;
            } finally { 
                this.responseTable.remove(opaque);
            }
        }

构建出responseFuture，放到responseTable中，然后调用Netty的channel.writeAndFlush(request)方法，将消息写入到channel中，并设置了回调，接着因为我们分析的同步发送，所以RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);会使用CountDownLatch等待执行完成，然后将response返回出去。

接下来就到了NettyRemotingServer中了，也就是Netty的服务端，收到这消息之后会经过一系列的filter进行处理，做ssl操作，编解码，心跳处理，最终要的则是最后的serverHandler,  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]  
也就是下面这个处理类，

@ChannelHandler.Sharable
        class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> { 
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
                processMessageReceived(ctx, msg);
            }
        }

一路点进方法，直到来到NettyRemotingAbstract抽象类的默认processRequestCommand实现方法，

public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) { 
            final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
            final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
            final int opaque = cmd.getOpaque();
    
            if (pair != null) { 
                Runnable run = new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
                            final RemotingResponseCallback callback = new RemotingResponseCallback() { 
                                @Override
                                public void callback(RemotingCommand response) { 
                                    doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
                                    if (!cmd.isOnewayRPC()) { 
                                        if (response != null) { 
                                            response.setOpaque(opaque);
                                            response.markResponseType();
                                            try { 
                                                ctx.writeAndFlush(response);
                                            } catch (Throwable e) { 
                                                log.error("process request over, but response failed", e);
                                                log.error(cmd.toString());
                                                log.error(response.toString());
                                            }
                                        } else { 
                                        }
                                    }
                                }
                            };
                            if (pair.getObject1() instanceof AsyncNettyRequestProcessor) { 
                                AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
                                processor.asyncProcessRequest(ctx, cmd, callback);
                            } else { 
                                NettyRequestProcessor processor = pair.getObject1();
                                RemotingCommand response = processor.processRequest(ctx, cmd);
                                callback.callback(response);
                            }
                        } catch (Throwable e) { 
                            log.error("process request exception", e);
                            log.error(cmd.toString());
    
                            if (!cmd.isOnewayRPC()) { 
                                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                    RemotingHelper.exceptionSimpleDesc(e));
                                response.setOpaque(opaque);
                                ctx.writeAndFlush(response);
                            }
                        }
                    }
                };
    
                if (pair.getObject1().rejectRequest()) { 
                    final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                        "[REJECTREQUEST]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                    return;
                }
    
                try { 
                    final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                    pair.getObject2().submit(requestTask);
                } catch (RejectedExecutionException e) { 
                    if ((System.currentTimeMillis() % 10000) == 0) { 
                        log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                            + ", too many requests and system thread pool busy, RejectedExecutionException "
                            + pair.getObject2().toString()
                            + " request code: " + cmd.getCode());
                    }
    
                    if (!cmd.isOnewayRPC()) { 
                        final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                            "[OVERLOAD]system busy, start flow control for a while");
                        response.setOpaque(opaque);
                    }
                }
            } else { 
                String error = " request type " + cmd.getCode() + " not supported";
                final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
            }
        }

分析以上代码，this.processorTable.get(cmd.getCode())是根据code找到之前各个模块注册到remoting中的process，先看254行代码，构建出requestTask然后交给pair的第二个对象也就是之前启动各个模块注册到remoting模块的线程池们，所以核心就在run这个对象里面

final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
    	pair.getObject2().submit(requestTask);

在run对象里面，可以看到pair.getObject1获取了process来处理这个response，所以我们要找到发送消息对应的process对象，也就是org.apache.rocketmq.broker.processor.SendMessageProcessor对象，

if (pair.getObject1() instanceof AsyncNettyRequestProcessor) { 
           AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
           processor.asyncProcessRequest(ctx, cmd, callback);
    } else { 
           NettyRequestProcessor processor = pair.getObject1();
           RemotingCommand response = processor.processRequest(ctx, cmd);
           callback.callback(response);
    }

直接进入到org.apache.rocketmq.broker.processor.SendMessageProcessor\#processRequest方法，发现这边调用的是异步请求asyncProcessRequest，然后asyncProcessRequest返回的是一个CompletableFuture然后调用get方法阻塞来同步返回response，

@Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                              RemotingCommand request) throws RemotingCommandException { 
            RemotingCommand response = null;
            try { 
                response = asyncProcessRequest(ctx, request).get();
            } catch (InterruptedException | ExecutionException e) { 
                log.error("process SendMessage error, request : " + request.toString(), e);
            }
            return response;
        }

接着来到第101行，this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader)，进入到asyncSendMessage方法

MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
            msgInner.setTopic(requestHeader.getTopic());
            msgInner.setQueueId(queueIdInt);
    
            if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) { 
                return CompletableFuture.completedFuture(response);
            }
            msgInner.setBody(body);
            msgInner.setFlag(requestHeader.getFlag());
            MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
            msgInner.setPropertiesString(requestHeader.getProperties());
            msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
            msgInner.setBornHost(ctx.channel().remoteAddress());
            msgInner.setStoreHost(this.getStoreHost());
            msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : 	requestHeader.getReconsumeTimes());
            String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();
            MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);
            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));

我们发现了消息的定义，这里为消息初始化了一下，然后就操作存储模块将消息存储到文件中=>putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner)，这个我们最后再分析，下面有段代码handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt)

private CompletableFuture<RemotingCommand> handlePutMessageResultFuture(CompletableFuture<PutMessageResult> putMessageResult,
          RemotingCommand response,
          RemotingCommand request,
          MessageExt msgInner,
          SendMessageResponseHeader responseHeader,
          SendMessageContext sendMessageContext,
          ChannelHandlerContext ctx,
          int queueIdInt) { 
            return putMessageResult.thenApply((r) ->
                handlePutMessageResult(r, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt)
            );
        }

重点是putMessageResult.thenApply(® ->handlePutMessageResult(r, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt))，果然如果不持续学习，代码都看不懂了，这边的r是上面调用remoting模块的一个task，而thenApply方法是将 remoting的操作和handlePutMessageResult的操作串行化，并且后者是依赖前者的结果的，而我们进入到handlePutMessageResult这个方法里面可以看到主要是做了一些对response的一些设置初始化，设置了任务的状态码，消息时间，id等内容，然后返回给client的就是这个response。

最后还剩下一个remoting模块是如何将消息存储到文件的  
打开DefaultMessageStore类，来到477行putMessage方法，就是上面的this.brokerController.getMessageStore().asyncPutMessage(msgInner)方法所执行的内容，该方法对msg进行了一些校验之后执行PutMessageResult result = this.commitLog.putMessage(msg)方法，这里的commitLog对象就是负责存储数据的对象。进入到该对象的putMessage方法，先是堆msg对象设置了时间，主题，队列，是否开启事务，一些地址信息然后调用putMessageLock.lock()方法，默认是走PutMessageSpinLock这个类里面的lock方法，如下可以看到是用CAS自旋锁，

public class PutMessageSpinLock implements PutMessageLock { 
        //true: Can lock, false : in lock.
        private AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);
    
        @Override
        public void lock() { 
            boolean flag;
            do { 
                flag = this.putMessageSpinLock.compareAndSet(true, false);
            }
            while (!flag);
        }
    
        @Override
        public void unlock() { 
            this.putMessageSpinLock.compareAndSet(false, true);
        }
    }

加上锁之后在849行result = mappedFile.appendMessage(msg, this.appendMessageCallback);通过MappedFile对象写入文件，重点在210行的 cb.doAppend方法，进入到这个方法，源码已经把注释写的清清楚楚，最重要的是byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);将消息写入到byteBuffer，这边的byteBuffer如果没有开启事务的话就是MappedByteBuffer对象，org.apache.rocketmq.store.MappedFile类中的init方法里面初始化的，

public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
                final MessageExtBrokerInner msgInner) { 
                // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
    
                // PHY OFFSET
                long wroteOffset = fileFromOffset + byteBuffer.position();
    
                int sysflag = msgInner.getSysFlag();
                //是否是ipv6，host地址长度不一样
                int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
                int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
                ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
                ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
                //重置bytebuffer
                this.resetByteBuffer(storeHostHolder, storeHostLength);
                //创建出msgid，
                String msgId;
                if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) { 
                    msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
                } else { 
                    msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
                }
    
                // Record ConsumeQueue information
                keyBuilder.setLength(0);
                keyBuilder.append(msgInner.getTopic());
                keyBuilder.append('-');
                keyBuilder.append(msgInner.getQueueId());
                String key = keyBuilder.toString();//拼接出topic-queue
                Long queueOffset = CommitLog.this.topicQueueTable.get(key);//拿到偏移量
                if (null == queueOffset) { 
                    queueOffset = 0L;
                    CommitLog.this.topicQueueTable.put(key, queueOffset);
                }
    			//处理事务，如果是TRANSACTION_PREPARED_TYPE TRANSACTION_ROLLBACK_TYPE
                // Transaction messages that require special handling
                final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
                switch (tranType) { 
                    // Prepared and Rollback message is not consumed, will not enter the
                    // consumer queuec
                    case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                    case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                        queueOffset = 0L;
                        break;
                    case MessageSysFlag.TRANSACTION_NOT_TYPE:
                    case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    default:
                        break;
                }
    
                /** * Serialize message */
                final byte[] propertiesData =
                    msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
    
                final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
    			//长度超过，就直接返回异常
                if (propertiesLength > Short.MAX_VALUE) { 
                    log.warn("putMessage message properties length too long. length={}", propertiesData.length);
                    return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
                }
    			
                final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
                final int topicLength = topicData.length;
    
                final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
    
                final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
    
                // Exceeds the maximum message
                if (msgLen > this.maxMessageSize) { 
                    CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
                        + ", maxMessageSize: " + this.maxMessageSize);
                    return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
                }
    
                // Determines whether there is sufficient free space
                if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) { 
                    this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
                    // 1 TOTALSIZE
                    this.msgStoreItemMemory.putInt(maxBlank);
                    // 2 MAGICCODE
                    this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
                    // 3 The remaining space may be any value
                    // Here the length of the specially set maxBlank
                    final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
                    byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
                    return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
                        queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
                }
    
                // Initialization of storage space
                this.resetByteBuffer(msgStoreItemMemory, msgLen);
                // 1 TOTALSIZE
                this.msgStoreItemMemory.putInt(msgLen);
                // 2 MAGICCODE
                this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
                // 3 BODYCRC
                this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
                // 4 QUEUEID
                this.msgStoreItemMemory.putInt(msgInner.getQueueId());
                // 5 FLAG
                this.msgStoreItemMemory.putInt(msgInner.getFlag());
                // 6 QUEUEOFFSET
                this.msgStoreItemMemory.putLong(queueOffset);
                // 7 PHYSICALOFFSET
                this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
                // 8 SYSFLAG
                this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
                // 9 BORNTIMESTAMP
                this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
                // 10 BORNHOST
                this.resetByteBuffer(bornHostHolder, bornHostLength);
                this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
                // 11 STORETIMESTAMP
                this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
                // 12 STOREHOSTADDRESS
                this.resetByteBuffer(storeHostHolder, storeHostLength);
                this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
                // 13 RECONSUMETIMES
                this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
                // 14 Prepared Transaction Offset
                this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
                // 15 BODY
                this.msgStoreItemMemory.putInt(bodyLength);
                if (bodyLength > 0)
                    this.msgStoreItemMemory.put(msgInner.getBody());
                // 16 TOPIC
                this.msgStoreItemMemory.put((byte) topicLength);
                this.msgStoreItemMemory.put(topicData);
                // 17 PROPERTIES
                this.msgStoreItemMemory.putShort((short) propertiesLength);
                if (propertiesLength > 0)
                    this.msgStoreItemMemory.put(propertiesData);
    
                final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
                // Write messages to the queue buffer
                byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
    
                AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
                    msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
    
                switch (tranType) { 
                    case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                    case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                        break;
                    case MessageSysFlag.TRANSACTION_NOT_TYPE:
                    case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                        // The next update ConsumeQueue information
                        CommitLog.this.topicQueueTable.put(key, ++queueOffset);
                        break;
                    default:
                        break;
                }
                return result;
            }

所以回到CommitLog类863行，result其实就是PUT\_OK的AppendMessageResult，接下来就到了handleDiskFlush(result, putMessageResult, msg)方法，同步刷盘中flushOkFuture.get()方法执行了org.apache.rocketmq.store.CommitLog.GroupCommitService中的run方法，调用了doCommit()方法，然后执行到CommitLog.this.mappedFileQueue.flush(0)这个方法，然后点进int offset = mappedFile.flush(flushLeastPages);这个方法，可以看到force方法刷盘成功，至此同步刷盘成功。

try { 
                        //We only append data to fileChannel or mappedByteBuffer, never both.
                        if (writeBuffer != null || this.fileChannel.position() != 0) { 
                            this.fileChannel.force(false);
                        } else { 
                            this.mappedByteBuffer.force();
                        }
                    } catch (Throwable e) { 
                        log.error("Error occurred when force data to disk.", e);
                    }

### 4.消费消息 ###

同样从example工程开始，注册了一个回调MessageListenerConcurrently，然后执行start();方法

public static void main(String[] args) throws InterruptedException, MQClientException { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
            consumer.setNamesrvAddr("localhost:9876");
    
            consumer.subscribe("TopicTest", "*");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //wrong time format 2017_0422_221800
            consumer.setConsumeTimestamp("20181109221800");
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }

执行了，this.defaultMQPushConsumerImpl.start();以及traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());方法分别是启动了消息消费的服务，以及发送消息轨迹服务

在org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl\#start这个方法中执行了mQClientFactory.start();而mQClientFactory是MQClientInstance对象  
在方法里面有很多start()操作，

public void start() throws MQClientException { 
    
            synchronized (this) { 
                switch (this.serviceState) { 
                    case CREATE_JUST:
                        this.serviceState = ServiceState.START_FAILED;
                        // If not specified,looking address from name server
                        if (null == this.clientConfig.getNamesrvAddr()) { 
                            this.mQClientAPIImpl.fetchNameServerAddr();
                        }
                        // Start request-response channel
                        this.mQClientAPIImpl.start();
                        // Start various schedule tasks
                        this.startScheduledTask();
                        // Start pull service
                        this.pullMessageService.start();
                        // Start rebalance service
                        this.rebalanceService.start();
                        // Start push service
                        this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                        log.info("the client factory [{}] start OK", this.clientId);
                        this.serviceState = ServiceState.RUNNING;
                        break;
                    case START_FAILED:
                        throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                    default:
                        break;
                }
            }
        }

其中this.pullMessageService.start()是拉取消息的定时器，实现在如下org.apache.rocketmq.client.impl.consumer.PullMessageService中的run方法：  
拿到一个拉取请求，然后调用pullMessage(pullRequest)方法

public void run() { 
            log.info(this.getServiceName() + " service started");
    
            while (!this.isStopped()) { 
                try { 
                    PullRequest pullRequest = this.pullRequestQueue.take();
                    this.pullMessage(pullRequest);
                } catch (InterruptedException ignored) { 
                } catch (Exception e) { 
                    log.error("Pull Message Service Run Method exception", e);
                }
            }
    
            log.info(this.getServiceName() + " service end");
        }

来到org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl\#pullMessage方法中，设置一些操作时间，进行一些校验，定义了callback，最后调用了如下方法，去拉取消息

try { 
                this.pullAPIWrapper.pullKernelImpl(
                    pullRequest.getMessageQueue(),
                    subExpression,
                    subscriptionData.getExpressionType(),
                    subscriptionData.getSubVersion(),
                    pullRequest.getNextOffset(),
                    this.defaultMQPushConsumer.getPullBatchSize(),
                    sysFlag,
                    commitOffsetValue,
                    BROKER_SUSPEND_MAX_TIME_MILLIS,
                    CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
                    CommunicationMode.ASYNC,
                    pullCallback
                );
            } catch (Exception e) { 
                log.error("pullKernelImpl exception", e);
                this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
            }

在pullKernelImpl这个方法里面，继续点进去

PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
                    brokerAddr,
                    requestHeader,
                    timeoutMillis,
                    communicationMode,
                    pullCallback);

可以看到RemotingCommand.createRequestCommand(RequestCode.PULL\_MESSAGE, requestHeader);这里构造了RemotingCommand并且将RequestCode.PULL\_MESSAGE放了进去，

public PullResult pullMessage(
            final String addr,
            final PullMessageRequestHeader requestHeader,
            final long timeoutMillis,
            final CommunicationMode communicationMode,
            final PullCallback pullCallback
        ) throws RemotingException, MQBrokerException, InterruptedException { 
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);
    
            switch (communicationMode) { 
                case ONEWAY:
                    assert false;
                    return null;
                case ASYNC:
                    this.pullMessageAsync(addr, request, timeoutMillis, pullCallback);
                    return null;
                case SYNC:
                    return this.pullMessageSync(addr, request, timeoutMillis);
                default:
                    assert false;
                    break;
            }
    
            return null;
        }

而在BrokerController启动的时候注入了好多process，我们找到 this.remotingServer.registerProcessor(RequestCode.PULL\_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);这个，他的默认实现类org.apache.rocketmq.broker.processor.PullMessageProcessor所以，在调用了remoting模块之后，最终会回到PullMessageProcessor这个类来处理拉取请求，所以我们先走完大致发送流程，在着重看看这个PullMessageProcessor

接下来，来到org.apache.rocketmq.client.impl.MQClientAPIImpl\#pullMessageSync，可以看到调用了remoting模块来执行请求，然后调用processPullResponse(response)将返回体中的body给PullResultExt构建出了PullResult返回出去

private PullResult pullMessageSync(
            final String addr,
            final RemotingCommand request,
            final long timeoutMillis
        ) throws RemotingException, InterruptedException, MQBrokerException { 
            RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
            assert response != null;
            return this.processPullResponse(response);
        }

如果是异步的，那就将设置一个回调，而回调里面就是执行的pullCallback回调，remoting模块这么处理的上面发送消息的时候有讲解到，然后remoting模块会调用该request的code找到消费消息的process进行处理

private void pullMessageAsync(
            final String addr,
            final RemotingCommand request,
            final long timeoutMillis,
            final PullCallback pullCallback
        ) throws RemotingException, InterruptedException { 
            this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() { 
                @Override
                public void operationComplete(ResponseFuture responseFuture) { 
                    RemotingCommand response = responseFuture.getResponseCommand();
                    if (response != null) { 
                        try { 
                            PullResult pullResult = MQClientAPIImpl.this.processPullResponse(response);
                            assert pullResult != null;
                            pullCallback.onSuccess(pullResult);
                        } catch (Exception e) { 
                            pullCallback.onException(e);
                        }
                    } else { 
                        if (!responseFuture.isSendRequestOK()) { 
                            pullCallback.onException(new MQClientException("send request failed to " + addr + ". Request: " + request, responseFuture.getCause()));
                        } else if (responseFuture.isTimeout()) { 
                            pullCallback.onException(new MQClientException("wait response from " + addr + " timeout :" + responseFuture.getTimeoutMillis() + "ms" + ". Request: " + request,
                                responseFuture.getCause()));
                        } else { 
                            pullCallback.onException(new MQClientException("unknown reason. addr: " + addr + ", timeoutMillis: " + timeoutMillis + ". Request: " + request, responseFuture.getCause()));
                        }
                    }
                }
            });
        }

PullMessageProcessor代码中的processRequest方法，对请求进行了一些校验之后来到如下代码去获取Message

final GetMessageResult getMessageResult =
                this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                    requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);

然后就来到了store模块中的getMessage方法，该方法在653行调用了SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);最终是从commitLog里面获取到消息

首先先找到对应offset的文件，也就是内存映射文件mappedFile，在改文件中读取的起始位置就是offeset%mappedFileSize，这不难理解，每个mappedFile偏移量都是固定一个G，所以想要获得在改文件中的位置，取余即可，然后读取size大小，既是消息。

public SelectMappedBufferResult getMessage(final long offset, final int size) { 
            int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
            MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
            if (mappedFile != null) { 
                int pos = (int) (offset % mappedFileSize);
                return mappedFile.selectMappedBuffer(pos, size);
            }
            return null;
        }

进入到该方法，一目了然，操作byteBuffer，最终返回回去

public SelectMappedBufferResult selectMappedBuffer(int pos, int size) { 
            int readPosition = getReadPosition();
            if ((pos + size) <= readPosition) { 
                if (this.hold()) { 
                    ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
                    byteBuffer.position(pos);
                    ByteBuffer byteBufferNew = byteBuffer.slice();
                    byteBufferNew.limit(size);
                    return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this);
                } else { 
                    log.warn("matched, but hold failed, request pos: " + pos + ", fileFromOffset: "
                        + this.fileFromOffset);
                }
            } else { 
                log.warn("selectMappedBuffer request pos invalid, request pos: " + pos + ", size: " + size
                    + ", fileFromOffset: " + this.fileFromOffset);
            }
    
            return null;
        }

最后，来到PullCallback这个回调，pullResult.getPullStatus()是被设置成FOUND的所以走第一个分支，

PullCallback pullCallback = new PullCallback() { 
                @Override
                public void onSuccess(PullResult pullResult) { 
                    if (pullResult != null) { 
                        pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                            subscriptionData);
    
                        switch (pullResult.getPullStatus()) { 
                            case FOUND:
                                long prevRequestOffset = pullRequest.getNextOffset();
                                pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                                long pullRT = System.currentTimeMillis() - beginTimestamp;
                                DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
                                    pullRequest.getMessageQueue().getTopic(), pullRT);
    
                                long firstMsgOffset = Long.MAX_VALUE;
                                if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) { 
                                    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                } else { 
                                    firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
    
                                    DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
                                        pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
    
                                    boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                                    DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                        pullResult.getMsgFoundList(),
                                        processQueue,
                                        pullRequest.getMessageQueue(),
                                        dispatchToConsume);
    
                                    if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) { 
                                        DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                                            DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                                    } else { 
                                        DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                    }
                                }
    
                                if (pullResult.getNextBeginOffset() < prevRequestOffset
                                    || firstMsgOffset < prevRequestOffset) { 
                                    log.warn(
                                        "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
                                        pullResult.getNextBeginOffset(),
                                        firstMsgOffset,
                                        prevRequestOffset);
                                }
    
                                break;
                            case NO_NEW_MSG:
                                pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                                DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                break;
                            case NO_MATCHED_MSG:
                                pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                                DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                break;
                            case OFFSET_ILLEGAL:
                                log.warn("the pull request offset illegal, {} {}",
                                    pullRequest.toString(), pullResult.toString());
                                pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                                pullRequest.getProcessQueue().setDropped(true);
                                DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() { 
    
                                    @Override
                                    public void run() { 
                                        try { 
                                            DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
                                                pullRequest.getNextOffset(), false);
    
                                            DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
    
                                            DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
    
                                            log.warn("fix the pull request offset, {}", pullRequest);
                                        } catch (Throwable e) { 
                                            log.error("executeTaskLater Exception", e);
                                        }
                                    }
                                }, 10000);
                                break;
                            default:
                                break;
                        }
                    }
                }
    
                @Override
                public void onException(Throwable e) { 
                    if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { 
                        log.warn("execute the pull request exception", e);
                    }
    
                    DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                }
            };

进入到this.consumeMessageService.submitConsumeRequest这个方法，来到ConsumeMessageConcurrentlyService这个默认的类里面org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService\#submitConsumeRequest可以看到构建了一个ConsunRequest对象，进去到该对象的run方法，

ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
                    try { 
                        this.consumeExecutor.submit(consumeRequest);
                    } catch (RejectedExecutionException e) { 
                        for (; total < msgs.size(); total++) { 
                            msgThis.add(msgs.get(total));
                        }
    
                        this.submitConsumeRequestLater(consumeRequest);
                    }

在run里面可以看到，最后是回调执行了listener.consumeMessage方法，而listener就是一开始设置进去的MessageListenerConcurrently这个类，

try { 
                    if (msgs != null && !msgs.isEmpty()) { 
                        for (MessageExt msg : msgs) { 
                            MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
                        }
                    }
                    status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
                } catch (Throwable e) { 
                    log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
                        RemotingHelper.exceptionSimpleDesc(e),
                        ConsumeMessageConcurrentlyService.this.consumerGroup,
                        msgs,
                        messageQueue);
                    hasException = true;
                }

至此，RocketMQ的整个骨干流程就已经完成，之后再分析异步，以及他的一些机制。

> 1.  [顺藤摸瓜RocketMQ之整体架构以及执行流程][RocketMQ]
> 2.  [顺藤摸瓜RocketMQ之注册中心debug解析][RocketMQ_debug]
> 3.  [顺藤摸瓜RocketMQ之消息发送debug解析][RocketMQ_debug 1]
> 4.  [顺藤摸瓜RocketMQ之消息消费debug解析][RocketMQ_debug 2]
> 5.  [顺藤摸瓜RocketMQ之刷盘机制debug解析][RocketMQ_debug 3]
> 6.  [顺藤摸瓜RocketMQ之主从同步(HA)解析][RocketMQ_HA]

> 如有哪里分析不对，请在评论区指出，或者私信我，我会即时改正，不能误导人

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]: /images/20221120/5e3962696abc4ccca17eba204efcdbd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]: /images/20221120/9f5d6494228b4a00bfccde4bd8da8c33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]: /images/20221120/90a10462d6184dcf87e37e7ff9606dd2.png
[RocketMQ]: https://blog.csdn.net/jianjun_fei/article/details/110622652
[RocketMQ_debug]: https://blog.csdn.net/jianjun_fei/article/details/117549909
[RocketMQ_debug 1]: https://blog.csdn.net/jianjun_fei/article/details/117158517
[RocketMQ_debug 2]: https://blog.csdn.net/jianjun_fei/article/details/117230633
[RocketMQ_debug 3]: https://blog.csdn.net/jianjun_fei/article/details/117448644
[RocketMQ_HA]: https://blog.csdn.net/jianjun_fei/article/details/117307639
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/9a4f00efe0f54fcb9eed40b24fd45991.png