RocketMQ 消费者核心配置和核心知识

Posted 2021-03-07 jwen1994

一、RocketMQ4.X 消费者核心配置

• consumeFromWhere 配置（某些情况失效：参考https://blog.csdn.net/a417930422/article/details/83585397）这个配置基本不用改，采用默认配置即可。
  • CONSUME_FROM_FIRST_OFFSET： 初次从消息队列头部开始消费，即历史消息（还储存在 broker 的）全部消费一遍，后续再启动接着上次消费的进度开始消费。
  • CONSUME_FROM_LAST_OFFSET： 默认策略，初次从该队列最尾开始消费，即跳过历史消息，后续再启动接着上次消费的进度开始消费。
  • CONSUME_FROM_TIMESTAMP：从某个时间点开始消费，默认是半个小时以前，后续再启动接着上次消费的进度开始消费。
• allocateMessageQueueStrategy：负载均衡策略算法，即消费者分配到 queue 的算法
  • 默认值是 AllocateMessageQueueAveragely 即取模平均分配
• offsetStore：消息消费进度存储器 offsetStore 有两个策略：LocalFileOffsetStore 和 RemoteBrokerOffsetStore 
  • 广播模式默认使用 LocalFileOffsetStore， 集群模式默认使用 RemoteBrokerOffsetStore
• consumeThreadMax：最大消费线程池数量
• consumeThreadMin：最小消费线程池数量
• pullBatchSize：消费者去 broker 拉取消息时，一次拉取多少条。可选配置
• consumeMessageBatchMaxSize：单次消费时一次性消费多少条消息，批量消费接口才有用，可选配置
• messageModel：消费者消费模式
  • CLUSTERING：集群模式（默认配置）
  • BROADCASTING：广播模式

 二、集群和广播模式下 RocketMQ 消费端处理

Topic 下队列的奇偶数会影响 Customer 个数里面的消费数量

• 如果是4个队列，8个消息，4个节点则会各消费2条，如果不对等，则负载均衡会分配不均。
• 如果 consumer 实例的数量比 message queue 的总数量还多的话，多出来的 consumer 实例将无法分到 queue，也就无法消费到消息，也就无法起到分摊负载的作用，所以需要控制让 queue 的总数量大于等于 consumer 的数量。

集群模式(默认)：

• Consumer 实例平均分摊消费生产者发送的消息
• 例子：订单消息，一般是只被消费一次（被标记为同一个 ConsumerGroup 组的消费者不会对消息重复消费）

广播模式：

• 广播模式下消费消息：投递到 Broker 的消息会被每个 Consumer 进行消费，一条消息被多个 Consumer 消费，广播消费中 ConsumerGroup 暂时无用。
• 例子：群公告，每个人都需要消费这个消息

怎么切换模式：通过 setMessageModel()

三、RocketMQ 里面的 Tag 作用和消息过滤原理

一个 Message 只有一个 Tag，Tag 是二级分类。过滤分为 Broker 端和 Consumer 端过滤。

• Broker 端过滤，减少了无用的消息的进行网络传输，增加了 broker 的负担
• Consumer 端过滤，完全可以根据业务需求进行过滤，但是增加了很多无用的消息传输

一般是监听 * ，或者指定 tag，|| 运算，SLQ92，FilterServer 等；

• Tag 性能高，逻辑简单
• SQL92 性能差点，支持复杂逻辑（只支持 PushConsumer 中使用） MessageSelector.bySql
  • 语法：> ，<，=，IS NULL，AND，OR，NOT 等，sql where 后续的语法即可(大部分)

 生产者

