首页
学习
活动
专区
工具
TVP
发布
您找到你想要的搜索结果了吗?
是的
没有找到

Kafka Consumer重置Offset

在Kafka Version为0.11.0.0之后,Consumer的Offset信息不再默认保存在Zookeeper上,而是选择用Topic的形式保存下来。...在命令行中可以使用kafka-consumer-groups的脚本实现Offset的相关操作。 更新Offset由三个维度决定:Topic的作用域、重置策略、执行方案。...Topic的作用域 --all-topics:为consumer group下所有topic的所有分区调整位移) --topic t1 --topic t2:为指定的若干个topic的所有分区调整位移...确定执行方案 什么参数都不加:只是打印出位移调整方案,不具体执行 --execute:执行真正的位移调整 --export:把位移调整方案按照CSV格式打印,方便用户成csv文件,供后续直接使用 注意事项 consumer...group状态必须是inactive的,即不能是处于正在工作中的状态 不加执行方案,默认是只做打印操作 常用示例 更新到当前group最初的offset位置 bin/kafka-consumer-groups.sh

9.5K40

RocketMQ详解(10)——Consumer详解

(this); //启动消费端 consumer.start(); log.info("Message Consumer Start...MessageModel.BROADCASTING——广播模式:同一个ConsumerGroup下的每个Consumer都能消费到所订阅Topic的所有消息,也就是一个消息会被多次分发,被多个Consumer...如果不想消费某个Topic下的所有消息,可以通过指定Tag进行消息过滤,如Consumer.subscribe(“TopicTest”,”tags1 || tag2 || tag3”),表示这个Consumer...长轮询”的主动权还是掌握在Consumer手上,即使Broker有大量的消息积压,也不会主动推送给Consumer。...Consumer的启动、关闭流程 Consumer分为Push和Pull两种模式,对于DefaultMQPullConsumer来说,使用者主动权很高,可以根据实际需要启动、暂停和停止消费过程。

1.9K10

RocketMq之Consumer原理浅析

Consumer是怎么启动的 源码很长,这里就不仔细看了,其实主要就是初始化了三个组件,然后启动后台定时任务 RebalanceImpl 均衡消息队列服务,负责分配当前 Consumer 可消费的消息队列...那么就会有一个问题,比如有2个Consumer,3个MessageQueue,那么这3个MessageQueue怎么分配呢?这就涉及到Consumer的负载均衡了。...首先 Consumer 在启动时,会把自己注册给所有 Broker ,并保持心跳,让每一个 Broker 都知道消费组中有哪些 Consumer 。...然后 Consumer 在消费时,会随机链接一台 Broker ,获取消费组中的所有 Consumer 。 主要流程如下: ? 注意这里会对Consumer集合做一个排序,为什么要这样做呢?...因为每个 consumer 都是在本地负载均衡,所以要排序,否则多个Consumer之间会有冲突。

1.8K10
领券