RocketMQ启动过程之Producer

producer 1.启动流程

Producer如何感知要发送消息的broker即brokerAddrTable中的值是怎么获得的,

1.      发送消息的时候指定会指定topic,如果producer集合中没有会根据指定topic到namesrv获取topic发布信息TopicPublishInfo,并放入本地集合

2.      定时从namesrv更新topic路由信息,

Producer与broker间的心跳

Producer定时发送心跳将producer信息(其实就是procduer的group)定时发送到, brokerAddrTable集合中列出的broker上去

Producer发送消息只发送到master的broker机器,在通过broker的主从复制机制拷贝到broker的slave上去

producer 2.如何发送消息

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡



1)  Topic下的所有队列如何理解:

比如broker1, broker2, borker3三台broker机器都配置了Topic_A

Broker1 的队列为queue0 , queue1

Broker2 的队列为queue0, queue2, queue3,

Broker3 的队列为queue0

当然一般情况下的broker的配置都是一样的

以上当broker启动的时候注册到namesrv的Topic_A队列为共6个分别为:

broker1_queue0, broker1_queue1,

broker2_queue0, broker2_queue1, broker2_queue2,

broker3_queue0,

2)  Producer如何实现轮询队列:

Producer从namesrv获取的到Topic_A路由信息TopicPublishInfo

          --ListmessageQueueList  //Topic_A的所有的队列

          --AtomicIntegersendWhichQueue        //自增整型

          方法selectOneMessageQueue方法用来选择一个发送队列

                    (++sendWitchQueue)% messageQueueList.size为队列集合的下标

                    每次获取queue都会通过sendWhichQueue加一来实现对所有queue的轮询

                            如果入参lastBrokerName不为空,代表上次选择的queue发送失败,这次选择应该避开同一个queue

3)  Producer发消息系统重试:

发送失败后,重试几次retryTimesWhenSendFailed = 2

发送消息超时sendMsgTimeout = 3000

Producer通过selectOneMessageQueue方法获取一个MessagQueue对象

          --topic            //Topic_A

          --brokerName          //代表发送消息到达的broker

          --queueId              //代表发送消息的在指定broker上指定topic下的队列编号

向指定broker的指定topic的指定queue发送消息

                  发送失败(1)重试次数不到两次(2)发送此条消息花费时间还没有到3000(毫秒), 换个队列继续发送。

 producer发送普通消息

producer 3.如何发送顺序消息

Rocketmq能够保证消息严格顺序,但是Rocketmq需要producer保证顺序消息按顺序发送到同一个queue中,比如购买流程(1)下单(2)支付(3)支付成功,

这三个消息需要根据特定规则将这个三个消息按顺序发送到一个queue如何实现把顺序消息发送到同一个queue:          一般消息是通过轮询所有队列发送的,顺序消息可以根据业务比如说订单号orderId相同的消息发送到同一个队列, 或者同一用户userId发送到同一队列等等messageQueueList [orderId%messageQueueList.size()]messageQueueList [userId%messageQueueList.size()]



producer 4.如何发布分布式事务消息

先引入官方文档图:


分布式事物是基于二阶段提交的

1)      一阶段,向broker发送一条prepared的消息,返回消息的offset即消息地址commitLog中消息偏移量。Prepared状态消息不被消费发送消息ok,执行本地事物分支, 本地事物方法需要实现rocketmq的回调接口

2) LocalTransactionExecuter,

处理本地事物逻辑返回处理的事物状态LocalTransactionState3)  二阶段,处理完本地事物中业务得到事物状态, 根据offset查找到commitLog中的prepared消息,设置消息状态commitType或者rollbackType,

让后将信息添加到commitLog中, 其实二阶段生成了两条消息

事物消息发送


producer 5.消息在落地broker落地之普通消息

Broker根据producer请求的RequestCode.SEND_MESSAGE选择对应的处理器SendMessageProcessor

        根据请求消息内容构建消息内部结构MessageExtBrokerInner

        调DefaultMessageStore加消息写入commitlog


producer 6.消息在落地broker落地之事务消息

1. 消息落地

commitLog针对事物消息的处理,消息的第20位开始的八位记录是的消息在逻辑队列中的queueoffset,

但是针对事物消息为preparedType和rollbackType的存储的是事物状态表的索引偏移量

2. 分发事物消息:   

分发消息位置信息到ConsumeQueue: 事物状态为preparedType和rollbackType的消息不会将请求分发到ConsumeQueue中去,即不处理,所以不会被消息  更新transactionstable table:如果是prepared消息记,通过TransactionStateService服务将消息加到存储事务状态的表格tranStateTable的文件中;

如果是commitType和rollbackType消息, 修改事物状态表格tranStateTable中的消息状态。  记录Transaction Redo Log日志: 记录了 commitLogOffset, msgSize,preapredTransactionOffset, storeTimestamp。

3. 事物状态表

        事物状态表是有MapedFileQueue将多个文件组成一个连续的队列,它的存储单元是定长为24个字节的数据,

        tranStateTableOffset可以认为是事物状态消息的个数,索引偏移量, 它的值是 tranStateTable.getMaxOffset()/ TSStoreUnitSize   



3. 事物回查

定时回查线程会定时扫描(默认每分钟)每个存储事务状态的表格文件,遍历存储事务状态的表格记录如果是已经提交或者回滚的消息调过过,如果是prepared状态的如果消息小于事务回查至少间隔时间(默认是一分钟)跳出终止遍历调transactionCheckExecuter.gotocheck方法向producer回查事物状态,        根据group随机选择一台producer        查询消息,根据commitLogOffset和msgSize到commitlog查找消息        向Producder发起请求,请求code类型为CHECK_TRANSACTION_STATE,producer的DefaultMQProducerImpl.checkTransactionState()方法

来处理broker定时回调的请求,这里构建一个Runnable任务异步执行producer注册的回调接口,处理回调,在调endTransactionOneway向broker

发送请求更新事物消息的最终状态无Prepared消息,且遍历完,则终止扫描这个文件的定时任务


4. 事物消息的load&recover

TransactionStateService.load ()事物状态服务加载, 加载只是建立文件映射

redoLog队列恢复,加载本地redoLog文件

tranStateTable事物状态表, 加载本地tranStateTable文件

 recover:

正常恢复:

        利用tranRedoLog文件的recover

        利用tranStateTable文件重建事物状态表

异常恢复:

        先按照正常流程恢复TranRedo Log

        commitLog异常恢复,commitLog根据checkpoint时间点重新生成 redolog,重新分发消息DispatchRequest,

分发消息到位置信息到ConsumeQueue

                  更新Transaction State Table

                  记录TransactionRedo Log

        删除事物状态表tranStateTable

通过RedoLog全量恢复StateTable

        重头扫描RedoLog, 过滤出所有prepared状态的消息, 将commit或者rollback的消息对应的prepared消息删除

        重建StateTable,  将上面过滤出的prepared消息,添加到事物状态表文件中

这个事物状态表transstable的作用是定期(1分钟)将状态为prepared事物回查producer端redolog这个队列其实标记消费到哪了,

事物状态的恢复根本上是有commitlog来做的

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 205,386评论 6 479
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 87,939评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,851评论 0 341
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,953评论 1 278
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,971评论 5 369
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,784评论 1 283
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,126评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,765评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,148评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,744评论 2 323
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,858评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,479评论 4 322
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,080评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,053评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,278评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,245评论 2 352
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,590评论 2 343

推荐阅读更多精彩内容