rocketmq-product-send方法源码分析

先看有哪些send方法

在这里插入图片描述

首先说红圈的

有3个红圈。归类成3种发送方式。假设前提条件，发送的topic，有3个broker，每个broker总共4个write队列，总共有12个队列。

  • 普通发送。负载均衡12个队列。指定超时时间
  • 指定MessageQueue,发送，指定超时时间
  • 指定selector器，指定特定参数，指定超时时间。一般用于局部有序，比如相同userId的，到同一个队列

默认超时时间时3秒

再说蓝圈

  • sendDefaultImpl 负载均衡的方式，选择队列。然后调sendKernelImpl
  • sendSelectImpl 指定队列selector和arg的方式，选择队列。然后调sendKernelImpl
  • sendKernelImpl 最核心的方式。这里已经明确队列，做真实的消息发送

很明显，只需要简单解读sendDefaultImpl和sendSelectImpl如何选择队列。然后重点在于查看sendKernelImpl方法实现

sendDefaultImpl选择队列分析

先看源码

private SendResult sendDefaultImpl(Message msg,final CommunicationMode communicationMode,final SendCallback sendCallback,final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {this.makeSureStateOK();Validators.checkMessage(msg, this.defaultMQProducer);final long invokeID = random.nextLong();long beginTimestampFirst = System.currentTimeMillis();long beginTimestampPrev = beginTimestampFirst;long endTimestamp = beginTimestampFirst;TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());if (topicPublishInfo != null && topicPublishInfo.ok()) {boolean callTimeout = false;MessageQueue mq = null;Exception exception = null;SendResult sendResult = null;int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;int times = 0;String[] brokersSent = new String[timesTotal];for (; times < timesTotal; times++) {String lastBrokerName = null == mq ? null : mq.getBrokerName();MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);if (mqSelected != null) {mq = mqSelected;brokersSent[times] = mq.getBrokerName();try {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();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);switch (communicationMode) {case ASYNC:return null;case ONEWAY:return null;case SYNC:if (sendResult.getSendStatus() != SendStatus.SEND_OK) {if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {continue;}}return sendResult;default:break;}

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendDefaultImpl

第一步，通过topic查找路由信息tryToFindTopicPublishInfo
先从内存中获取。内存是DefaultMQProducerImpl#topicPublishInfoTable
如果内存没有，则从nameserver获取
org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String)

内存是什么时候添加的呢？是有定时器任务更新的。详情看我写的文章rocketmq-push模式-消费侧重平衡-类流程图分析

第二步、设定默认重试3次（包含首次）,选择topic的其中一个队列
org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {if (lastBrokerName == null) {return selectOneMessageQueue();} else {for (int i = 0; i < this.messageQueueList.size(); i++) {int index = this.sendWhichQueue.incrementAndGet();int pos = Math.abs(index) % this.messageQueueList.size();if (pos < 0)pos = 0;MessageQueue mq = this.messageQueueList.get(pos);if (!mq.getBrokerName().equals(lastBrokerName)) {return mq;}}return selectOneMessageQueue();}
}

可以发现，topic对应的TopicPublishInfo，维护者一个ThreadLocalIndex对象。
每个线程先会获取一个index，然后对index取模，得到某一个队列。
这意味着，sendDefaultImpl中，队列的负载均衡是线程独立的。每个线程维护着自己的index，每发送一次，index+1。

public int incrementAndGet() {Integer index = this.threadLocalIndex.get();if (null == index) {index = Math.abs(random.nextInt());this.threadLocalIndex.set(index);}this.threadLocalIndex.set(++index);return Math.abs(index & POSITIVE_MASK);}

第三步、选择完MessageQueue后，调用sendKernelImpl发送消息

sendSelectImpl选择队列分析

先看源码

private SendResult sendSelectImpl(Message msg,MessageQueueSelector selector,Object arg,final CommunicationMode communicationMode,final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {long beginStartTime = System.currentTimeMillis();this.makeSureStateOK();Validators.checkMessage(msg, this.defaultMQProducer);TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());if (topicPublishInfo != null && topicPublishInfo.ok()) {MessageQueue mq = null;try {List<MessageQueue> messageQueueList =mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());Message userMessage = MessageAccessor.cloneMessage(msg);String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), mQClientFactory.getClientConfig().getNamespace());userMessage.setTopic(userTopic);mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));} catch (Throwable e) {throw new MQClientException("select message queue threw exception.", e);}long costTime = System.currentTimeMillis() - beginStartTime;if (timeout < costTime) {throw new RemotingTooMuchRequestException("sendSelectImpl call timeout");}if (mq != null) {return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout - costTime);} else {throw new MQClientException("select message queue return null.", null);}}validateNameServerSetting();throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);}

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendSelectImpl

