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

2024-07-31 22:21:57

之前使用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实现Kafka生产者消费者代码实例

Kafka的结构与RabbitMQ类似,消息生产者向Kafka服务器发送消息,Kafka接收消息后,再投递给消费者. 生产者的消费会被发送到Topic中,Topic中保存着各类数据,每一条数据都使用键.值进行保存. 每一个Topic中都包含一个或多个物理分区(Partition),分区维护着消息的内容和索引,它们有可能被保存在不同服务器. 新建一个Maven项目,pom.xml 加入依赖: <dependency> <groupId>org.apache.kafka</gro
Java kafka如何实现自定义分区类和拦截器

生产者发送到对应的分区有以下几种方式: (1)指定了patition,则直接使用:(可以查阅对应的java api, 有多种参数) (2)未指定patition但指定key,通过对key的value进行hash出一个patition: (3)patition和key都未指定,使用轮询选出一个patition. 但是kafka提供了,自定义分区算法的功能,由业务手动实现分布: 1.实现一个自定义分区类,CustomPartitioner实现Partitioner import org.apache
Java实现Kafka生产者和消费者的示例

Kafka简介 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka的目标是为处理实时数据提供一个统一.高吞吐.低延迟的平台. 方式一:kafka-clients 引入依赖在pom.xml文件中,引入kafka-clients依赖: <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId
Kafka Java Producer代码实例详解

根据业务需要可以使用Kafka提供的Java Producer API进行产生数据,并将产生的数据发送到Kafka对应Topic的对应分区中,入口类为:Producer Kafka的Producer API主要提供下列三个方法: public void send(KeyedMessage<K,V> message) 发送单条数据到Kafka集群 public void send(List<KeyedMessage<K,V>> messages) 发送多条数据(数据集)到
Java将CSV的数据发送到kafka的示例

为什么将CSV的数据发到kafka flink做流式计算时,选用kafka消息作为数据源是常用手段,因此在学习和开发flink过程中,也会将数据集文件中的记录发送到kafka,来模拟不间断数据: 整个流程如下: 您可能会觉得这样做多此一举:flink直接读取CSV不就行了吗?这样做的原因如下: 首先,这是学习和开发时的做法,数据集是CSV文件,而生产环境的实时数据却是kafka数据源: 其次,Java应用中可以加入一些特殊逻辑,例如数据处理,汇总统计(用来和flink结果对比验证): 另外,如果
java自己手动控制kafka的offset操作

之前使用kafka的KafkaStream,让每个消费者和对应的patition建立对应的流来读取kafka上面的数据,如果comsumer得到数据,那么kafka就会自动去维护该comsumer的offset,例如在获取到kafka的消息后正准备入库(未入库),但是消费者挂了,那么如果让kafka自动去维护offset,它就会认为这条数据已经被消费了,那么会造成数据丢失. 但是kafka可以让你自己去手动提交,如果在上面的场景中,那么需要我们手动commit,如果comsumer挂了那么程序
java selenium 常见web UI 元素操作及API使用

本篇介绍我们如何利用selenium 来操作各种页面元素阅读目录链接(link) 输入框 textbox 按钮(Button) 下拉选择框(Select) 单选按钮(Radio Button) 多选框 check box 链接(link) <div> <p>链接 link</p> <a href="www.cnblogs.com/tankxiao">小坦克</a> </div> 链接的操作 // 找到链接元素
Java基于jdbc连接mysql数据库操作示例

本文实例讲述了Java基于jdbc连接mysql数据库操作.分享给大家供大家参考,具体如下: import java.sql.Connection; import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; public class MySQLDemo { private Connection conn = null; pri
Java I/O技术之文件操作详解

在java程序设计中,I/O操作是通过java.io包中的类和接口来实现的,因此,我们第一步要做的就是import这个包. java.io提供了一个File类,这是类很容易让人产生误会,它表示的是一个文件名或者目录名,而不是文件本身,所以通过这个类没法对文件里面的数据进行操作.File类提供了一序列对文件操作的功能:删除文件,创建目录,查询文件大小等等.要想对文件数据进行操作那就需要流对象了,在这里就暂时不做介绍. 下面通过一个叫做FileExtension类来对File类中的各种操作进行封装,
Java对xls文件进行读写操作示例代码

前言本文主要给大家介绍的是关于Java对xls文件进行读写操作的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细的介绍: win7_x64 IDEA Java读写xls文件,使用库jxl.jar 读写xls文件,这里是在知道表格格式的前提下进行操作的. 目前无法操作xlsx文件准备工作将库jxl.jar添加到工程依赖中 Java代码示例示例:从几个文件中读取数据并汇总到一个文件中表格中的数据规定为:首行为标题,以下是数据和名称:例如单位名金额单位1 948.34
Java线程中的notifyAll唤醒操作(推荐)

注意: java中的notifyAll和notify都是唤醒线程的操作,notify只会唤醒等待池中的某一个线程,但是不确定是哪一个线程,notifyAll是针对指定对象里面的所有线程执行唤醒操作,指定对象一旦唤醒成功.则会立即加入线程的资源争夺中去. 例如: package TestThread.ThreadSynchronized; public class TestWaitAll { public static void main(String[] args) { Test1 test1
Java泛型的继承和实现操作示例

本文实例讲述了Java泛型的继承和实现操作.分享给大家供大家参考,具体如下: 一点睛泛型类和泛型接口被定义后,是可以被继承和实现的. 二泛型类的继承 1 代码 class A<E> { E t; } public class B<T,T1>extends A<T>{ public static void main(String[] args){ System.out.println("B test"); } } 2 运行 B test 3 说明
Java数组传递及可变参数操作实例详解

本文实例讲述了Java数组传递及可变参数操作.分享给大家供大家参考,具体如下: 方法可以操作传递和返回基本数据类型,但是方法中也可用来传递和返回数组.如果要向方法中传递一个数组,则方法的接收参数处必须是符合其类型的数组.而且数组属于引用数据类型,所以在把数组传递进方法之后,如果方法对数组本身做了任何修改,修改结果都是会保存下来的. 向方法中传递数组在java中,所有对象都是通过引用进行操作的.而数组也是一种对象,当把数组作为参数传递给方法时,传递的实际上就是数组对象的引用.在方法中对数组的所有
Java接口继承和使用接口操作示例

本文实例讲述了Java接口继承和使用接口操作.分享给大家供大家参考,具体如下: 一接口的继承 1 点睛接口支持多继承,一个接口可以有多个父接口,子接口扩展某个父接口,将会获得父接口里定义的所有抽象方法.常量. 一个接口继承多个父接口时,多个父接口排在extends关键字之后. 2 代码 interface interfaceA { int PROP_A = 5; void testA(); } interface interfaceB { int PROP_B = 6; void testB
Java实现的简单字符串反转操作示例

本文实例讲述了Java实现的简单字符串反转操作.分享给大家供大家参考,具体如下: 最常用的方式是,反向取出每个字符,然后打印: public class StringReverse { public static void main(String[] args) { // 原始字符串 String s = "A quick brown fox jumps over the lazy dog."; System.out.println("原始的字符串:" + s);

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

注意：

相关推荐

随机推荐