@RequestMapping("/api/v1/pay_cb")
public Object callback( String tag, String amount) throws Exception {

    Message message = new Message(JmsConfig.TOPIC,tag, "",tag.getBytes());
    // 设置属性，用于sql过滤
    message.putUserProperty("amount",amount);
    
    SendResult sendResult =  payProducer.getProducer().send(message);
    System.out.printf("发送结果=%s, sendResult=%s 
", sendResult.getSendStatus(), sendResult.toString());
    return new HashMap<>();
}

消费者

package net.xdclass.xdclassmq.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class PayConsumer {


    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "pay_consumer_group";

    public  PayConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //多标签订阅
        //consumer.subscribe(JmsConfig.TOPIC, "order_pay || order_finish || order_create");

        //根据sql语法进行过滤消息
        consumer.subscribe(JmsConfig.TOPIC, MessageSelector.bySql(" amount > 5 "));

        consumer.registerMessageListener( new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);

                try {
                System.out.printf("%s 2 Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            } catch (Exception e) {
                System.out.println("消费异常");
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            }
        });

        consumer.start();
        System.out.println("consumer start ...");
    }

}

注意：消费者订阅关系要一致，不然会消费混乱，甚至消息丢失。订阅关系一致：订阅关系由 Topic 和 Tag 组成，同一个 group name，订阅的 Topic 和 Tag 必须是一样的。

在 Broker 端进行 MessageTag 过滤原理：遍历 message queue 存储的 message tag 和 订阅传递的 tag 的 hashcode 是否一样，不一样则跳过，符合的则传输给 Consumer，在 consumer queue 存储的是对应的 hashcode，对比也是通过 hashcode 对比；Consumer 收到过滤消息后也会进行匹配操作，但是是对比真实的 message tag 而不是 hashcode。

• consume queue 存储使用 hashcode 定长，节约空间
• 过滤中不访问 commit log，可以高效过滤
• 如果存在 hash 冲突，Consumer 端可以进行再次确认

建议：单一职责，多个队列；如果想使用多个 Tag，可以使用 sql 表达式，但是不建议。

常见错误：

The broker does not support consumer to filter message by SQL92

解决：broker.conf 里面配置如下
enablePropertyFilter=true

备注，修改之后要重启 Broker
master 节点配置：vim conf/2m-2s-async/broker-a.properties
slave 节点配置：vim conf/2m-2s-async/broker-a-s.properties

 

四、PushConsumer/PullConsumer 消费消息模式

Push 和 Pull 优缺点分析

• Push：实时性高；但增加服务端负载，消费端能力不同，如果 Push 推送过快，消费端会出现很多问题
• Pull：消费者从 Server 端拉取消息，主动权在消费者端，可控性好；但间隔时间不好设置，间隔太短，则空请求，浪费资源；间隔时间太长，则消息不能及时处理
• 长轮询： Client 请求 Server 端也就是 Broker 的时候， Broker 会保持当前连接一段时间，默认是15s，如果这段时间内有消息到达，则立刻返回给 Consumer；没消息的话，超过15s，则返回空，再进行重新请求；主动权在 Consumer 中，Broker 即使有大量的消息也不会主动推送给 Consumer。 缺点：服务端需要保持 Consumer 的请求，会占用资源，需要客户端连接数可控，否则会存在一堆连接

PushConsumer 本质是长轮训

• 系统收到消息后自动处理消息和 offset，如果有新的 Consumer 加入会自动做负载均衡,
• 在 broker 端可以通过 longPollingEnable=true 来开启长轮询
• 虽然是 push，但是代码里面大量使用了pull，是因为使用长轮训方式达到 push 效果，既有 pull 有的，又有 push 的实时性
• 优雅关闭：主要是释放资源和保存 Offset, 调用 shutdown() 即可 ，参考 @PostConstruct、@PreDestroy

PullConsumer 需要自己维护 Offset（参考官方例子）

• 官方源码包例子路径：org.apache.rocketmq.example.simple.PullConsumer
• 获取 MessageQueue 遍历
• 客户维护 Offset，需用用户本地存储 Offset，存储内存、磁盘、数据库等
• 处理不同状态的消息 FOUND、NO_NEW_MSG、OFFSET_ILLRGL、NO_MATCHED_MSG、4种状态
• 灵活性高可控性强，但是编码复杂度会高
• 优雅关闭：主要是释放资源和保存 Offset，需用程序自己保存好 Offset，特别是异常处理的时候