RocketMQ - 在RocketMQ的生产实践中积累的各种一手经验总结

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

（1）灵活的运用tags来过滤数据

在真正的生产项目中，建议大家合理的规划Topic和里面的tags，一个Topic代表了一类业务消息数据，然后对于这类业务得到消息数据，如果希望继续划分一些类别的话，可以在发送消息的时候设置tags。

举个例子，比如我们都知道现在常见的外卖平台有美团、饿了么，还有别的一些外卖，那么假如你现在一个系统要发送外卖订单数据到MQ里去，就可以针对性的设置tags，比如不同的外卖数据都到一个"waimaiOrderTopic"里去，但是不同类型的外卖可以有不同的tags，meituan_waimai、eleme_waimai等等。

然后对于消费"WaimaiOrderTopic"的系统，可以根据tags来筛选，可能你就需要某一种类别的外卖数据就可以了。

（2）基于消息Key来定位消息是否丢失

在消息零丢失方案中，可能要解决的是消息是否丢失的问题，那么如果消息真的丢失了，我们是不是要排查？此时是不是要从MQ里查一下，这个消息是否丢失了？

那么怎么从MQ里查询消息是否丢失呢？可以基于消息key来实现，比如通过下面方式设置一个消息的key为订单id，message.setKeys(orderId); 这样这个消息就具备一个key了。

接着这个消息到Broker上，会基于key构建hash索引，这个hash索引就存放在IndexFile索引文件里。

然后后续我们可以通过MQ提供的命令去根据key查询这个消息，类似下面这样

mqadmin queryMsgByKey -n 127.0.0.1:9876 -t SCANRECORD -K orderId

具体的命令可以查官方手册

（3）提高消费者的吞吐量

如果消费的时候发现消费的比较慢，那么可以提高消费者的并行度，常见的就是部署更多的consumer机器。

但是这里要注意，你的Topic得到MessageQueue的对应有所增加，因为如果你Consumer机器有5台，然后MessageQueue只有4个，那就意味着有一个consumer机器是获取不到消息的。

然后就是可以增加consumer的线程数量，可以设置consumer端的参数: consumeThreadMin、consumeThreadMax，这样一台consume机器上的消费线程越多，消费的速度越快。

此外，还可以开启消费者的批量消费功能，就是设置
consumeMessageBatchMaxSize参数，他默认是1，但是你可以设置的多一些，那么一次就会给你回调函数一批消息来给你处理了。

（4）要不要消费历史消息

其实consumer是支持设置从哪里开始消费消息的，常见的有两种，一个是从Topic的第一条数据开始消费，一个是从最后一次消费过的消息之后开始消费，对应的是：CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET。