RocketMQ源码阅读-Broker消息接收

这篇具有很好参考价值的文章主要介绍了RocketMQ源码阅读-Broker消息接收。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。


Broker接收 Producer发送的消息。

Broker在RocketMQ中也是一个独立的Model,rocketmq-broker。

Broker的核心类为SendMessageProcessor。
RocketMQ源码阅读-Broker消息接收,RocketMQ源码阅读,rocketmq,log4j

1. 从单元测试入手

同样从单元测试入手,看Broker接收消息的流程。
SendMessageProcessor的单元测试类为org.apache.rocketmq.broker.processor.SendMessageProcessorTest。
RocketMQ源码阅读-Broker消息接收,RocketMQ源码阅读,rocketmq,log4j
包含上面这些方法,其中init()方法是启动类,其他是测试流程的方法。
先看init()方法中Broker的启动流程。

2. Broker启动流程

单元测试中的inti()方法为:

@Before
public void init() {
    brokerController.setMessageStore(messageStore);
    when(messageStore.now()).thenReturn(System.currentTimeMillis());
    Channel mockChannel = mock(Channel.class);
    when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
    when(handlerContext.channel()).thenReturn(mockChannel);
    when(messageStore.lookMessageByOffset(anyLong())).thenReturn(new MessageExt());
    sendMessageProcessor = new SendMessageProcessor(brokerController);
}

init方法新建一个SendMessageProcessor,并一个传入brokerController,指定一个messageStore:

  • BrokerController: Broker的管理器
  • MessageStore: 消息存储的接口

继续看接收消息的流程SendMessageProcessor#sendMessage。

3. Broker接收消息

@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                      RemotingCommand request) throws RemotingCommandException {
    SendMessageContext mqtraceContext;
    switch (request.getCode()) {
        case RequestCode.CONSUMER_SEND_MSG_BACK:
            return this.consumerSendMsgBack(ctx, request);
        default:
            // 解析请求
            SendMessageRequestHeader requestHeader = parseRequestHeader(request);
            if (requestHeader == null) {
                return null;
            }

            // 发送请求Context。在hook场景下使用
            mqtraceContext = buildMsgContext(ctx, requestHeader);
            // hook:处理发送消息前逻辑
            this.executeSendMessageHookBefore(ctx, request, mqtraceContext);

            RemotingCommand response;
            if (requestHeader.isBatch()) {
                // 处理批量发消息逻辑
                response = this.sendBatchMessage(ctx, request, mqtraceContext, requestHeader);
            } else {
                // 处理发送消息逻辑
                response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
            }

            this.executeSendMessageHookAfter(response, mqtraceContext);
            return response;
    }
}

可以看到,此方法负责解析RPC请求,最终是调用SendMessageProcessor#sendMessage方法:

private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
                                    final RemotingCommand request,
                                    final SendMessageContext sendMessageContext,
                                    final SendMessageRequestHeader requestHeader) throws RemotingCommandException {

    // 初始化响应
    final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
    final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();

    response.setOpaque(request.getOpaque());

    response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
    response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));

    log.debug("receive SendMessage request command, {}", request);

    // 如果未开始接收消息,抛出系统异常
    final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
    if (this.brokerController.getMessageStore().now() < startTimstamp) {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
        return response;
    }

    // 消息配置(Topic配置)校验
    response.setCode(-1);
    super.msgCheck(ctx, requestHeader, response);
    if (response.getCode() != -1) {
        return response;
    }

    final byte[] body = request.getBody();

    // 如果队列小于0,从可用队列随机选择
    int queueIdInt = requestHeader.getQueueId();
    TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());

    if (queueIdInt < 0) {
        queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
    }

    // 创建MessageExtBrokerInner
    MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    msgInner.setTopic(requestHeader.getTopic());
    msgInner.setQueueId(queueIdInt);

    if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
        return 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());
    PutMessageResult putMessageResult = null;
    Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
    String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    // 校验是否不允许发送事务消息
    if (traFlag != null && Boolean.parseBoolean(traFlag)) {
        if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark(
                "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                    + "] sending transaction message is forbidden");
            return response;
        }
        putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
    } else {
        // 添加消息
        putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    }
	// 处理result
    return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);

}

可以看到,此方法进行了topic的校验,并创建创建MessageExtBrokerInner,随后添加消息的流程主要是调用了MessageStore#putMessage方法:
MessageStore是接口,其默认实现为DefaultMessageStore:
RocketMQ源码阅读-Broker消息接收,RocketMQ源码阅读,rocketmq,log4j
DefaultMessageStore#putMessage方法:

public PutMessageResult putMessage(MessageExtBrokerInner msg) {
    if (this.shutdown) {
        log.warn("message store has shutdown, so putMessage is forbidden");
        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    }

	// 从节点不允许写入
    if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
        long value = this.printTimes.getAndIncrement();
        if ((value % 50000) == 0) {
            log.warn("message store is slave mode, so putMessage is forbidden ");
        }

        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    }
	// store是否允许写入
    if (!this.runningFlags.isWriteable()) {
        long value = this.printTimes.getAndIncrement();
        if ((value % 50000) == 0) {
            log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());
        }

        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    } else {
        this.printTimes.set(0);
    }
	// 消息过长
    if (msg.getTopic().length() > Byte.MAX_VALUE) {
        log.warn("putMessage message topic length too long " + msg.getTopic().length());
        return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
    }
	// 消息附加属性过长
    if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
        log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
        return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
    }

    if (this.isOSPageCacheBusy()) {
        return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
    }

    long beginTime = this.getSystemClock().now();
	// 添加消息到commitLog
    PutMessageResult result = this.commitLog.putMessage(msg);

    long eclipseTime = this.getSystemClock().now() - beginTime;
    if (eclipseTime > 500) {
        log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length);
    }
    this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);

    if (null == result || !result.isOk()) {
        this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
    }

    return result;
}

可以看到,首先是检查Broker是否可以写入,从节点不能写入,然后消息经过一系列的格式与大小校验,最终通过CommitLog.putMessage进行存储。

4. Broker接收消息时序图

RocketMQ源码阅读-Broker消息接收,RocketMQ源码阅读,rocketmq,log4j

5. 小结

Producer作为客户端发送消息后,Broker作为服务端需要接收消息并存储消息。
本篇分析了broker接收消息的流程:

  • SendMessageProcessor#processRequest是RPC执行接收消息的方法,此方法主要负责解析RPC请求
  • processRequest调用SendMessageProcessor#sendMessage方法,进行了topic的校验,并创建创建MessageExtBrokerInner,随后添加消息的流程主要是调用了DefaultMessageStore#putMessage方法
  • DefaultMessageStore#putMessage方法检查Broker是否可以写入,从节点不能写入,然后消息经过一系列的格式与大小校验,最终通过commitLog.putMessage进行存储

消息存储是通过CommitLong#putMessage进行的,这个流程在下一篇《RocketMQ源码阅读-Broker消息存储》学习。文章来源地址https://www.toymoban.com/news/detail-799245.html

