6、Broker消息处理流程（六）

前面分析完Broker启动会启动RemotingServer服务同时会注册Processor处理器，接着分析Producer进行消息的发送，当Producer发送完消息后就得到Broker去接收Producer发送的消息了。
Producer发送给Broker消息时候，发送的请求code为SEND_MESSAGE(这里在上一章节有过分析)，根据消息发送过来的Code，这时会调用NettyRemotingAbstract的processRequestCommand方法，该方法里面会根据消息传输的Code来取出对应的Processor，进入Processor系列类的SendMessageProcessor的asyncProcessRequest方法（前面这一部分之前都有过分析，接下来我们一起看看后面的操作，正好也将之前的知识串在一起更有利于理解和记忆）


public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final SendMessageContext mqtraceContext;switch (request.getCode()) {// 消息重回队列case RequestCode.CONSUMER_SEND_MSG_BACK:return this.asyncConsumerSendMsgBack(ctx, request);default:// 解析消息头SendMessageRequestHeader requestHeader = parseRequestHeader(request);if (requestHeader == null) {return CompletableFuture.completedFuture(null);}// 构建上下文，并调用处理前钩子函数mqtraceContext = buildMsgContext(ctx, requestHeader);this.executeSendMessageHookBefore(ctx, request, mqtraceContext);// 判断批量消息还是单条消息if (requestHeader.isBatch()) {return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);} else {return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);}}
}

首先解析消息头构建上下文，处理消息发送前钩子函数，最后异步处理消息请求，如果是批量消息调用asyncSendBatchMessage方法，如果是单条消息调用asyncSendMessage方法。

