开发者社区 > 云原生 > 云消息队列 > 正文

请问下rocketMQ中怎么去获取mq的偏移量?

请问下rocketMQ中怎么去获取mq的偏移量?网上找的没搞明白。即批量检查自建(容器)rocketmq消息消费是否消费完(消费组总量-偏移量)?

展开
收起
小易01 2023-07-19 19:52:44 317 0
2 条回答
写回答
取消 提交回答
  • 要获取RocketMQ的偏移量,您可以使用RocketMQ提供的OffsetStore类来获取消费者组的偏移量信息。

    以下是一个示例代码片段,展示如何获取消费者组在指定主题(Topic)和队列(Queue)上的偏移量:

    import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
    import org.apache.rocketmq.client.consumer.MQConsumer;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.message.MessageQueue;
    
    public class OffsetExample {
        public static void main(String[] args) throws MQClientException {
            // 创建消费者实例
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("your_consumer_group_name");
    
            // 设置NameServer地址
            consumer.setNamesrvAddr("your_nameserver_address");
    
            // 启动消费者
            consumer.start();
    
            // 获取指定Topic和Queue的偏移量
            MessageQueue queue = new MessageQueue("your_topic", "your_queue", 0);
            long offset = consumer.fetchConsumeOffset(queue, false);
    
            System.out.println("Offset for Topic: your_topic, Queue: your_queue = " + offset);
    
            // 关闭消费者
            consumer.shutdown();
        }
    }
    

    在上述示例中,我们创建了一个DefaultMQPullConsumer对象,并设置消费者组名、NameServer地址等属性。然后,通过调用fetchConsumeOffset()方法来获取指定Topic和Queue的偏移量。最后,我们打印出偏移量。

    请注意,这里使用的是DefaultMQPullConsumer,它是基于拉取模式的消费者,与推模式(PushConsumer)不同。您可以根据实际情况选择使用合适的消费者类型。

    2023-07-28 13:41:24
    赞同 展开评论 打赏
  • 意中人就是我呀!

    admin 接口中 examineConsumeStats 可以获取 lag = broker offset - consumer offset
    448f7ed6ebf32be9a5e67ee0fc176a76.png此回答整理至钉群“群2-Apache RocketMQ 中国开发者钉钉群”。

    2023-07-19 23:00:00
    赞同 展开评论 打赏

涵盖 RocketMQ、Kafka、RabbitMQ、MQTT、轻量消息队列(原MNS) 的消息队列产品体系,全系产品 Serverless 化。RocketMQ 一站式学习:https://rocketmq.io/

相关产品

  • 云消息队列 MQ
  • 热门讨论

    热门文章

    相关电子书

    更多
    RocketMQ Client-GO 介绍 立即下载
    RocketMQ Prometheus Exporter 打造定制化 DevOps 平台 立即下载
    基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台 立即下载