前言
主要学习RocketMQ生产者的基本API使用，以及各种投递消息模式，消息的返回状态等等。
PushConsumer核心参数详解
- consumerFromWhere 在broker记录的offset进行消费,会有一下几种方式
public enum ConsumeFromWhere {
CONSUME_FROM_LAST_OFFSET,
@Deprecated
CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST,
@Deprecated
CONSUME_FROM_MIN_OFFSET,
@Deprecated
CONSUME_FROM_MAX_OFFSET,
CONSUME_FROM_FIRST_OFFSET,
CONSUME_FROM_TIMESTAMP,
}
// 调用方式
- allocateMessageQueueStrategy 消费策略（采用默认平均方式）
（1）AllocateMachineRoomNearby：相近机房策略
（2）AllocateMessageQueueConsistentHash：一致性hash策略
（3）AllocateMessageQueueByMachineRoom：根据机房hash
（4）AllocateMessageQueueByConfig：配置策略
（5）AllocateMessageQueueAveragelyByCircle：hash轮询策略
（6）AllocateMessageQueueAveragely：平均策略 - subscription 订阅主题，正则表达过滤
consumer.subscribe("test_quick_topic", "*");
- offsetStore （分本地和远程）
- consumeThreadMin/consumThreadMax 线程池，可以自动调节
- consumeConcurrentlyMaxSpan/pullThreasholdForQueue 对流进行控制
- pullinterval/pullBatchSize 消息拉取时间/ 消息拉取数据（默认90）
- consumeMessageBatchMaxSize 默认是1
Consumer消费者集群模式
- 默认是Clustering模式
- GroupName用于把舵个Consumer组织到一起
- 相同GroupName的Consumer只消费所订阅消息的一部分
- 目的: 达到天然的负载均衡机制
Consumer消费者广播模式
- Broadcasting模式（广播模式）
- 同一个consumerGroup的Consumer都消费订阅Topic全部信息，
RocketMQ 中的消息会被消费组中的每个消费者都消费一次 - setMessageModel方法
consumer.setMessageModel(MessageModel.BROADCASTING);
- 示例
producer
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
String group_name = "test_producer_name";
DefaultMQProducer producer = new DefaultMQProducer(group_name);
producer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
producer.start();
for (int i = 0; i < 10; i++) {
try {
String tag = (i % 2 ==0) ? "TagA":"TagB";
Message msg = new Message("test_topic2",// topic
tag,// tag
("信息内容" + i).getBytes()// body
);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
}
发送10条消息
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27730000, offsetMsgId=C0A8010300002A9F000000000005C824, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=2], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A277E0001, offsetMsgId=C0A8010300002A9F000000000005C8D5, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=3], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27800002, offsetMsgId=C0A8010300002A9F000000000005C986, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27860003, offsetMsgId=C0A8010300002A9F000000000005CA37, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27880004, offsetMsgId=C0A8010300002A9F000000000005CAE8, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=2], queueOffset=9]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A278A0005, offsetMsgId=C0A8010300002A9F000000000005CB99, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=3], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A278C0006, offsetMsgId=C0A8010300002A9F000000000005CC4A, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A278E0007, offsetMsgId=C0A8010300002A9F000000000005CCFB, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=1], queueOffset=9]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27900008, offsetMsgId=C0A8010300002A9F000000000005CDAC, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=2], queueOffset=10]
SendResult [sendStatus=SEND_OK, msgId=C0A8010342E818B4AAC20D8A27970009, offsetMsgId=C0A8010300002A9F000000000005CE5D, messageQueue=MessageQueue [topic=test_topic2, brokerName=SC-202111172212, queueId=3], queueOffset=9]
##### consumer1
public class Consumer1 {
public Consumer1() {
try {
String group_name = "test_consumer_name1";
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group_name);
consumer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
consumer.subscribe("test_topic2", "TagB || TagA");
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new Listener());
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}
class Listener implements MessageListenerConcurrently {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
try {
for(MessageExt msg : msgs){
String topic = msg.getTopic();
String msgBody = new String(msg.getBody(),"utf-8");
String tags = msg.getTags();
System.out.println("收到消息：" + " topic :" + topic + " ,tags : " + tags + " ,msg : " + msgBody);
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
public static void main(String[] args) {
Consumer1 c1 = new Consumer1();
System.out.println("c1 start..");
}
}
打印日志
收到消息： topic :test_topic2 ,tags : TagB ,msg : 信息内容1
收到消息： topic :test_topic2 ,tags : TagA ,msg : 信息内容0
收到消息： topic :test_topic2 ,tags : TagA ,msg : 信息内容8
收到消息： topic :test_topic2 ,tags : TagA ,msg : 信息内容2
收到消息： topic :test_topic2 ,tags : TagB ,msg : 信息内容3
收到消息： topic :test_topic2 ,tags : TagA ,msg : 信息内容4
收到消息： topic :test_topic2 ,tags : TagB ,msg : 信息内容5
收到消息： topic :test_topic2 ,tags : TagB ,msg : 信息内容7
收到消息： topic :test_topic2 ,tags : TagA ,msg : 信息内容6
收到消息： topic :test_topic2 ,tags : TagB ,msg : 信息内容9
consumer2 也是跟上面一个复制一份名字叫test_consumer_name2开启监听得到同样的效果如果想要针对Tag做过滤的话，
consumer1 针对TagA
consumer.subscribe("test_topic2", "TagA");
consumer2 针对TagB
consumer.subscribe("test_topic2", "TagA");
消息存储偏量移动offset
- Offset是消息消费进度的核心
- Offset指某个topic下的一条消息在某个MessageQueue里的位置
- 通过Offset可以进行定位到这条消息
- offset的存储实现分为远程文件类型和本地文件类型两种
集群模式是用远程文件存储offset，例如一个producer发送10条记录，两个consumer负载均衡平均5条记录，这是通过broker去控制offset，是哦那个RemoteBrokerOffsetStore
广播模式是本地文件存储，因为每个Consumer都会被消费，没有干扰，
Consumer长轮询模式
- DefalutPushConsumer是使用长轮询进行实现
- 通常主流消息获取模式：Push消息推送模式 & Pull消息拉取模式
- Pull 方式主要做了三件事：
- 获取Message Queue并遍历
- 维护OffsetStore
- 根据不同的消息状态做不同的处理
- 示例
producer
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
String group_name = "test_pull_producer_name";
DefaultMQProducer producer = new DefaultMQProducer(group_name);
producer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
producer.start();
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("test_pull_topic",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes()// body
);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
Thread.sleep(1000);
}
catch (Exception e) {
e.printStackTrace();
Thread.sleep(3000);
}
}
producer.shutdown();
}
}
消费者
每次回去拉取数据，直到拉取自动关闭，但重新拉取的时候还是从头开始拉，没有记录offset位置
public class PullConsumer {
//Map<key, value> key为指定的队列，value为这个队列拉取数据的最后位置
private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();
public static void main(String[] args) throws MQClientException {
String group_name = "test_pull_consumer_name";
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(group_name);
consumer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
consumer.start();
System.err.println("consumer start");
// 从TopicTest这个主题去获取所有的队列（默认会有4个队列）
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("test_pull_topic");
// 遍历每一个队列，进行拉取数据
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ: while (true) {
try {
// 从queue中获取数据，从什么位置开始拉取数据 单次对多拉取32条记录
PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
System.out.println(pullResult);
System.out.println(pullResult.getPullStatus());
System.out.println();
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
switch (pullResult.getPullStatus()) {
case FOUND:
List<MessageExt> list = pullResult.getMsgFoundList();
for(MessageExt msg : list){
System.out.println(new String(msg.getBody()));
}
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
System.out.println("没有新的数据啦...");
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}
consumer.shutdown();
}
private static void putMessageQueueOffset(MessageQueue mq, long offset) {
offseTable.put(mq, offset);
}
private static long getMessageQueueOffset(MessageQueue mq) {
Long offset = offseTable.get(mq);
if (offset != null)
return offset;
return 0;
}
}
模拟定时器去拉取消息并定时更新下一个offset位置
public class PullScheduleService {
public static void main(String[] args) throws MQClientException {
String group_name = "test_pull_consumer_name";
final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService(group_name);
scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
scheduleService.setMessageModel(MessageModel.CLUSTERING);
scheduleService.registerPullTaskCallback("test_pull_topic", new PullTaskCallback() {
@Override
public void doPullTask(MessageQueue mq, PullTaskContext context) {
MQPullConsumer consumer = context.getPullConsumer();
System.err.println("-------------- queueId: " + mq.getQueueId() + "-------------");
try {
// 获取从哪里拉取
long offset = consumer.fetchConsumeOffset(mq, false);
if (offset < 0)
offset = 0;
PullResult pullResult = consumer.pull(mq, "*", offset, 32);
switch (pullResult.getPullStatus()) {
case FOUND:
List<MessageExt> list = pullResult.getMsgFoundList();
for(MessageExt msg : list){
//消费数据...
System.out.println(new String(msg.getBody()));
}
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
break;
default:
break;
}
consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
// 设置再过3000ms后重新拉取
context.setPullNextDelayTimeMillis(3000);
}
catch (Exception e) {
e.printStackTrace();
}
}
});
scheduleService.start();
}
}
日志打印,每次生产者发，定时任务都回更新记录并及时打印下一个offset
Hello RocketMQ 5
Hello RocketMQ 6
Hello RocketMQ 4
-------------- queueId: 3-------------
-------------- queueId: 1-------------
-------------- queueId: 2-------------
Hello RocketMQ 9
-------------- queueId: 0-------------
-------------- queueId: 3-------------
Hello RocketMQ 8
Hello RocketMQ 7
结尾
主要学习了对消费端基本API的使用，也了解了集群消费和广播消费模式，以及基本的offset偏移量概念，也分为远程文件存储和本地文件存储方式。还有Pull拉取以及定时器消费消息。
