《RocketMQ实战与原理解析》
前言
本章介绍在大流量场景下,提高 RocketMQ 集群吞吐量的一些方法,有些方法当服务器出异常时会增大丢消息的概率,用户需要根据业务需求酌情使用。
在 Broker端进行消息过滤
在 Broker 端进行消息过滤,可以减少无效消息发送到 Consumer ,少占用网络带宽从而提高吞吐量。 Broker 端有三种方式进行消息过滤。
消息的Tag和Key
对一个应用来说,尽可能只用一个 Topic ,不同的消息子类型用 Tag 来标识(每条消息只能有一个 Tag ),服务器端基于 Tag 进行过滤,并不需要读取消息体的内容,所以效率很高。发送消息设置了 Tag 以后,消费方在订阅消息时,才可以利用 Tag 在 Broker 端做消息过滤。
其次是消息的Key。对发送的消息设置好 Key ,以后可以根据这个 Key 来查找消息。所以这个 Key 一般用消息在业务层面的唯一标识码来表示,这样后续查询消息异常，消息丢失等都很方便 Broker 会创建专门的索引文件，来存 Key 到消息的映射，由于是哈希索引，应尽 Key ，避免潜在的哈希冲突。
Tag 和 Key 的主要差别是使用场景不同， Tag 用在 Consumer 的代码中，用来进行服务端消息过滤， Key 主要用于通过命令行查询消息。
通过 Tag 进行过滤
用 Tag 方式进行过滤的方法是传人感兴趣的 Tag 标签，Tag 标签是一个普通字符串，是在创建 Message 的时候添加的， Message 只能有一个 Tag 使用 Tag 方式过滤非常高效， Broker 端可以在 ConsumeQueue 中做这种过洁 只从 CommitLog 里读取过滤后被命中的消息 ConsumerQueue 的存储 格式，如图所示
image
Consume Queue 部分存储的是 Tag 对应的 hashcode ，是一个定长的 字符串，通过 Tag 过滤的过程就是对比定长的 hashcode。经过 hashcode 对比, 符合要求的消息被从 CommitLog 读取出来，不用担心 Hash 冲突问题，消息在被消费前，会对比完整的 Message Tag 字符串，消除 Hash 冲突造成的误读。
用 SQL 表达式的方式进行过滤
使用 Tag 方式过滤虽然高效，但是支持的逻辑比较简单，在构造 Message 的时候，还可以通过 putUserProperty 函数来增加多个自定义的属性，基于这些属性可以做复杂的过逻辑。
Message msg =new Message("TopicTest", tag, ("Hello RocketMQ " + i).getBytes(RemotingHelpe.DEFAULT_CHARSET));
/// Set some properties .
msg.putUserProperty("a", String.valueOf(i));
msg.putUserProperty("b", "hello");
代码中这个消息就有了两个特殊的属性值 a 和 b ，我们用类似 SQL 表达式的方式对消息进行过滤，用法如下(目前只支持在 PushConsumer 中实现这种过滤) :
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pleas
rename_unique_group_name_4");
// only subsribe messages have property a , also a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3 ");
consumer.registerMessageListener(new MessageListenerConcurrently(){
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs , ConsumeConcurrentlyContext context){
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
类似 SQL 的过滤表达式,支持如下语法:
- 数字对比,比如>、>=、<、<=、 BETWEEN、=
- 字符串对比,比如=、<>、IN;
- IS NULL or IS NOT NULL
- 逻辑符号AND、OR、NOT。
支持的数据类型:
- 数字型,比如123、3.1415;
- 字符型,比如'abc'、注意必须用单引号;
- NULL,这个特殊字符;
- 布尔型, TRUEOrFALSE。
SQL 表达式方式的过滤需要 Broker先读出消息里的属性内容,然后做SQL计算,增大磁盘压力,没有 Tag 方式高效。
Filter Server方式过滤
Filter Server 种比 SQL 达式更灵活的过滤方式，允许用户自定义 Java 函数，根据 Java 逻辑对消息进行过滤。
使用 Filter Server , 首先要在启动 Broker 在配置文件里加上 filterServerNums = 3 这样的配置, Broker 启动的时, 就会在本机启动 3 个 Filter Server 进程。Filter Server 类似一个 RocketMQ 的 Consumer 进程，它从本机 Broker 获取消息，然后根据用户上传过来的 Java 进行过滤，过滤后的消息再传给远端的 Consumer。这种方式会占用很多 Broker 机器的 CPU 资源，要根据实际情况谨慎使用。上传的 java 代码也要经过检查，不能有申请大内存、创建线程等这样的操作，否则容易造成 Broker 服务器宕机。实现过滤逻辑的示例, 如代码清单所示。
public class MessageFilterImpl implements MessageFilter {
@Override
public boolean match(MessageExt messageExt, FilterContext filterContext) {
String property = messageExt.getUserProperty("SequenceId");
if (property != null) {
int id = Integer.parseInt(property);
if ((id % 3) == 0 && (id > 10)) {
return true;
}
}
return false;
}
}
上面代码实现了过滤逻辑，它是根据消息的“Sequenceld”这个属性来过滤的，其实不一定要根据消息属性来过滤，也可以根据消息体的内容或其他特征过滤，如代码清单所示。
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Consumer-GroupNamecc4");
// 使用Java代码,在服务器做消息过滤
String filterCode = MixAll.file2String("/home/admin/MessageFilterImpl.java");
consumer.subscribe("TopicFilter7", "com.alibaba.rocketmqexample.filter.MessageFilterImpl", filterCode);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
}
在使用 Filter Server 的 Consumer 例子中，主要是把实现过滤逻辑的类作为参数传到 Broker 端， Broker 端的 Filter Server 会解析这个类，然后根据 match 函数里的逻辑进行过滤。
提高 Consumer 处理能力
当 Consumer 的处理速度眼不上消息的产生速度，会造成越来越多的消息积压，这个时候首先查看消费逻辑本身有没有优化空间，除此之外还有三种方法可以提高 Consumer 的处理能力。
(1) 提高消费并行度
在同一个 Consumer Group 下 ( Clustering方式), 可以通过增加 Consumer 实例的数量来提高并行度, 通过加机器, 或者在已有机器中启动多个 Consumer 进程都可以增加 Consumer实例数。注意总的 Consumer 数量不要超过 Topic 下 Read Queue 数量, 超过的 Consumer 实例接收不到消息。此外,通过提高单个 Consumer 实例中的并行处理的线程数, 可以在同一个 Consumer 内增加并行度来提高吞吐量(设置方法是修改 consumeThreadMin 和 consumeThreadMax)。
(2) 以批量方式进行消费
某些业务场景下,多条消息同时处理的时间会大大小于逐个处理的时间总和,比如消费消息中涉及update 某个数据库,一次 update 10 条的时间会大大小于 10 次 update 1 条数据的时间。这时可以通过批量方式消费来提高消费的吞吐量。实现方法是设置 Consumer的 consumeMessageBatchMaxSize 这个参数, 默认是 1 , 如果设置为N,在消息多的时候每次收到的是个长度为N的消息链表。
(3) 检测延时情况, 跳过非重要消息
Consumer在消费的过程中,如果发现由于某种原因发生严重的消息堆积短时间无法消除堆积,这个时候可以选择丢弃不重要的消息,使 Consumer 尽快追上 Producer 的进度,如代码清单所示。
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
long offset = msgs.get(0).getQueueOffset();
String maxOffset = msgs.get(0).getProperty("MAX_OFFSET");
long diff = Long.parseLong(maxOffset) - offset;
if (diff > 90000) {
// 不处理直接返回成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else {
System.out.println("正常消费消息");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
});
如代码所示,当某个队列的消息数堆积到 90000 条以上,就直接丢弃,以便快速追上发送消息的进度。
Consumer 的负载均衡
节中讲到,想要提高 Consumer的处理速度,可以启动多个 Consumer 并发处理,这个时候就涉及如何在多个 Consumer之间负载均衡的问题,接下来结合源码分析 Consumer的负载均衡实现。
要做负载均衡,必须知道一些全局信息,也就是一个 Consumer Group里到底有多少个 Consumer,知道了全局信息,才可以根据某种算法来分配,比如简单地平均分到各个 Consumer在 RocketMQ 中,负载均衡或者消息分配是在 Consumer端代码中完成的, Consumer从 Broker 处获得全局信息, 然后自己做负载均衡,只处理分给自己的那部分消息。
DefaultMQPushConsumer的负载均衡
DefaultMQPushConsumer 的负载均衡过程不需要使用者操心,客户端程序会自动处理,每个 DefaultMQPush Consumer启动后,会马上会触发一个 doRebalance 动作; 而且在同一个 Consumer Group 里加入新的 DefaultMQPushConsumer 时,各个 Consumer 都会被触发 doRebalance 动作。
默认用的是 AllocateMessageQueueAveragely。AllocateMessageQueueAveragely 平均分配策略; AllocateMessageQueueAveragelyByCircle 环形分配策; AllocateMessageQueueByConfig 手动配置分配策略; AllocateMessageQueueByMachineRoom 机房分配策略。
image
负载均衡的结果与 Topic的 Message Queue 数量,以及 Consumer Group 里的 Consumer 的数量有关。负载均衡的分配粒度只到 Message Queue, 把 Topic 下的所有 Message Queue 分配到不同的 Consumer 中, 所以 Message Queue 和 Consumer 的数量关系,或者整除关系影响负载均衡结果。
以 AllocateMessageQueueAveragely 策略为例,如果创建 Topic 的时候,把 MessageQueue 数设为 3 ,当 Consumer 数量为 2 的时候,有一个 Consumer需要处理 Topic 三分之二的消息, 另一个处理三分之一的消息; 当 Consumer 数量为 4 的时候, 有一个 Consumer 无法收到消息, 其他 3 个 Consumer各处理 Topic 三分之一的消息。可见 Message Queue 数量设置过小不利于做负载均衡, 通常情况下,应把一个 Topic的 Message Queue 数设置为 16。
DefaultMQPulLConsumer的负载均衡
Pull Consumer 可以看到所有的 Message Queue, 而且从哪个 Message Queue 读取消息,读消息时的 Offset 都由使用者控制, 使用者可以实现任何特殊方式的负载均衡。
DefaultMQPullConsumer 有两个辅助方法可以帮助实现负载均衡, 一个是 registerMessageQueueListener 函数,如代码清单所示。registerMessageQueueListener 函数在有 Consumer 加人或退出时被触发。
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name");
consumer.registerMessageQueueListener("", new MessageQueueListener() {
@Override
public void messageQueueChanged(String s, Set<MessageQueue> set, Set<MessageQueue> set1) {
// todo
}
});
另一个辅助工具是 MQPullConsumerScheduleService 类，使用这个 Class 类似使用 DefaultMQPushConsumer ，但是它把 Pull 消息的 主动性留给了使用者，如代码清单所示。
public static void main(String[] args) throws MQClientException {
final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("please_rename_unique_group_name");
scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(RocketMqConfig.NAMESRV_ADDR);
scheduleService.setMessageModel(MessageModel.CLUSTERING);
scheduleService.registerPullTaskCallback("testPullConsumer", new PullTaskCallback() {
@Override
public void doPullTask(MessageQueue messageQueue, PullTaskContext pullTaskContext) {
MQPullConsumer consumer = pullTaskContext.getPullConsumer();
try {
long offset = consumer.fetchConsumeOffset(messageQueue, false);
if (offset < 0) {
offset = 0;
}
PullResult pullResult = consumer.pull(messageQueue, "*", offset, 32);
System.out.printf("%s%n", +offset + "\t" + messageQueue + "\t" + pullResult);
switch (pullResult.getPullStatus()) {
case FOUND:
break;
case NO_NEW_MSG:
break;
case NO_MATCHED_MSG:
break;
case OFFSET_ILLEGAL:
break;
default:
break;
}
consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
pullTaskContext.setPullNextDelayTimeMillis(1000);
} catch (MQClientException | InterruptedException | RemotingException | MQBrokerException e) {
e.printStackTrace();
}
}
});
scheduleService.start();
}
然后我们看一看在 MQPullConsumerScheduleService 类的实现里, 实现负载均衡的代码, 如下代码清单所示。
class MessageQueueListenerImpl implements MessageQueueListener {
MessageQueueListenerImpl() {
}
public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
MessageModel messageModel = MQPullConsumerScheduleService.this.defaultMQPullConsumer.getMessageModel();
switch(messageModel) {
case BROADCASTING:
MQPullConsumerScheduleService.this.putTask(topic, mqAll);
break;
case CLUSTERING:
MQPullConsumerScheduleService.this.putTask(topic, mqDivided);
}
}
}
从源码中可以看出, 用户通过更改 MessageQueueListenerImpl 的实现来做自己的负载均衡策略。
提高 Producer 的发送速度
发送一条消息出去要经过三步, 一是客户端发送请求到服务器, 二是服务器处理该请求, 三是服务器向客户端返回应答, 一次消息的发送耗时是上述三个步骤的总和。在一些对速度要求高,但是可靠性要求不高的场景下, 比如日志收集类应用, 可以采用 Oneway 方式发送, Oneway 方式只发送请求不等待应答, 即将数据写入客户端的 Socket 缓冲区就返回, 不等待对方返回结果, 用这种方式发送消息的耗时可以缩短到微秒级。
另一种提高发送速度的方法是增加 Producer 的并发量,使用多个 Producer 同时发送, 我们不用担心多 Producer 同时写会降低消息写磁盘的效率 RocketMQ 引入了一个并发窗口,在窗口内消息可以并发地写入 DirectMem 中, 然后异步地将连续一段无空洞的数据刷入文件系统当中。顺序写 CommitLog 可让 RocketMQ 无论在 HDD 还是 SSD 磁盘情况下都能保持较高的写入性能。目前在阿里内部经过调优的服务器上, 写入性能达到90万+的TPS, 我们可以参考这个数据进行系统优化。
在 Linux 操作系统层级进行调优,推荐使用EXT4文件系统, IO调度算法使用 deadline 算法。
如图所示,EXT4 创建/删除文件的性能比 EXT3 及其他文件系统要好, RocketMQ的 Commitlog会有频繁的创建/删除动作。
image
另外, IO 调度算法也推荐调整为 deadline。deadline 算法大致思想如下 : 实现四个队列,其中两个处理正常的 read 和 write 操作, 另外两个处理超时的 read 和 write 操作。正常的 read 和 write 队列中, 元素按扇区号排序,进行正常的 IO 合并处理以提高吞吐量。因为 IO 请求可能会集中在某些磁盘位置,这样会导致新来的请求一直被合并,可能会有其他磁盘位置的IO请求被饿死。超时的 read 和 write 的队列中, 元素按请求创建时间排序, 如果有超时的请求出现, 就放进这两个队列,调度算法保证超时(达到最终期限时间)的队列中的 IO 请求会优先被处理。
网友评论