重庆分公司
重庆分公司，新征程启航
为企业提供网站建设、域名注册、服务器等服务
Kafka2.7是如何重设消费者组位移
本篇内容介绍了“Kafka2.7是如何重设消费者组位移”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
创新互联公司-专业网站定制、快速模板网站建设、高性价比贡井网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式贡井网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖贡井地区。费用合理售后完善，10多年实体公司更值得信赖。
重设消费
Earliest
首先看看重置位移前的消费进度
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --describe
根据进度截图，能看到所有分区的Lag均为0，说明消息已经被消费完，现在根据Earliest策略重置消费进度，要求重置后所有的消息均可重新消费。
脚本命令方式
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-earliest --execute
此时再度查看消费进度，可以看到 
此时消费者可以重新消费这些消息。
脚本命令方式(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-earliest --execute Java API方式
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
final String topic = "mytopic";
try (KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
Collection partitions = consumer.partitionsFor(topic).stream()
.map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
.collect(Collectors.toList());
consumer.seekToBeginning(partitions);
consumer.partitionsFor(topic).forEach(i -> consumer.position(new TopicPartition(topic, i.partition())));
}
需要特殊说明的是，seekToBeginning、seekToEnd等方法执行完需要执行position才会立刻生效
Java API方式(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
final String topic = "mytopic";
try (KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
List partitions = new ArrayList();
partitions.add(new TopicPartition(topic, 1));
partitions.add(new TopicPartition(topic, 2));
consumer.seekToBeginning(partitions);
consumer.position(new TopicPartition(topic, 1));
consumer.position(new TopicPartition(topic, 2));
} Latest
首先看看重置位移前的消费进度。
根据上图可以看到，kafka当前没有任何消息被消费，现在根据Latest策略重置消费进度，要求重置后原消息不再消费。
脚本命令
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-latest --execute
重置后
脚本命令(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-latest --execute Java API
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
consumer.seekToEnd(consumer.partitionsFor(topic).stream()
.map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
.collect(Collectors.toList()));
consumer.partitionsFor(topic).forEach(i -> consumer.position(new TopicPartition(topic, i.partition())));
} Java API(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
List partitions = new ArrayList();
partitions.add(new TopicPartition(topic, 1));
partitions.add(new TopicPartition(topic, 2));
consumer.seekToEnd(partitions);
consumer.position(new TopicPartition(topic, 1));
consumer.position(new TopicPartition(topic, 2));
} Current
此方法暂时联想不到相应的应用场景，粗略跳过，待以后了解后再补充。
脚本命令
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-current --execute
脚本命令(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-current --execute Java API
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
consumer.partitionsFor(topic).stream().map(info -> new TopicPartition(topic, info.partition())).forEach(tp -> {
long committedOffset = consumer.committed(tp).offset();
consumer.seek(tp, committedOffset);
});
} Java API(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
TopicPartition tp1 = new TopicPartition(topic, 1);
TopicPartition tp2 = new TopicPartition(topic, 2);
consumer.seek(tp1, consumer.committed(tp1).offset());
consumer.seek(tp2, consumer.committed(tp2).offset());
} Specified-Offset
重置前
脚本命令
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-offset 5 --execute
脚本命令(指定分区)
通常来说，各个分区的提交位移往往是不同的，所以将所有分区的位移设置成同一个值并不显示，需要指定分区。
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --to-offset 11 --execute Java API
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
consumer.partitionsFor(topic).stream().forEach(pi -> {
TopicPartition tp = new TopicPartition(topic, pi.partition());
consumer.seek(tp, 5L);
});
} Java API(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
consumer.seek(new TopicPartition(topic, 2), 10L);
} Shift-By-N
重置前
脚本命令
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --shift-by -1 --execute
脚本命令(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --shift-by -2 --execute Java API
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
for (PartitionInfo info : consumer.partitionsFor(topic)) {
TopicPartition tp = new TopicPartition(topic, info.partition());
consumer.seek(tp, consumer.committed(tp).offset() - 1L);
}
} Java API(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
TopicPartition tp = new TopicPartition(topic, 2);
consumer.seek(tp, consumer.committed(tp).offset() + 2L);
}
DateTime
有时按照时间点来重置位移是个不错的方式，重置前:
脚本命令
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-datetime 2021-05-09T00:00:00.000 --execute
脚本命令(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --to-datetime 2020-05-09T00:00:00.000 --executeJava API
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
long ts = new Date().getTime() - 24 * 60 * 60 * 1000;
Map timeToSearch = consumer.partitionsFor(topic).stream()
.map(pi -> new TopicPartition(topic, pi.partition()))
.collect(Collectors.toMap(Function.identity(), tp -> ts));
for (Entry entry : consumer.offsetsForTimes(timeToSearch).entrySet()) {
consumer.seek(entry.getKey(), entry.getValue() == null ? consumer.committed(entry.getKey()).offset() : entry.getValue().offset());
}
} Java API(指定分区)
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
final String topic = "mytopic";
try (final KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList(topic));
consumer.poll(0);
long ts = new Date().getTime() - 365 * 24 * 60 * 60 * 1000;
Map timeToSearch = new HashMap(){{
put(new TopicPartition(topic, 2), ts);
}};
for (Entry entry : consumer.offsetsForTimes(timeToSearch).entrySet()) {
consumer.seek(entry.getKey(), entry.getValue() == null ? consumer.committed(entry.getKey()).offset() : entry.getValue().offset());
}
} Duration
重置前
脚本命令
首先需要了解Java Duration的格式PnDTnHnMnS，这里不做详细展开。
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --by-duration P1DT0H0M0S --execute
脚本命令(指定分区)
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --by-duration P1DT0H0M0S --execute
Java API方式
同DateTime
“Kafka2.7是如何重设消费者组位移”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！
当前文章：Kafka2.7是如何重设消费者组位移
浏览路径：http://cqcxhl.cn/article/johppd.html