java自己手动控制kafka的offset操作
之前使用kafka的KafkaStream,让每个消费者和对应的patition建立对应的流来读取kafka上面的数据,如果comsumer得到数据,那么kafka就会自动去维护该comsumer的offset,例如在获取到kafka的消息后正准备入库(未入库),但是消费者挂了,那么如果让kafka自动去维护offset,它就会认为这条数据已经被消费了,那么会造成数据丢失。
但是kafka可以让你自己去手动提交,如果在上面的场景中,那么需要我们手动commit,如果comsumer挂了 那么程序就不会执行commit这样的话 其他同group的消费者又可以消费这条数据,保证数据不丢,先要做如下设置:
//设置不自动提交,自己手动更新offset properties.put("enable.auto.commit", "false");
使用如下api提交:
consumer.commitSync();
注意:
刚做了个测试,如果我从kafka中取出5条数据,分别为1,2,3,4,5,如果消费者在执行一些逻辑在执行1,2,3,4的时候都失败了未提交commit,然后消费5做逻辑成功了提交了commit,那么offset也会被移动到5那一条数据那里,1,2,3,4 相当于也会丢失
如果是做消费者取出数据执行一些操作,全部都失败的话,然后重启消费者,这些数据会从失败的时候重新开始读取
所以消费者还是应该自己做容错机制
测试项目结构如下:
其中ConsumerThreadNew类:
package com.lijie.kafka; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * * * @Filename ConsumerThreadNew.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email lijiewj39069@touna.cn * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */ public class ConsumerThreadNew implements Runnable { private static Logger LOG = LoggerFactory.getLogger(ConsumerThreadNew.class); //KafkaConsumer kafka生产者 private KafkaConsumer<String, String> consumer; //消费者名字 private String name; //消费的topic组 private List<String> topics; //构造函数 public ConsumerThreadNew(KafkaConsumer<String, String> consumer, String topic, String name) { super(); this.consumer = consumer; this.name = name; this.topics = Arrays.asList(topic); } @Override public void run() { consumer.subscribe(topics); List<ConsumerRecord<String, String>> buffer = new ArrayList<>(); // 批量提交数量 final int minBatchSize = 1; while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { LOG.info("消费者的名字为:" + name + ",消费的消息为:" + record.value()); buffer.add(record); } if (buffer.size() >= minBatchSize) { //这里就是处理成功了然后自己手动提交 consumer.commitSync(); LOG.info("提交完毕"); buffer.clear(); } } } }
MyConsume类如下:
package com.lijie.kafka; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * * * @Filename MyConsume.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email lijiewj39069@touna.cn * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */ public class MyConsume { private static Logger LOG = LoggerFactory.getLogger(MyConsume.class); public MyConsume() { // TODO Auto-generated constructor stub } public static void main(String[] args) { Properties properties = new Properties(); properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093"); //设置不自动提交,自己手动更新offset properties.put("enable.auto.commit", "false"); properties.put("auto.offset.reset", "latest"); properties.put("zookeeper.connect", "10.0.4.141:2181,10.0.4.142:2181,10.0.4.143:2181"); properties.put("session.timeout.ms", "30000"); properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("group.id", "lijieGroup"); properties.put("zookeeper.connect", "192.168.80.123:2181"); properties.put("auto.commit.interval.ms", "1000"); ExecutorService executor = Executors.newFixedThreadPool(5); //执行消费 for (int i = 0; i < 7; i++) { executor.execute(new ConsumerThreadNew(new KafkaConsumer<String, String>(properties), "lijietest", "消费者" + (i + 1))); } } }
MyProducer类如下:
package com.lijie.kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; /** * * * @Filename MyProducer.java * * @Description * * @Version 1.0 * * @Author Lijie * * @Email lijiewj39069@touna.cn * * @History *<li>Author: Lijie</li> *<li>Date: 2017年3月21日</li> *<li>Version: 1.0</li> *<li>Content: create</li> * */ public class MyProducer { private static Properties properties; private static KafkaProducer<String, String> pro; static { //配置 properties = new Properties(); properties.put("bootstrap.servers", "10.0.4.141:19093,10.0.4.142:19093,10.0.4.143:19093"); //序列化类型 properties .put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //创建生产者 pro = new KafkaProducer<>(properties); } public static void main(String[] args) throws Exception { produce("lijietest"); } public static void produce(String topic) throws Exception { //模拟message // String value = UUID.randomUUID().toString(); for (int i = 0; i < 10000; i++) { //封装message ProducerRecord<String, String> pr = new ProducerRecord<String, String>(topic, i + ""); //发送消息 pro.send(pr); Thread.sleep(1000); } } }
pom文件如下:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>lijie-kafka-offset</groupId> <artifactId>lijie-kafka-offset</artifactId> <version>0.0.1-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.10.1.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>1.0.3</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>1.0.3</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>jdk.tools</groupId> <artifactId>jdk.tools</artifactId> <version>1.7</version> <scope>system</scope> <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.3.6</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>1.7</source> <target>1.7</target> </configuration> </plugin> </plugins> </build> </project>
补充:kafka javaAPI 手动维护偏移量
我就废话不多说了,大家还是直接看代码吧~
package com.kafka; import kafka.javaapi.PartitionMetadata; import kafka.javaapi.consumer.SimpleConsumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.junit.Test; import java.util.*; public class ConsumerManageOffet { //broker的地址, //与老版的kafka的区别是,新版本的kafka把偏移量保存到了broker,而老版本的是把偏移量保存到了zookeeper中 //所以在读取数据时,应当设置broker的地址 private static String ips = "192.168.136.150:9092,192.168.136.151:9092,192.168.136.152:9092"; public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers",ips); props.put("group.id","test02"); props.put("auto.offset.reset","earliest"); props.put("max.poll.records","10"); props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); System.out.println("---------------------"); while(true){ ConsumerRecords<String,String> records = consumer.poll(10); System.out.println("+++++++++++++++++++++++"); for(ConsumerRecord<String,String> record: records){ System.out.println("---"); System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(), record.key(),record.value()); } } } //手动维护偏移量 @Test public void autoManageOffset2(){ Properties props = new Properties(); //broker的地址 props.put("bootstrap.servers",ips); //这是消费者组 props.put("group.id","groupPP"); //设置消费的偏移量,如果以前消费过则接着消费,如果没有就从头开始消费 props.put("auto.offset.reset","earliest"); //设置自动提交偏移量为false props.put("enable.auto.commit","false"); //设置Key和value的序列化 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); //new一个消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); //指定消费的topic consumer.subscribe(Arrays.asList("my-topic")); while(true){ ConsumerRecords<String, String> records = consumer.poll(1000); //通过records获取这个集合中的数据属于那几个partition Set<TopicPartition> partitions = records.partitions(); for(TopicPartition tp : partitions){ //通过具体的partition把该partition中的数据拿出来消费 List<ConsumerRecord<String, String>> partitionRecords = records.records(tp); for(ConsumerRecord r : partitionRecords){ System.out.println(r.offset() +" "+r.key()+" "+r.value()); } //获取新这个partition中的最后一条记录的offset并加1 那么这个位置就是下一次要提交的offset long newOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1; consumer.commitSync(Collections.singletonMap(tp,new OffsetAndMetadata(newOffset))); } } } }
以上为个人经验,希望能给大家一个参考,也希望大家多多支持猪先飞。如有错误或未考虑完全的地方,望不吝赐教。
相关文章
- 这篇文章主要介绍了如何利用java语言实现经典《复杂迷宫》游戏,文中采用了swing技术进行了界面化处理,感兴趣的小伙伴可以动手试一试...2022-02-01
java 运行报错has been compiled by a more recent version of the Java Runtime
java 运行报错has been compiled by a more recent version of the Java Runtime (class file version 54.0)...2021-04-01- 这篇文章主要介绍了在java中获取List集合中最大的日期时间操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-15
- 这篇文章主要介绍了教你怎么用Java获取国家法定节假日,文中有非常详细的代码示例,对正在学习java的小伙伴们有非常好的帮助,需要的朋友可以参考下...2021-04-23
- 这篇文章主要介绍了Java如何发起http请求的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-03-31
- 这篇文章主要介绍了.NET Core下使用Kafka的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22
- 说起C#和Java这两门语言(语法,数据类型 等),个人以为,大概有90%以上的相似,甚至可以认为几乎一样。但是在工作中,我也发现了一些细微的差别...2020-06-25
- 这篇文章主要介绍了解决Java处理HTTP请求超时的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-29
- 这篇文章主要介绍了java 判断两个时间段是否重叠的案例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-15
java 画pdf用itext调整表格宽度、自定义各个列宽的方法
这篇文章主要介绍了java 画pdf用itext调整表格宽度、自定义各个列宽的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-01-31- 这篇文章主要介绍了超简洁java实现双色球若干注随机号码生成(实例代码),本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-04-02
- 这篇文章主要介绍了Java生成随机姓名、性别和年龄的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-10-01
- 这篇文章主要介绍了java正则表达式判断前端参数修改表中另一个字段的值,需要的朋友可以参考下...2021-05-07
Java使用ScriptEngine动态执行代码(附Java几种动态执行代码比较)
这篇文章主要介绍了Java使用ScriptEngine动态执行代码,并且分享Java几种动态执行代码比较,需要的朋友可以参考下...2021-04-15- 这篇文章主要介绍了Java开发实现人机猜拳游戏,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2020-08-03
- 这篇文章主要介绍了Java List集合返回值去掉中括号('[ ]')的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-29
Java 8 Stream 的终极技巧——Collectors 功能与操作方法详解
这篇文章主要介绍了Java 8 Stream Collectors 功能与操作方法,结合实例形式详细分析了Java 8 Stream Collectors 功能、操作方法及相关注意事项,需要的朋友可以参考下...2020-05-20Java中lombok的@Builder注解的解析与简单使用详解
这篇文章主要介绍了Java中lombok的@Builder注解的解析与简单使用,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-01-06- 下面小编就为大家带来一篇java中String类型变量的赋值问题介绍。小编觉得挺不错的。现在分享给大家,给大家一个参考。...2016-03-28
- 这篇文章主要介绍了Java线程池中的各个参数如何合理设置操作,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-06-19