处理单条消息 private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,SendMessageContext mqtraceContext,SendMessageRequestHeader requestHeader) {// 准备响应命令对象final RemotingCommand response = preSend(ctx, request, requestHeader);final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();if (response.getCode() != -1) {return CompletableFuture.completedFuture(response);}final byte[] body = request.getBody();int queueIdInt = requestHeader.getQueueId();TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());if (queueIdInt < 0) {queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());}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());Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());MessageAccessor.setProperties(msgInner, origProps);// 时间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();// ...省略CompletableFuture<PutMessageResult> putMessageResult = null;String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);// 事务消息if (Boolean.parseBoolean(transFlag)) {if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()+ "] sending transaction message is forbidden");return CompletableFuture.completedFuture(response);}// 事务消息的状态（后面再分析）putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);} else {// 消息存储putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}// 生成结果返回return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);}

构建MessageExtBrokerInner对象，设置相关属性执行asyncPutMessage方法存储消息并将结果返回客户端。

创建响应，验证以及自动创建topic


// 准备响应，验证以及自动创建topic
private RemotingCommand preSend(ChannelHandlerContext ctx, RemotingCommand request,SendMessageRequestHeader requestHeader) {// 准备响应final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);// 设置唯一idresponse.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);// 获取broker处理请求服务的起始时间final long startTimestamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();if (this.brokerController.getMessageStore().now() < startTimestamp) {response.setCode(ResponseCode.SYSTEM_ERROR);response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimestamp)));return response;}response.setCode(-1);// 验证topic以及自动创建逻辑super.msgCheck(ctx, requestHeader, response);return response;
}

this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()用来判断是否支持自动创建topic，根据权限来判断如果是不支持自动创建就将权限设置为可读可写不可继承，后面我们去判断是否可以去继承，如果能继承就说明支持自动创建，这是就会new一个TopicConfig，这样就通过autoCreateTopicEnable自动来控制是否能够自动创建topic，同时也会调用registerBrokerAll方法注册到Broker路由信息里面，当然官方建议我们还是不要开启这个配置因为它没有做到压力的分摊。

存盘 asyncPutMessage方法
根据topic查询对应的路由信息即broker。

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {   msg.setStoreTimestamp(System.currentTimeMillis());msg.setBodyCRC(UtilAll.crc32(msg.getBody()));AppendMessageResult result = null;StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();String topic = msg.getTopic();final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE|| tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {// Delay Delivery// 延迟消息转到系统Topic(后面在分析)if (msg.getDelayTimeLevel() > 0) {// ...省略}}// 发送消息地址InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();if (bornSocketAddress.getAddress() instanceof Inet6Address) {msg.setBornHostV6Flag();}// 存储消息地址InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();if (storeSocketAddress.getAddress() instanceof Inet6Address) {msg.setStoreHostAddressV6Flag();}// 更新消息大小PutMessageThreadLocal putMessageThreadLocal = this.putMessageThreadLocal.get();updateMaxMessageSize(putMessageThreadLocal);if (!multiDispatch.isMultiDispatchMsg(msg)) {PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);if (encodeResult != null) {return CompletableFuture.completedFuture(encodeResult);}msg.setEncodedBuff(putMessageThreadLocal.getEncoder().getEncoderBuffer());}PutMessageContext putMessageContext = new PutMessageContext(generateKey(putMessageThreadLocal.getKeyBuilder(), msg));long elapsedTimeInLock = 0;MappedFile unlockMappedFile = null;// 写入CommitLog文件前加锁,保证文件操作并发安全putMessageLock.lock(); //spin or ReentrantLock ,depending on store configtry {// 获取最后一个mapperFileMappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();this.beginTimeInLock = beginLockTimestamp;msg.setStoreTimestamp(beginLockTimestamp);// 如果不存在或者满了就创建一个if (null == mappedFile || mappedFile.isFull()) {mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise}if (null == mappedFile) {log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));}// 实际写入CommitLog，在后面追加result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);switch (result.getStatus()) {// 添加成功直接breakcase PUT_OK:break;// 表示当前文件存放不下，只保存了一部分case END_OF_FILE:unlockMappedFile = mappedFile;// 创建一个新的文件mappedFile = this.mappedFileQueue.getLastMappedFile(0);if (null == mappedFile) {// XXX: warn and notify melog.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));}// 继续追加进去result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);break;case MESSAGE_SIZE_EXCEEDED:case PROPERTIES_SIZE_EXCEEDED:return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));case UNKNOWN_ERROR:return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));default:return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));}// 锁的时间elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;} finally {beginTimeInLock = 0;putMessageLock.unlock();}if (elapsedTimeInLock > 500) {log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);}if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {this.defaultMessageStore.unlockMappedFile(unlockMappedFile);}PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);// StatisticsstoreStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).add(1);storeStatsService.getSinglePutMessageTopicSizeTotal(topic).add(result.getWroteBytes());// 提交刷盘申请CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);// 提交主从复制申请CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {if (flushStatus != PutMessageStatus.PUT_OK) {putMessageResult.setPutMessageStatus(flushStatus);}if (replicaStatus != PutMessageStatus.PUT_OK) {putMessageResult.setPutMessageStatus(replicaStatus);}return putMessageResult;});}

首先它会去处理延时消息这里我不做过细的分析，后面针对各种消息在来具体分析，接着就将消息进行编码然后加锁并写入消息以获取最后文件进行追加的方式来将消息内存文件里面，最后进行刷盘以及通知主从同步的操作。

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

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

相关文章

模板函数模板类入门

模板函数模板类入门

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、pandas是什么&#xff1f;二、使用步骤 1.引入库2.读入数据总结 前言 提示&#xff1a;这里可以添加本文要记录的大概内容&#xff1a; 因为做牛客的题目…
阅读更多...
（2）(2.4) TerraRanger Tower/Tower EVO(360度)

（2）(2.4) TerraRanger Tower/Tower EVO(360度)

文章目录 前言 1 安装传感器并连接 2 通过地面站进行配置 3 参数说明 前言 TeraRanger Tower 可用于在 Loiter 和 AltHold 模式下进行目标规避。传感器的最大可用距离约为 4.5m。 TeraRanger Tower EVO 可用于在 Loiter 和 AltHold 模式下进行目标规避。传感器的最大可用…
阅读更多...
Vue实现条件渲染

Vue实现条件渲染

&#x1f4d1;前言 本文主要是【Vue】——Vue实现条件渲染的文章&#xff0c;如果有什么需要改进的地方还请大佬指出⛺️ &#x1f3ac;作者简介&#xff1a;大家好&#xff0c;我是听风与他&#x1f947; ☁️博客首页&#xff1a;CSDN主页听风与他 &#x1f304;每日一句&am…
阅读更多...
2023-12-06 LeetCode每日一题（最小化旅行的价格总和）

2023-12-06 LeetCode每日一题（最小化旅行的价格总和）

2023-12-06每日一题 一、题目编号 2646. 最小化旅行的价格总和二、题目链接 点击跳转到题目位置 三、题目描述 现有一棵无向、无根的树&#xff0c;树中有 n 个节点&#xff0c;按从 0 到 n - 1 编号。给你一个整数 n 和一个长度为 n - 1 的二维整数数组 edges &#xff0…
阅读更多...
dtaidistance 笔记：相似度压缩

dtaidistance 笔记：相似度压缩

1 相似度 相似度&#xff1a;1表示相等&#xff0c;0表示疏远 给定一组时间序列&#xff08;每一行是一个&#xff09;&#xff0c;计算基于DTW的逐对相似度 from dtaidistance import dtw, similarity s np.array([[0., 0, 1, 2, 1, 0, 1, 0, 0],[0., 1, 2, 0, 0, 0, 0, 0,…
阅读更多...
word 目录生成

word 目录生成

参考&#xff1a;百度安全验证 什么word生成的目录没有标题4&#xff1f;  我来答 分享 举报 1个回答 #热议# 海关有哪些禁运商品&#xff1f;查到后怎么办&#xff1f; 缘来是我LXZ 高粉答主 2023-09-28 缘来缘去终会散&#xff0c;花开花败总归尘。 关注 在word…
阅读更多...
PyQt6 QTabWidget选项卡控件

PyQt6 QTabWidget选项卡控件

​锋哥原创的PyQt6视频教程&#xff1a; 2024版 PyQt6 Python桌面开发 视频教程(无废话版) 玩命更新中~_哔哩哔哩_bilibili2024版 PyQt6 Python桌面开发 视频教程(无废话版) 玩命更新中~共计37条视频&#xff0c;包括&#xff1a;2024版 PyQt6 Python桌面开发 视频教程(无废话…
阅读更多...
Unity - yield return相关用法

Unity - yield return相关用法

yield return null; // 下一帧再执行后续代码yield return 0; //下一帧再执行后续代码yield return 6;//(任意数字) 下一帧再执行后续代码yield break; //直接结束该协程的后续操作yield return asyncOperation;//等异步操作结束后再执行后续代码yield return StartCoroution(/…
阅读更多...
测试岗外包干了3个月，技术退步明显。。。。。

测试岗外包干了3个月，技术退步明显。。。。。

先说一下自己的情况&#xff0c;本科生生&#xff0c;21年通过校招进入成都某软件公司&#xff0c;干了接近2年的功能测试&#xff0c;今年年初&#xff0c;感觉自己不能够在这样下去了&#xff0c;长时间呆在一个舒适的环境会让一个人堕落!而我已经在一个企业干了2年的功能测试…
阅读更多...
【Intel/Altera】 全系列FPGA最新汇总说明，持续更新中

【Intel/Altera】 全系列FPGA最新汇总说明，持续更新中

前言 2023年11月14日英特尔 FPGA中国技术日&#xff0c;Intel刚发布了新的FPGA系列&#xff0c;官网信息太多&#xff0c;我这里结合以前的信息&#xff0c;简单汇总更新一下&#xff0c;方便大家快速了解Intel/Altera FPGA家族。 目录 前言 Altera和Intel 型号汇总 1. Agi…
阅读更多...
Redis 命令全解析之 String类型

Redis 命令全解析之 String类型

文章目录 ⛄String 介绍⛄命令⛄对应 RedisTemplate API⛄应用场景 ⛄String 介绍 String 类型&#xff0c;也就是字符串类型&#xff0c;是Redis中最简单的存储类型。 其value是字符串&#xff0c;不过根据字符串的格式不同&#xff0c;又可以分为3类&#xff1a; ● string&…
阅读更多...
使用 PyTorch 进行 K 折交叉验证

使用 PyTorch 进行 K 折交叉验证

一、说明 中号机器学习模型在训练后必须使用测试集进行评估。我们这样做是为了确保模型不会过度拟合&#xff0c;并确保它们适用于现实生活中的数据集&#xff0c;与训练集相比&#xff0c;现实数据集的分布可能略有偏差。 但为了使您的模型真正稳健&#xff0c;仅仅通过训练/测…
阅读更多...
mysql5.6密码忘记重置

mysql5.6密码忘记重置

在 MySQL 5.6 中&#xff0c;使用 ALTER USER 语句来修改用户密码是不支持的。可以使用以下步骤来重置 root 用户密码&#xff1a; 停止 MySQL 服务。 在终端中执行以下命令&#xff1a; systemctl stop mysql启动 MySQL 服务&#xff0c;并跳过权限验证。 在终端中执行以下命…
阅读更多...
oracle 19c创建db_link名称带.com域名问题处理

oracle 19c创建db_link名称带.com域名问题处理

文章目录 一、修改PDB的global_name二、重启数据库实例三、修改domain后重试 一、修改PDB的global_name SYSorcl1>sho pdbsCON_ID CON_NAME OPEN MODE RESTRICTED ---------- ------------------------------ ---------- ----------2 PDB$SEED …
阅读更多...
从遍历到A星寻路算法

从遍历到A星寻路算法

在游戏当中&#xff0c;经常需要找一个点到其它点的路径。在之前的一篇博文(地图编辑器开发&#xff08;三&#xff09;)中也有使用到到A*寻路算法。我们期望能找到最短的路径&#xff0c;同时也需要考虑到查找路径的时间消耗。游戏中的地图可以图的数据结构来表示&#xff0c;…
阅读更多...
ThreadPoolExecutor应用源码剖析（四）

ThreadPoolExecutor应用源码剖析（四）

3.3.8 ThreadPoolExecutor的关闭方法 首先查看shutdownNow方法&#xff0c;可以从RUNNING状态转变为STOP // shutDownNow方法&#xff0c;shutdownNow不会处理阻塞队列的任务&#xff0c;将任务全部给你返回了。 public List<Runnable> shutdownNow() { // 声明返回结果…
阅读更多...
LeetCode双指针：第一个错误的版本

LeetCode双指针：第一个错误的版本

LeetCode双指针&#xff1a;第一个错误的版本 题目描述 你是产品经理&#xff0c;目前正在带领一个团队开发新的产品。不幸的是&#xff0c;你的产品的最新版本没有通过质量检测。由于每个版本都是基于之前的版本开发的&#xff0c;所以错误的版本之后的所有版本都是错的。 …
阅读更多...
2023年【危险化学品经营单位主要负责人】及危险化学品经营单位主要负责人模拟考试

2023年【危险化学品经营单位主要负责人】及危险化学品经营单位主要负责人模拟考试

题库来源&#xff1a;安全生产模拟考试一点通公众号小程序 危险化学品经营单位主要负责人根据新危险化学品经营单位主要负责人考试大纲要求&#xff0c;安全生产模拟考试一点通将危险化学品经营单位主要负责人模拟考试试题进行汇编&#xff0c;组成一套危险化学品经营单位主要…
阅读更多...
深度优先搜索(DFS)LeetCode 2477. 到达首都的最少油耗

深度优先搜索(DFS)LeetCode 2477. 到达首都的最少油耗

2477. 到达首都的最少油耗 给你一棵 n 个节点的树&#xff08;一个无向、连通、无环图&#xff09;&#xff0c;每个节点表示一个城市&#xff0c;编号从 0 到 n - 1 &#xff0c;且恰好有 n - 1 条路。0 是首都。给你一个二维整数数组 roads &#xff0c;其中 roads[i] [ai,…
阅读更多...
（十四）Flask之闪现flash

（十四）Flask之闪现flash

闪现—flash 这可不是LOL或是王者荣耀里的闪现哦~ Flask 中的 “闪现”&#xff08;flash&#xff09;是一种在请求之间传递消息的机制。它允许你将一条消息保存在一个请求中&#xff0c;在下一个请求中获取并显示该消息&#xff0c;然后立即将其删除【设置完之后阅后即焚&am…
阅读更多...
最新文章

Copyright @ 2022~2023