码迷,mamicode.com
首页 > 其他好文 > 详细

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

时间:2020-02-26 01:32:25      阅读:138      评论:0      收藏:0      [点我收藏+]

标签:配置   clu   网络传输   大量   最大   常见错误   setname   设置   const   

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

 

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

标签:配置   clu   网络传输   大量   最大   常见错误   setname   设置   const   

原文地址:https://www.cnblogs.com/jwen1994/p/12364557.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!