RocketMQ - 生产者如何发送消息

时间:2023-11-21 10:35:04  热度:0°C

收藏小程序IT藏经楼大量资源免费分享

创建Topic的时候为何要指定MessageQueue数量

首先想要搞明白生产者的工作原理,那么就必须先明白一个概念,MessageQueue是什么?

而要明白MessageQueue是什么,就必须把他跟Topic以及Broker综合起来看,才能搞明白。如果我们要使用RocketMQ,你先部署出来一套RocketMQ集群这个肯定是必须的,有了集群之后,就必须根据你的业务需求去创建一些Topic。

像我们创建的这些Topic就可以在上面提到的rocketmq-console控制台去创建,在创建Topic的时候,需要指定一个很关键的参数,就是MessageQueue。

简单来说,就是你要指定你和这个Topic对应了多少个队列,也就是多少个MessageQueue。

那么这个MessageQueue是用来干嘛的?

Topic、MessageQueue以及Broker之间的关系

假设我们现在创建了一个Topic,指定了4个MessageQueue,那么这个Topic的数据在Broker集群中是如何分布的?

每个Topic的数据都是分布式存储在多个Broker中的,那么我们如何决定这个topic的哪些数据放在这个broker上,哪些数据放在那个broker上呢?

所以RocketMQ引入了MessageQueue的概念,本质上就是一个数据分片的机制。在这个机制中,假设Topic有1万条数据,然后Topic有4个MessageQueue,那么大致可以认为会在每个MessageQueue中放入2500条数据。

当然,这不是绝对的,有可能有MessageQueue得到数据多,有的数据少,这个要根据你的消息写入MessageQueue的策略来定。

我们可以先假定每个MessageQueue中会平均分配Topic的数据吧,那么我们有4个MessageQueue平均分配了Topic的数据,这些MessageQueue放在哪里?

当然是放在Broker上了,也就是说很可能就是2个broker上,每个broker放两个MessageQueue。

所以其实MessageQueue就是RocketMQ中非常关键的一个数据分片机制,他通过MessageQueue将一个Topic的数据拆分为了很多个数据分片,然后在每个Broker机器上都存储一些MessageQueue。

通过这个方法,就可以实现Topic数据的分布式存储。

生产者发送消息的时候写入哪个MessageQueue

接着要思考一个问题,就是生产者在发送消息的时候,会写入哪个MessageQueue中?

要解决这个问题,大家首先要记得一个重要的点,生产者会跟NameServer进行通信获取Topic的路由数据。所以生产者从NameServer中就会知道,一个Topic有几个MessageQueue,哪些MessageQueue在哪台Broker机器上,哪些MessageQueue在另外一台Broker机器上。

现在我们暂时先认为生产者会均匀的把消息写入各个MessageQueue中,就是比如这个生产者发送出去20条数据,那么4个MessageQueue就是每个都会写入5条数据。

通过这个方法,就可以让生产者把写入请求分散给多个broker,可以让每个broker都均匀分摊到一定的写入请求压力。

假设单个broker可以抗每秒7万并发,那么两个broker就可以抗到每秒14万并发,这样就可以实现RocketMQ集群抗下每秒10万+超高并发的场景了。

另外通过这个方法,也可以让一个Topic中的数据分散在多个MessageQueue中,进而分散在多个Broker机器上,就可以实现RocketMQ集群分布式存储海量的消息数据了。

如果某个broker出现故障该怎么办

如果某个Broker临时出现故障了,比如Master Broker挂了,此时正在等待的其他Slave Broker自动热切换为Master Broker,那么这个时候对这一组Broker就没有Master Broker可以写入了。

此时如果还是按照之前的策略来均匀把数据写入各个Broker上的MessageQueue,那么会导致你在一段时间内,每次访问到这个挂掉的Master Broker都会访问失败。

对于这个问题,通常来说建议大家在Producer中开启一个开关,就是sendLatencyFaultEnable。

一旦打开了这个开关,那么他会有一个自动容错机制,比如如果某次访问一个Broker发现网络延迟有500ms,然后还无法访问,那么就会自动回避访问这个Broker一段时间,比如接下来的300ms内,就不会访问这个Broker了。

这样的话,就可以避免一个Broker故障之后,短时间内生产者频繁的发送消息到这个故障的Broker上去,出现较多次的异常,而是在一个Broker故障之后,自动回避一段时间不要访问这个Broker,过段时间再去访问他。

那么这样过一段时间之后,可能这个Master Broker就已经恢复好了,比如他的Slave Broker已经切换为了Master可以让别人访问了。

免责声明:
1. 《RocketMQ - 生产者如何发送消息》内容来源于互联网,版权归原著者或相关公司所有。
2. 若《65667893文库网》收录的文本内容侵犯了您的权益或隐私,请立即通知我们删除。