第一步，通过topic查找路由信息tryToFindTopicPublishInfo。分析同上
第二步，通过MessageQueueSelector，找出发送的MessageQueue
MessageQueueSelector的实现方式，可以自定义。提供了2种
SelectMessageQueueByRandom 随机一个
SelectMessageQueueByHash 根据arg的hashcode取模一个。适合局部有序

public class SelectMessageQueueByHash implements MessageQueueSelector {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {int value = arg.hashCode() % mqs.size();if (value < 0) {value = Math.abs(value);}return mqs.get(value);}
}

第三步、选择完MessageQueue后，调用sendKernelImpl发送消息

sendKernelImpl发送分析

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendKernelImpl
第一步、通过MessageQueue，获取对应的master节点地址
第二步、设置消息的唯一id。详情看以下实现。明显是客户端生成的，（由于不是分布式唯一ID的创建方式，有点怀疑会重复。后续查看）
org.apache.rocketmq.common.message.MessageClientIDSetter#createUniqID
第三步、对消息body做消息压缩
第四步、判断该消息是否是事务消息。给sysFlag位标志变量加标志
第五步、发送前可做一些自定义的检查CheckForbiddenHook、SendMessageHook
第六步、构建SendMessageRequestHeader requestHeader，将msg的一些内容设置到header上
第七部、根据发送模式communicationMode，调用不同的sendMessage方法
org.apache.rocketmq.client.impl.MQClientAPIImpl#sendMessage

