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

一、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，特别是异常处理的时候