java自己手动控制kafka的offset操作

Arce

之前使用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)));
    }
  }
  }
}

以上为个人经验，希望能给大家一个参考，也希望大家多多支持CodeAE代码之家。如有错误或未考虑完全的地方，望不吝赐教。
原文链接：https://blog.csdn.net/qq_20641565/article/details/64440425

[Java] java自己手动控制kafka的offset操作

相关帖子

Java面试为何阿里强制要求不在foreach里执行删除操作

图解Java排序算法之希尔排序

图解Java排序算法之快速排序的三数取中法

图解Java排序算法之堆排序

利用Linux防火墙隔离本地欺骗地址的方法详解

用Java8 stream处理数据

电脑nvidia控制面板为什么总是闪退？

笔记本nvidia控制面板闪退解决方法

怎么格式化U盘启动盘？

win7鼠标乱动不受控制解决方法

Arce LV4