到了这里,关于RocketMQ源码阅读-Broker消息接收的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • [RocketMQ] Broker asyncPutMessage处理消息以及存储的高性能设计措施 (十一)

    asyncPutMessage方法真正的用来存储消息。 1.asyncPutMessage存储普通消息 DefaultMessageStore#asyncPutMessage() checkStoreStatus, checkMessage, checkLmqMessage校验。 CommitLog#asyncPutMessage存储消息, 更新耗时时间和失败次数。 1.1 checkStoreStatus检查存储状态 如果DefaultMessageStore是shutdown状态, 返回SERVICE_NOT_

    2024年02月13日
    浏览(49)
  • [RocketMQ] Broker CommitLogDispatcher 异步构建ConsumeQueue和IndexFile源码解析 (十四)

    CommitLogDispatcherBuildConsumeQueue: 异步构建ConsumerQueue。 CommitLogDispatcherBuildIndex: 异步构建IndexFile。 1.CommitLogDispatcherBuildConsumeQueue构建ConsumeQueue CommitLogDispatcherBuildConsumeQueue用于接收分发请求并构建ConsumeQueue。 对于非事务消息或者是事务commit消息, 调用DefaultMessageStore#putMessagePositionI

    2024年02月17日
    浏览(42)
  • Rocketmq 定时消息源码分析

    生产者将消息投放到broker后,不会马上被消费者消费。需要等待到特定时间才会被消费。  producer 将定时消息写入commitLog 线程ReputThead 休息1毫秒,读取一次commitlog数据,写入ConsumeQueue和IndexFile 线程ScheduledService 首次延时1秒执行,以后延迟100毫秒执行。职责是将到期的延时消

    2024年02月14日
    浏览(35)
  • 二、RocketMQ消息存储源码分析

    Broker模块涉及到的内容非常多,本课程重点讲解以下技术点: 1、Broker启动流程分析 2、消息存储设计 3、消息写入流程 4、亮点分析:NRS与NRC的功能号设计 5、亮点分析:同步双写数倍性能提升的CompletableFuture 6、亮点分析:Commitlog写入时使用可重入锁还是自旋锁? 7、亮点分析

    2024年02月16日
    浏览(40)
  • 全网最细RocketMQ源码四:消息存储

    看完上一章之后,有没有很好奇,生产者发送完消息之后,server是如何存储,这一章节就来学习 SendMessageProcessor.processRequest 实际真正的负责存储就是DefaultMessageStore, 不过在讲述DefaultMessageStore的时候,我们是自底往上学,因为DefaultMessageStore比较复杂,从顶往下学容易学乱。先

    2024年01月16日
    浏览(42)
  • 消息中间件之RocketMQ源码分析(十)

    启动命令 nohup ./bin/mqnamesrv -c ./conf/namesrv.conf dev/null 21 通过脚本配置启动基本参数,比如配置文件路径、JVM参数,调用NamesrvStartup.main()方法,解析命令行的参数,将处理好的参数转化为Java实例,传递给NamesrvController实例 加载命令行传递的配置参数,调用controller.initialize()方法初

    2024年02月20日
    浏览(57)
  • RocketMQ(三) broker启动

    RocketMQ源码版本V5.0.0,可兼容之前的版本,因为整理资料的时候,之前的版本,和V5版本有所出入,核心流程基本还是大同小异的。 此前已经总结了NameServer的启动流程源码:现在来了解Broker的启动流程。在RocketMQ启动的时候,首先要启动NameServer,然后再启动Broker。 Broker模块主

    2024年02月08日
    浏览(47)
  • RocketMQ broker启动失败

    版本:4.9.3 现象:NameServer启动没问题,Broker无法启动。 查看日志,没有broker方面的报错,应该是整个服务都没起来。 于是开始网上搜索解决方案: 方案1: 删除store文件夹。 删除之后问题依旧 方案2: 更改broker.conf,加上IP等配置。 发现这些配置已经有了,于是更改无效。

    2024年02月09日
    浏览(43)
  • 38.RocketMQ之Broker的主从架构

    本文摘自:Broker的主从架构是怎么实现的 Broker是RocketMQ的核心模块,负责接收并存储消息,为了保证整个MQ的高可用,一般情况都会将Broker部署成集群,集群中的每一部分都由Master和Slave组成,那么Master与Slave之间的数据是如何保证同步一致的呢? 是Master主动把数据推送给Slav

    2024年02月13日
    浏览(46)
  • 【分布式技术专题】RocketMQ延迟消息实现原理和源码分析

    痛点背景 业务场景 假设有这么一个需求,用户下单后如果30分钟未支付,则该订单需要被关闭。你会怎么做? 之前方案 最简单的做法,可以服务端启动个定时器,隔个几秒扫描数据库中待支付的订单,如果(当前时间-订单创建时间)30分钟,则关闭订单。 方案评估 优点:是实

    2024年02月13日
    浏览(53)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包