producer

水深无声 2022-11-01 05:58 221阅读 0赞

启动：

创建clientId(ip@\{instanceName\}), 创建或获取clientId对应的MQClientInstance

instanceName：优先用户设置，其次系统属性，最后pid，一般来说同一进程所有客户端相同，即一般来说只有一instanceName，一clientId，一MQClientInstance

两特殊情况: 同机器多进程用了同一clientId，用户设置或系统属性相同了

单进程多clientId，不同客户端设置的instanceName不同

MQClientInstance的启动 （关联了多个producer和consumer，即多个topic，多个broker）

1 MQClientAPIImpl-netty客户端（在public线程也可以处理几种请求，如查询客户端事务状态）启动

2 定时任务：

未设置ns定时获取ns

向ns获取topic路由信息，并设置给各producer和consumer，topic配置里的写队列数配置给producer的TopicPublishInfo，读队列数配置给consumer

移除不在topic路由信息的broker数据，对所有相关broker发心跳（心跳包含clientId和所有producer和consumer数据，若无consumer则不给slave心跳），

所有consumer涉及到的topic，如果是push且类过滤模式，则上传源码到所有filterSrv（一个topic关联多brokerName，一brokerName多brokerAddr，一brokerAddr多filterSrv）

所有consumer持久化位点

所有push的consumer根据消息堆积情况（如consumer消费3队列，队列最大offset为10，现在消费到5，则consumer在此队列上堆积5）调整消费线程池大小

3 PullMessageService和RebalanceService 线程启动

4 MQClientInstance内部使用的producer启动

消息发送producer侧

retry：注意sync和async的重试区别，oneway不重试

default topic：取不到topic的路由信息，取AUTO\_CREATE\_TOPIC\_KEY 的路由信息，注意取到后读写队列数量需修改

select mq：若SendLatencyFaultEnable 未开启，轮询选择且retry尽量不选上次的broker，若开启，根据发送时延或异常推迟broker的可用时间，优先选择可用且上次的broker，否则选择可用性时延较好的broker

若broker busy，又retry 到此busy的broker，直接退出

生成messageId，超过4k消息compress

消息发送broker侧

若broker开启了autoCreateTopicEnable，则保留了AUTO\_CREATE\_TOPIC\_KEY的配置信息，则可以在broker创建自定义topic，也可以创建retry topic

pageCache busy：putMessage过程是加锁的，加锁开始记录时间，解锁清零，如果当前时间-加锁开始时间 有点大，则busy，会拒绝消息写入

延迟消息会重置topic和队列

如果mappedFile剩余空间不足，也要写入total size+blank的magic code，然后创建新mappedfile写入

mappedfile即便直接使用page cache（无transient pool），也可以锁定page cache在内存（需配置开启），新mappedfile创建后老的mappedfile就会解锁

同步刷盘且wait会等待，同步的ha master且wait会等待

事务producer

broker侧放入prepare/half消息的专属topic和队列0，客户端执行本地事务，根据事务结果endTransaction，broker处理endTransaction

broker处理endTransaction

commit 找到事务消息放入真实topic，opQueue放入

rollback opQueue放入

broker定时对halfQueue里包含但opQueue不包含的消息进行反查，客户端执行本地检查并endTransaction

halfQueue和opQueue的位点broker都会不断更新，half消息重放入halfQueue后再进行反查 （即half消息会不断地自我复制和反查，除非反查有明确结果导致opQueue含有或者超过最大反查次数）

发表评论取消回复

表情：

评论列表（有 0 条评论，221人围观）

还没有评论，来说两句吧...

相关阅读

相关 java kafka producer_Java实现Kafka Producer方法

import org.apache.kafka.clients.producer.\; import java.util.Properties; public class

﹏ヽ暗。殇╰゛Y/ 2022年11月03日 11:15/ 0 赞/ 148 阅读

相关 producer

启动：创建clientId(ip@\{instanceName\}), 创建或获取clientId对应的MQClientInstance instanceName：优先用

水深无声/ 2022年11月01日 05:58/ 0 赞/ 222 阅读

相关 12.kafka Producer Example

生产者 Producer类用于为特定主题和可选分区创建新消息。如果使用Java，您需要为Producer和支持类包括几个包： <table style="width

刺骨的言语ヽ痛彻心扉/ 2022年07月08日 15:57/ 0 赞/ 154 阅读

相关 RocketMQ观后感--Producer，Consumer

![这里写图片描述][SouthEast] 先讲Producer的总结，之后讲Consumer的总结。 Producer观后感总结： 1. Producer与NameS

系统管理员/ 2022年05月27日 02:08/ 0 赞/ 208 阅读

相关 Kafka Producer详解(二)

文章目录一、Producer介绍同步和异步的方式调用send() 二、Producer工作原理介绍

骑猪看日落/ 2022年05月18日 07:58/ 0 赞/ 193 阅读

相关 Kafka之Producer

producer 发布消息 1、内容将数据转化为“消息”，并通过网络发送给Broker，每条数据被称为“消息”，每条信息表示为一个三元组，即<t

た入场券/ 2022年02月05日 03:13/ 0 赞/ 181 阅读

相关 RocketMQ消息发送【Producer】

RocketMQ发送普通消息有三种方式：可靠同步发送、可靠异步发送、单向发送【OneWay】。\[RocketMQ消息结构-消息生产者启动流程-消息发送流程-批量消息发送\]

男娘i/ 2021年12月12日 11:33/ 0 赞/ 382 阅读

相关 Kafka幂等Producer和事务Producer

一、Kafka 消息交付可靠性 Kafka 对Producer和Consumer 需要处理的消息提供三种承诺：最多一次（at most once）：消息可能丢失

た入场券/ 2021年11月11日 13:44/ 0 赞/ 314 阅读

相关 kafak producer 配置介绍

producer的配置中，bootstrap.servers、key.serializer、value.serializer 这三个是必须的。接下来我们介绍一些重要的配置，更详

梦里梦外;/ 2021年06月22日 15:37/ 0 赞/ 376 阅读

相关 Kafka Producer配置

java生产者配置： <table> <thead> <tr> <th>NAME</th> <th>DESCRIPTION</th> <th

ゞ浴缸里的玫瑰/ 2021年06月22日 15:36/ 0 赞/ 389 阅读