switch (communicationMode) {case ASYNC:Message tmpMessage = msg;boolean messageCloned = false;if (msgBodyCompressed) {//If msg body was compressed, msgbody should be reset using prevBody.//Clone new message using commpressed message body and recover origin massage.//Fix bug:https://github.com/apache/rocketmq-externals/issues/66tmpMessage = MessageAccessor.cloneMessage(msg);messageCloned = true;msg.setBody(prevBody);}if (topicWithNamespace) {if (!messageCloned) {tmpMessage = MessageAccessor.cloneMessage(msg);messageCloned = true;}msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));}long costTimeAsync = System.currentTimeMillis() - beginStartTime;if (timeout < costTimeAsync) {throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");}sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr,mq.getBrokerName(),tmpMessage,requestHeader,timeout - costTimeAsync,communicationMode,sendCallback,topicPublishInfo,this.mQClientFactory,this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),context,this);break;case ONEWAY:case SYNC:long costTimeSync = System.currentTimeMillis() - beginStartTime;if (timeout < costTimeSync) {throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");}sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr,mq.getBrokerName(),msg,requestHeader,timeout - costTimeSync,communicationMode,context,this);break;default:assert false;break;
}

第八步、最终会调用NettyRemotingClient的发送方法
SYNC：
org.apache.rocketmq.remoting.netty.NettyRemotingClient#invokeSync
ONEWAY:
org.apache.rocketmq.remoting.netty.NettyRemotingClient#invokeOneway
ASYNC:
org.apache.rocketmq.remoting.netty.NettyRemotingClient#invokeAsync

总结

product的发送有几种API模式，其实目的都是为了选择MessageQueue

  • 默认的发送，是根据topic的队列，做负载均衡的方式，topicPublishInfo内部维护着ThreadLocalIndex对象，做线程级别的负载均衡。而且默认都3次重试机会，意味可以选择不同队列做发送；
  • 指定messageQueue，是调用方明确知道发送的MessageQueue，这种失败不会做重试；
  • 指定MessageQueueSelector等，这种是通过传入的参数，计算出对应的MessageQueue，这种失败不会做重试，适合作为局部有序的发送方式

选择好队列后，就会调用org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendKernelImpl方法，主要是构建SendMessageRequestHeader，执行自定义的发送before和after的处理。
sendKernelImpl最终会调用NettyRemotingClient提供的接口，分别处理SYNC、ONEWAY、ASYNC的三种模式

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/7339.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

新电脑安装系统找不到硬盘原因和解决方法来了

新电脑安装系统找不到硬盘原因和解决方法来了

有不少网友反馈新电脑采用官方u盘方式装win10或win100出现找不到硬盘是怎么回事&#xff1f;后来研究半天发现是bios中开启了rst(vmd)模式。如果关闭rst模式肯定是可以安装的&#xff0c;但这会影响硬盘性能&#xff0c;有没有办法解决开启rst模式的情况安装win10或win11呢&…
阅读更多...
蓝桥杯之c++入门（一）【第一个c++程序】

蓝桥杯之c++入门（一）【第一个c++程序】

目录 前言一、第⼀个C程序1.1 基础程序1.2 main函数1.3 字符串1.4 头文件1.5 cin 和 cout 初识1.6 名字空间1.7 注释 二、四道简单习题&#xff08;点击跳转链接&#xff09;练习1&#xff1a;Hello,World!练习2&#xff1a;打印飞机练习3&#xff1a;第⼆个整数练习4&#xff…
阅读更多...
Electron学习笔记，安装环境（1）

Electron学习笔记，安装环境（1）

1、支持win7的Electron 的版本是18&#xff0c;这里node.js用的是14版本&#xff08;node-v14.21.3-x86.msi&#xff09;云盘有安装包 Electron 18.x (截至2023年仍在维护中): Chromium: 96 Node.js: 14.17.0 2、安装node环境&#xff0c;node-v14.21.3-x86.msi双击运行选择安…
阅读更多...
【机器学习】自定义数据集使用框架的线性回归方法对其进行拟合

【机器学习】自定义数据集使用框架的线性回归方法对其进行拟合

一、使用框架的线性回归方法 1. 基础原理 在自求导线性回归中&#xff0c;我们需要先自定义参数&#xff0c;并且需要通过数学公式来对w和b进行求导&#xff0c;然后在反向传播过程中通过梯度下降的方式来更新参数&#xff0c;从而降低损失值。 2. 实现步骤 ① 散点输入 有一…
阅读更多...
微服务搭建----springboot接入Nacos2.x

微服务搭建----springboot接入Nacos2.x

springboot接入Nacos2.x nacos之前用的版本是1.0的&#xff0c;现在重新搭建一个2.0版本的&#xff0c;学如逆水行舟&#xff0c;不进则退&#xff0c;废话不多说&#xff0c;开搞 1、 nacos2.x搭建 1&#xff0c;首先第一步查询下项目之间的版本对照&#xff0c;不然后期会…
阅读更多...
扣子平台音频功能：让声音也能“智能”起来

扣子平台音频功能：让声音也能“智能”起来

在数字化时代&#xff0c;音频内容的重要性不言而喻。无论是在线课程、有声读物&#xff0c;还是各种多媒体应用&#xff0c;音频都是传递信息、增强体验的关键元素。扣子平台的音频功能&#xff0c;为开发者和内容创作者提供了一个强大而灵活的工具&#xff0c;让音频的使用和…
阅读更多...
全面了解 Web3 AIGC 和 AI Agent 的创新先锋 MelodAI

全面了解 Web3 AIGC 和 AI Agent 的创新先锋 MelodAI

不管是在传统领域还是 Crypto&#xff0c;AI 都是公认的最有前景的赛道。随着数字内容需求的爆炸式增长和技术的快速迭代&#xff0c;Web3 AIGC&#xff08;AI生成内容&#xff09;和 AI Agent&#xff08;人工智能代理&#xff09;正成为两大关键赛道。 AIGC 通过 AI 技术生成…
阅读更多...
【Uniapp-Vue3】动态设置页面导航条的样式

【Uniapp-Vue3】动态设置页面导航条的样式

1. 动态修改导航条标题 uni.setNavigationBarTitle({ title:"标题名称" }) 点击修改以后顶部导航栏的标题会从“主页”变为“动态标题” 2. 动态修改导航条颜色 uni.setNavigationBarColor({ backgroundColor:"颜色" }) 3. 动态添加导航加载动画 // 添加加…
阅读更多...
GitLab配置免密登录和常用命令

GitLab配置免密登录和常用命令

SSH 免密登录 Windows免密登录 删除现有Key 访问目录&#xff1a;C:\Users\Administrator\ .ssh&#xff0c;删除公钥&#xff1a;id_rsa.pub &#xff0c;私钥&#xff1a;id_rsa 2.生成.ssh 秘钥 运行命令生成.ssh 秘钥目录&#xff08; ssh-keygen -t rsa -C xxxxxx126.…
阅读更多...
Spring Boot 自动配置

Spring Boot 自动配置

目录 什么是自动配置&#xff1f; Spring 加载 Bean ComponentScan Import 导入类 导入 ImportSelector 接口的实现类 SpringBoot 原理分析 EnableAutoConfiguration Import(AutoConfigurationImportSelector.class) AutoConfigurationPackage SpringBoot 自动配置流…
阅读更多...
16.好数python解法——2024年省赛蓝桥杯真题

16.好数python解法——2024年省赛蓝桥杯真题

问题描述 一个整数如果按从低位到高位的顺序,奇数位(个位、百位、万位…)上的数字是奇数,偶数位(十位、千位、十万位…)上的数字是偶数,我们就称之为“好数”。 给定一个正整数N,请计算从1到N一共有多少个好数。 输入格式 一个整数N。 输出格式 一个整数代表答案。 样例输入 1 …
阅读更多...
wxwidgets直接获取系统图标，效果类似QFileIconProvider

wxwidgets直接获取系统图标，效果类似QFileIconProvider

目前只做了windows版本&#xff0c;用法类似QFileIconProvider // 头文件 #ifndef WXFILEICONPROVIDER_H #define WXFILEICONPROVIDER_H#include <wx/wx.h> #include <wx/icon.h> #include <wx/image.h> #include <wx/bmpcbox.h> // Include for wxB…
阅读更多...
微服务学习-服务调用组件 OpenFeign 实战

微服务学习-服务调用组件 OpenFeign 实战

1. OpenFeign 接口方法编写规范 1.1. 在编写 OpenFeign 接口方法时&#xff0c;需要遵循以下规范 1.1.1.1. 接口中的方法必须使用 RequestMapping、GetMapping、PostMapping 等注解声明 HTTP 请求的类型。 1.1.1.2. 方法的参数可以使用 RequestParam、RequestHeader、PathVa…
阅读更多...
鸿蒙模块概念和应用启动相关类（HAP、HAR、HSP、AbilityStage、UIAbility、WindowStage、window）

鸿蒙模块概念和应用启动相关类（HAP、HAR、HSP、AbilityStage、UIAbility、WindowStage、window）

目录 鸿蒙模块概念 HAP entry feature har shared 使用场景 HAP、HAR、HSP介绍 HAP、HAR、HSP开发 应用的启动 AbilityStage UIAbility WindowStage Window 拉起应用到显示到前台流程 鸿蒙模块概念 HAP hap包是手机安装的最小单元&#xff0c;1个app包含一个或…
阅读更多...
想品客老师的第六天：函数

想品客老师的第六天：函数

函数基础的部分写在这里 函数声明 在js里万物皆对象&#xff0c;函数也可以用对象的方式定义 let func new Function("title", "console.log(title)");func(我是参数title); 也可以对函数赋值&#xff1a; let cms function (title) {console.log(tit…
阅读更多...
Python：元组构造式和字典推导式

Python：元组构造式和字典推导式

&#xff08;Python 元组构造式和字典推导式整理笔记&#xff09; 1. 元组构造式 1.1 创建元组 使用圆括号&#xff1a; tuple1 (1, 2.5, (three, four), [True, 5], False) print(tuple1) # 输出: (1, 2.5, (three, four), [True, 5], False) 省略圆括号&#xff1a; tup…
阅读更多...
Linux之Tcp粘包笔记

Linux之Tcp粘包笔记

目录 一.网络传输四层模型 二.数据传输中数据包传输的两个限制概念 三.数据传输的中粘包问题 四.数据组装的原因 Nagle算法原理: 五.关闭Nagle优化处理粘包问题吗&#xff1f; 六.粘包处理方法 a.设置消息边界&#xff1a; b.定义消息长度&#xff1a; 七.UDP是否会出…
阅读更多...
【C语言算法刷题】第2题 图论 dijkastra

【C语言算法刷题】第2题 图论 dijkastra

题目描述 一个局域网内有很多台电脑&#xff0c;分别标注为 0 ~ N-1 的数字。相连接的电脑距离不一样&#xff0c;所以感染时间不一样&#xff0c;感染时间用 t 表示。 其中网络内一台电脑被病毒感染&#xff0c;求其感染网络内所有的电脑最少需要多长时间。如果最后有电脑不…
阅读更多...
软件测试压力太大了怎么办？

软件测试压力太大了怎么办？

本文其实是知乎上针对一个问题的回答&#xff1a; 目前在做软件测试&#xff0c;主要负责的是手机端的项目测试&#xff0c;项目迭代很快&#xff0c;每次上线前验正式都会发现一些之前验测试包时候没有发现的问题&#xff0c;压力太大了&#xff0c;应该怎么调整 看过我之前其…
阅读更多...
枚举与模拟 练习

枚举与模拟 练习

练习题基于《C/C程序设计竞赛真题实战特训教程&#xff08;图解版&#xff09;》 目录 1.1 卡片 题目描述 代码实现 题解笔记 总评 注意点 重点解释 1.2 回文日期 题目描述 输入描述 输出描述 代码实现 题解笔记 总评 注意点 重点解释 1.1 卡片 题目描述 小蓝…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023