当前位置：首页>后端>正文

RocketMQ之入门学习消费者基本API使用（三）

后端2024-04-30 02:14:41

前言

主要学习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拉取以及定时器消费消息。

查看全文

https://www.xamrdz.com/backend/3gt1938940.html

前言

PushConsumer核心参数详解

Consumer消费者集群模式

Consumer消费者广播模式

消息存储偏量移动offset

Consumer长轮询模式

消费者

结尾

相关文章：