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

Stella981
• 阅读 201

一、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,特别是异常处理的时候
点赞
收藏
评论区
推荐文章
刚刚好 刚刚好
2个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
晴空闲云 晴空闲云
2个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
1个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
Stella981 Stella981
1年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Wesley13 Wesley13
1年前
MySQL查询按照指定规则排序
1.按照指定(单个)字段排序selectfromtable_nameorderiddesc;2.按照指定(多个)字段排序selectfromtable_nameorderiddesc,statusdesc;3.按照指定字段和规则排序selec
Stella981 Stella981
1年前
Angular material mat
IconIconNamematiconcode_add\_comment_addcommenticon<maticonadd\_comment</maticon_attach\_file_attachfileicon<maticonattach\_file</maticon_attach\
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
helloworld_34035044 helloworld_34035044
5个月前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
helloworld_28799839 helloworld_28799839
2个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue