快捷導(dǎo)航

java自己手動(dòng)控制kafka的offset操作

更新時(shí)間：2021年02月20日 10:48:14 作者：lijie_cq

這篇文章主要介紹了java自己手動(dòng)控制kafka的offset操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧

之前使用kafka的KafkaStream，讓每個(gè)消費(fèi)者和對(duì)應(yīng)的patition建立對(duì)應(yīng)的流來(lái)讀取kafka上面的數(shù)據(jù)，如果comsumer得到數(shù)據(jù)，那么kafka就會(huì)自動(dòng)去維護(hù)該comsumer的offset，例如在獲取到kafka的消息后正準(zhǔn)備入庫(kù)（未入庫(kù)），但是消費(fèi)者掛了，那么如果讓kafka自動(dòng)去維護(hù)offset，它就會(huì)認(rèn)為這條數(shù)據(jù)已經(jīng)被消費(fèi)了，那么會(huì)造成數(shù)據(jù)丟失。

但是kafka可以讓你自己去手動(dòng)提交，如果在上面的場(chǎng)景中，那么需要我們手動(dòng)commit，如果comsumer掛了那么程序就不會(huì)執(zhí)行commit這樣的話其他同group的消費(fèi)者又可以消費(fèi)這條數(shù)據(jù)，保證數(shù)據(jù)不丟，先要做如下設(shè)置：

//設(shè)置不自動(dòng)提交，自己手動(dòng)更新offset
properties.put("enable.auto.commit", "false");

使用如下api提交：

consumer.commitSync();

注意：

剛做了個(gè)測(cè)試，如果我從kafka中取出5條數(shù)據(jù)，分別為1,2,3,4,5，如果消費(fèi)者在執(zhí)行一些邏輯在執(zhí)行1,2,3，4的時(shí)候都失敗了未提交commit，然后消費(fèi)5做邏輯成功了提交了commit，那么offset也會(huì)被移動(dòng)到5那一條數(shù)據(jù)那里，1,2,3,4 相當(dāng)于也會(huì)丟失

如果是做消費(fèi)者取出數(shù)據(jù)執(zhí)行一些操作，全部都失敗的話，然后重啟消費(fèi)者，這些數(shù)據(jù)會(huì)從失敗的時(shí)候重新開(kāi)始讀取

所以消費(fèi)者還是應(yīng)該自己做容錯(cuò)機(jī)制

測(cè)試項(xiàng)目結(jié)構(gòu)如下:

其中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生產(chǎn)者
  private KafkaConsumer<String, String>  consumer;
  //消費(fèi)者名字
  private String             name;
  //消費(fèi)的topic組
  private List<String>          topics;
  //構(gòu)造函數(shù)
  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<>();
    // 批量提交數(shù)量
    final int minBatchSize = 1; 
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        LOG.info("消費(fèi)者的名字為:" + name + ",消費(fèi)的消息為：" + record.value());
        buffer.add(record);
      }
      if (buffer.size() >= minBatchSize) {
        //這里就是處理成功了然后自己手動(dòng)提交
        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");
    //設(shè)置不自動(dòng)提交，自己手動(dòng)更新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);
    //執(zhí)行消費(fèi)
    for (int i = 0; i < 7; i++) {
      executor.execute(new ConsumerThreadNew(new KafkaConsumer<String, String>(properties),
        "lijietest", "消費(fèi)者" + (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");
    //創(chuàng)建生產(chǎn)者
    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 + "");
      //發(fā)送消息
      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>

補(bǔ)充：kafka javaAPI 手動(dòng)維護(hù)偏移量

我就廢話不多說(shuō)了，大家還是直接看代碼吧~

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的區(qū)別是，新版本的kafka把偏移量保存到了broker，而老版本的是把偏移量保存到了zookeeper中
//所以在讀取數(shù)據(jù)時(shí)，應(yīng)當(dāng)設(shè)置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());
      }
    }
  }
  //手動(dòng)維護(hù)偏移量
  @Test
  public void autoManageOffset2(){
    Properties props = new Properties();
    //broker的地址
    props.put("bootstrap.servers",ips);
    //這是消費(fèi)者組
    props.put("group.id","groupPP");
    //設(shè)置消費(fèi)的偏移量，如果以前消費(fèi)過(guò)則接著消費(fèi)，如果沒(méi)有就從頭開(kāi)始消費(fèi)
    props.put("auto.offset.reset","earliest");
    //設(shè)置自動(dòng)提交偏移量為false
    props.put("enable.auto.commit","false");
    //設(shè)置Key和value的序列化
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    //new一個(gè)消費(fèi)者
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    //指定消費(fèi)的topic
    consumer.subscribe(Arrays.asList("my-topic"));
    while(true){
      ConsumerRecords<String, String> records = consumer.poll(1000);
      //通過(guò)records獲取這個(gè)集合中的數(shù)據(jù)屬于那幾個(gè)partition
      Set<TopicPartition> partitions = records.partitions();
      for(TopicPartition tp : partitions){
        //通過(guò)具體的partition把該partition中的數(shù)據(jù)拿出來(lái)消費(fèi)
        List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
        for(ConsumerRecord r : partitionRecords){
          System.out.println(r.offset()  +"   "+r.key()+"   "+r.value());
        }
        //獲取新這個(gè)partition中的最后一條記錄的offset并加1 那么這個(gè)位置就是下一次要提交的offset
        long newOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1;
        consumer.commitSync(Collections.singletonMap(tp,new OffsetAndMetadata(newOffset)));
      }
    }
  }
}

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教。

您可能感興趣的文章: