0


Kafka系列之消息重新消费

概述

需求来源,在review前人留下的屎山代码时发现如下截图所示的代码片段:
在这里插入图片描述
也就是说代码是空实现的。另外,从类名定义也知道需求未实现。

于是有此需求:已经消费过的消息重新消费。

调研

调研下来,主要有以下3种可能性方案

实现方案

  1. 修改偏移量,即offset,可通过脚本快速实现
  2. 新增Group,这将使 Kafka 认为您正在使用一个新的消费者组,并从起始偏移量开始重新消费消息。需通过代码实现
  3. 消息产生者重新发送消息,实际业务中不太常见,也不现实

Kafka的偏移量的保存方式,根据不同版本号有3种方式:保存在zookeeper中、保存在kafka的自带

_consumer_offset

这个topic中、保存在自定义的存储系统中。

版本

首先需要知道Kafka什么版本,找到Kafka的安装目录。可以通过find命令:

find / -name '*kafka*'


根据命令输出得知安装目录为:

/usr/local/kafka

,进入到

libs

目录,发现很多

kafka_2.11-2.2.0.*

文件。其中,2.11为scala版本,2.2.0为kafka版本。

脚本

使用Kafka自带的

bin

目录下的

kafka-consumer-groups.sh

脚本设置消费者组(consumer group)的位移, 这是0.11.0.0版本提供的新功能且只适用于新版本consumer。在此版本之前,如果要为已有的consumer group调整位移必须要手动编写Java程序调用

KafkaConsumer.seek()

方法。

使用此脚本可修改consumer group的位移,有个前提:consumer group必须是inactive的,即不能是处于正在工作中的状态。

格式:

./kafka-consumer-groups.sh --bootstrap-server ip:9092 --group <> --topic <topic> --reset-offsets <offset_option>

如:

./kafka-consumer-groups.sh --bootstrap-server 172.100.200.200:9092 --group collect_data_business_service --topic topic_event:0 --reset-offsets --to-offset 195725 --execute

topic:指定topic的作用域

  • --all-topics:为consumer group下所有topic的所有分区调整位移
  • --topic t1 --topic t2:为指定的若干个topic的所有分区调整位移
  • --topic t1:0,1,2:为指定的topic的某个分区调整位移

reset-offsets:确定位移重设策略

  • --to-earliest:把位移调整到分区当前最小位移
  • --to-latest:把位移调整到分区当前最新位移
  • --to-current:把位移调整到分区当前位移
  • --to-offset <offset>:把位移调整到指定位移处
  • --shift-by N:把位移调整到当前位移 + N处,N为负数表示向前移动
  • --to-datetime <datetime>:把位移调整到大于给定时间的最早位移处,datetime格式yyyy-MM-ddTHH:mm:ss.xxx
  • --by-duration <duration>:把位移调整到距离当前时间指定间隔的位移处,duration格式是PnDTnHnMnS,如PT0H5M0S
  • --from-file <file>:从CSV文件中读取调整策略

常见报错:

  • Error: Executing consumer group command failed due to org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. 原因:命令指定的IP有误,或端口不对,或未开放等原因导致连接超时
  • Error: Assignments can only be reset if the group ‘collect_data_business_service’ is inactive, but the current state is Stable. 原因:在修改offset时,需要停止消费者应用程序,如杀掉Java进程,停止container容器等
  • WARN New offset (0) is lower than earliest offset for topic partition topic_event-0. Value will be set to 195551 (kafka.admin.ConsumerGroupCommand$) 原因:

代码

也可以通过代码的方式,指定一个新的消费者组,即

group.id

@OverridepublicvoidafterPropertiesSet()throwsException{Properties p =newProperties();
    p.setProperty("bootstrap.servers","172.100.200.200:9092");
    p.setProperty("group.id","collect_data_business_service");
    p.setProperty("auto.offset.reset","earliest");
    p.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
    p.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String,String> consumer =newKafkaConsumer<>(p);
    consumer.subscribe(Collections.singleton("topic_event"));
    consumer.poll(Duration.ofSeconds(1));
    consumer.seekToBeginning(consumer.assignment());while(true){ConsumerRecords<String,String> message = consumer.poll(Duration.ofSeconds(1));
        log.info("topic_event---collect_data_business_service--消费消息:"+ message);CollectDataV2Content content =JsonUtil.jsonToBean(message.toString(),CollectDataV2Content.class);
        service.saveContent(content);}}

配置:

  • auto.offset.reset:只能配置以下3种情况:latest, earliest, none,Spring-Kafka应用启动时会检查此项配置,不正确的配置会报错,应用启动失败
  • key.deserializer:必须配置,否则报错:Caused by: org.apache.kafka.common.config.ConfigException: Missing required configuration "key.deserializer" which has no default value.,应用启动失败。
  • value.deserializer:同上

参考

标签: kafka

本文转载自: https://blog.csdn.net/lonelymanontheway/article/details/129902751
版权归原作者 johnny233 所有, 如有侵权,请联系我们删除。

“Kafka系列之消息重新消费”的评论:

还没有评论