Kafka生产者与消费者api示例

这篇具有很好参考价值的文章主要介绍了Kafka生产者与消费者api示例。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

生产者api示例

 文章来源地址https://www.toymoban.com/news/detail-471093.html

一个正常的生产逻辑需要具备以下几个步骤

  1. 配置生产者参数及创建相应的生产者实例

  2. 构建待发送的消息

  3. 发送消息

  4. 关闭生产者实例

采用默认分区方式将消息散列的发送到各个分区当中

 

package com.doitedu;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 1.构建一个kafka的客户端
         * 2.创建一些待发送的消息,构建成kafka所需要的格式
         * 3.调用kafka的api去发送消息
         * 4.关闭kafka生产实例
         */
        //1.创建kafka的对象,配置一些kafka的配置文件
        //它里面有一个泛型k,v
        //要发送数据的key
        //要发送的数据value
        //他有一个隐含之意,就是kafka发送的消息,是一个key,value类型的数据,但是不是必须得,其实只需要发送value的值就可以了
        Properties pros = new Properties();
        //指定kafka集群的地址
        pros.setProperty("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092");
        //指定key的序列化方式
        pros.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //指定value的序列化方式
        pros.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //ack模式,取值有0,1,-1(all),all是最慢但最安全的  服务器应答生产者成功的策略
        pros.put("acks", "all");
        //这是kafka发送数据失败的重试次数,这个可能会造成发送数据的乱序问题
        pros.setProperty("retries", "3");
        //数据发送批次的大小 单位是字节
        pros.setProperty("batch.size", "10000");
        //一次数据发送请求所能发送的最大数据量
        pros.setProperty("max.request.size", "102400");
        //消息在缓冲区保留的时间,超过设置的值就会被提交到服务端
        pros.put("linger.ms", 10000);
        //整个Producer用到总内存的大小,如果缓冲区满了会提交数据到服务端
        //buffer.memory要大于batch.size,否则会报申请内存不足的错误
        pros.put("buffer.memory", 10240);

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(pros);
        for (int i = 0; i < 1000; i++) {
            //key value  0 --> doit32+-->+0
            //key value  1 --> doit32+-->+1
            //key value  2 --> doit32+-->+2
            //2.创建一些待发送的消息,构建成kafka所需要的格式
            ProducerRecord<String, String> record = new ProducerRecord<>("test01", i + "", "doit32-->" + i);
            //3.调用kafka的api去发送消息
            kafkaProducer.send(record);
            Thread.sleep(100);
        }
        kafkaProducer.flush();
        kafkaProducer.close();
    }
}

 对于properties配置的第二种写法,相对来说不会出错,简单举例:

public static void main(String[] args) {
    Properties pros = new Properties();
    pros.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092,linux02:9092,linux03:9092");
    pros.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    pros.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
}

 

1.kafka的生产者可以持续不断的向topic中发送数据不?

可以

2.kafak的生产者有哪些必须配置的参数:

//指定kafka集群的地址 pros.setProperty("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092"); //指定key的序列化方式 pros.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); //指定value的序列化方式 pros.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

3.kafka生产者发送数据的时候,可以用jdk的序列化器来将数据进行序列化不?

不可以,kafka有指定的序列化接口 org.apache.kafka.common.serialization.Serializer

4.构造一个kafka的生产者后,是不是就已经确定了,我的数据是需要发送到哪一个topic里面不?

不是哦,咱们构造生产者对象的时候不需要指定topic,是在构造发送数据对象的时候才指定的

 消费者Api示例

一个正常的消费逻辑需要具备以下几个步骤:

  1. 配置消费者客户端参数及创建相应的消费者实例;

  2. 订阅主题topic;

  3. 拉取消息并消费;

  4. 定期向__consumer_offsets主题提交消费位移offset;

  5. 关闭消费者实例

 

package com.doitedu;

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Optional;
import java.util.Properties;

public class ConsumerDemo {
    public static void main(String[] args) {
        //1.创建kafka的消费者对象,附带着把配置文件搞定
        Properties props = new Properties();
        //props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        //props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 定义kakfa 服务的地址,不需要将所有broker指定上
       // props.put("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092");
        // 制定consumer group
        props.put("group.id", "g3");
        // 是否自动提交offset  __consumer_offset   有多少分区  50 
        props.put("enable.auto.commit", "true");
        // 自动提交offset的时间间隔   -- 这玩意设置的大小怎么控制
        props.put("auto.commit.interval.ms", "5000");  //50000   1000
        // key的反序列化类
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // value的反序列化类
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 如果没有消费偏移量记录,则自动重设为起始offset:latest, earliest, none
        props.put("auto.offset.reset","earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        //2.订阅主题(确定需要消费哪一个或者多个主题)
        consumer.subscribe(Arrays.asList("test02"));
        //3.开始从topic中获取数据
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            for (ConsumerRecord<String, String> record : records) {
                //这是数据所属的哪一个topic
                String topic = record.topic();
                //该条数据的偏移量
                long offset = record.offset();
                //这条数据是哪一个分区的
                int partition = record.partition();
                //这条数据记录的时间戳,但是这个时间戳有两个类型
                long timestamp = record.timestamp();
                //上面时间戳的类型,这个类型有两个,一个是CreateTime(这条数据创建的时间), LogAppendTime(这条数据往日志里面追加的时间)
                TimestampType timestampType = record.timestampType();
                //这个数据key的值
                String key = record.key();
                //这条数据value的值
                String value = record.value();
                //分区leader的纪元
                Optional<Integer> integer = record.leaderEpoch();
                //key的长度
                int keySize = record.serializedKeySize();
                //value的长度
                int valueSize = record.serializedValueSize();
                //数据的头部信息
                Headers headers = record.headers();
//            for (Header header : headers) {
//                String hKey = header.key();
//                byte[] hValue = header.value();
//                String valueString = new String(hValue);
//                System.out.println("header的key值 = " + hKey + "header的value的值 = "+ valueString);
//            }
                System.out.printf("topic = %s ,offset = %d, partition = %d, timestampType = %s ,timestamp = %d , key = %s , value = %s ,leader的纪元 = %d , key序列化的长度 = %d ,value 序列化的长度 = %d \r\n" ,
                        topic,offset,partition,timestampType + "",timestamp,key,value,integer.get(),keySize,valueSize);
            }
        }

        //4.关闭消费者对象
//        consumer.close();
    }
}

 subscribe订阅主题

 subscribe有如下重载方法:

public void subscribe(Collection<String> topics,ConsumerRebalanceListener listener) 
public void subscribe(Collection<String> topics) 
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) 
public void subscribe(Pattern pattern)
  1. 指定集合方式订阅主题

 consumer.subscribe(Arrays.asList(topicl ));

2.正则方式订阅主题

如果消费者采用的是正则表达式的方式(subscribe(Pattern))订阅, 在之后的过程中,如果有人又创建了新的主题,并且主题名字与正表达式相匹配,那么这个消费者就可以消费到新添加的主题中的消息。如果应用程序需要消费多个主题,并且可以处理不同的类型,那么这种订阅方式就很有效。

正则表达式的方式订阅的示例

 consumer.subscribe(Pattern.compile ("topic.*" ));

 利用正则表达式订阅主题,可实现动态订阅

 assign订阅主题

消费者不仅可以通过 KafkaConsumer.subscribe() 方法订阅主题,还可直接订阅某些主题的指定分区;

在 KafkaConsumer 中提供了 assign() 方法来实现这些功能,此方法的具体定义如下:

 

package com.doitedu;

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
import java.util.Properties;

public class ConsumerDemo1 {
    public static void main(String[] args) {
        //1.创建kafka的消费者对象,附带着把配置文件搞定
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"doit01");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        //2.订阅主题(确定需要消费哪一个或者多个主题)
//        consumer.subscribe(Arrays.asList("test03"));

//        consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
//        //我现在想手动指定,我需要从哪边开始消费
//        //如果用subscribe去订阅主题的时候,他内部会给这个消费者组来一个自动再均衡
//        consumer.seek(new TopicPartition("test03",0),2);
        TopicPartition tp01 = new TopicPartition("test03", 0);

        //他就是手动去订阅主题和partition,有了这个就不需要再去订阅subscribe主题了,手动指定以后,他的内部就不会再来自动均衡了
        consumer.assign(Arrays.asList(tp01)); // 手动订阅指定主题的指定分区的指定位置
        consumer.seek(new TopicPartition("test03",0),2);

        //3.开始从topic中获取数据
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            for (ConsumerRecord<String, String> record : records) {
                //这是数据所属的哪一个topic
                String topic = record.topic();
                //该条数据的偏移量
                long offset = record.offset();
                //这条数据是哪一个分区的
                int partition = record.partition();
                //这条数据记录的时间戳,但是这个时间戳有两个类型
                long timestamp = record.timestamp();
                //上面时间戳的类型,这个类型有两个,一个是CreateTime(这条数据创建的时间), LogAppendTime(这条数据往日志里面追加的时间)
                TimestampType timestampType = record.timestampType();
                //这个数据key的值
                String key = record.key();
                //这条数据value的值
                String value = record.value();

                //分区leader的纪元
                Optional<Integer> integer = record.leaderEpoch();
                //key的长度
                int keySize = record.serializedKeySize();
                //value的长度
                int valueSize = record.serializedValueSize();
                //数据的头部信息
                Headers headers = record.headers();
//            for (Header header : headers) {
//                String hKey = header.key();
//                byte[] hValue = header.value();
//                String valueString = new String(hValue);
//                System.out.println("header的key值 = " + hKey + "header的value的值 = "+ valueString);
//            }
                System.out.printf("topic = %s ,offset = %d, partition = %d, timestampType = %s ,timestamp = %d , key = %s , value = %s ,leader的纪元 = %d , key序列化的长度 = %d ,value 序列化的长度 = %d \r\n" ,
                        topic,offset,partition,timestampType + "",timestamp,key,value,integer.get(),keySize,valueSize);
            }
        }

        //4.关闭消费者对象
//        consumer.close();
    }
}

 这个方法只接受参数partitions,用来指定需要订阅的分区集合。示例如下:

consumer.assign(Arrays.asList(new TopicPartition ("tpc_1" , 0),new TopicPartition(“tpc_2”,1))) ;

 subscribe与assign的区别

 

  • 通过subscribe()方法订阅主题具有消费者自动再均衡功能 ;

在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组的消费者增加或减少时,分区分配关系会自动调整,以实现消费负载均衡及故障自动转移。

  • assign() 方法订阅分区时,是不具备消费者自动均衡的功能的;

其实这一点从assign方法参数可以看出端倪,两种类型subscribe()都有ConsumerRebalanceListener类型参数的方法,而assign()方法却没有。

 消息的消费模式

Kafka中的消费是基于拉取模式的。

 消息的消费一般有两种模式:推送模式和拉取模式。推模式是服务端主动将消息推送给消费者,而拉模式是消费者主动向服务端发起请求来拉取消息。

 

public class ConsumerRecord<K, V> {
    public static final long NO_TIMESTAMP = RecordBatch.NO_TIMESTAMP;
    public static final int NULL_SIZE = -1;
    public static final int NULL_CHECKSUM = -1;

    private final String topic;
    private final int partition;
    private final long offset;
    private final long timestamp;
    private final TimestampType timestampType;
    private final int serializedKeySize;
    private final int serializedValueSize;
    private final Headers headers;
    private final K key;
    private final V value;

    private volatile Long checksum;
  • topic partition 这两个属性分别代表消息所属主题的名称和所在分区的编号。

  • offset 表示消息在所属分区的偏移量。

  • timestamp 表示时间戳,与此对应的timestampType 表示时间戳的类型。

  • timestampType 有两种类型 CreateTime 和LogAppendTime ,分别代表消息创建的时间戳和消息追加到日志的时间戳。

  • headers 表示消息的头部内容。

  • key value 分别表示消息的键和消息的值,一般业务应用要读取的就是value ;

  • serializedKeySize、serializedValueSize分别表示key、value 经过序列化之后的大小,如果 key 为空,则 serializedKeySize 值为 -1,同样,如果value为空,则serializedValueSize 的值也会为 -1;

  • checksum 是CRC32的校验值。

示例代码片段 

/**
 * 订阅与消费方式2
 */
TopicPartition tp1 = new TopicPartition("x", 0);
TopicPartition tp2 = new TopicPartition("y", 0);
TopicPartition tp3 = new TopicPartition("z", 0);
List<TopicPartition> tps = Arrays.asList(tp1, tp2, tp3);
consumer.assign(tps);
​
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (TopicPartition tp : tps) {
        List<ConsumerRecord<String, String>> rList = records.records(tp);
        for (ConsumerRecord<String, String> r : rList) {
            r.topic();
            r.partition();
            r.offset();
            r.value();
            //do something to process record.
        }
    }
}

 指定位移消费

有些时候,我们需要一种更细粒度的掌控,可以让我们从特定的位移处开始拉取消息,而 KafkaConsumer 中的seek()方法正好提供了这个功能,让我们可以追前消费或回溯消费。

seek()方法的具体定义如下:

seek都是和assign这个方法一起用 指定消费位置
public void seek(TopicPartiton partition,long offset)

 代码示例:

public class ConsumerDemo3指定偏移量消费 {
    public static void main(String[] args) {

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"g002");
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"doit01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 是否自动提交消费位移
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");

        // 限制一次poll拉取到的数据量的最大值
        props.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,"10240000");
         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // assign方式订阅doit27-1的两个分区
        TopicPartition tp0 = new TopicPartition("doit27-1", 0);
        TopicPartition tp1 = new TopicPartition("doit27-1", 1);
        
        consumer.assign(Arrays.asList(tp0,tp1));
        // 指定分区0,从offset:800开始消费    ;  分区1,从offset:650开始消费
        consumer.seek(tp0,200);
        consumer.seek(tp1,250);

        // 开始拉取消息
        while(true){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(3000));
            for (ConsumerRecord<String, String> rec : poll) {
                System.out.println(rec.partition()+","+rec.key()+","+rec.value()+","+rec.offset());
            }
        }
    }
}

 自动提交消费者偏移量

 

Kafka中默认的消费位移的提交方式是自动提交,这个由消费者客户端参数enable.auto.commit 配置,默认值为 true 。当然这个默认的自动提交不是每消费一条消息就提交一次,而是定期提交,这个定期的周期时间由客户端参数 auto.commit.interval.ms配置,默认值为5秒,此参数生效的前提是 enable. auto.commit 参数为 true。

在默认的方式下,消费者每隔5秒会将拉取到的每个分区中最大的消息位移进行提交。自动位移提交的动作是在 poll() 方法的逻辑里完成的,在每次真正向服务端发起拉取请求之前会检查是否可以进行位移提交,如果可以,那么就会提交上一次轮询的位移。

Kafka 消费的编程逻辑中位移提交是一大难点,自动提交消费位移的方式非常简便,它免去了复杂的位移提交逻辑,让编码更简洁。但随之而来的是重复消费和消息丢失的问题

  • 重复消费

假设刚刚提交完一次消费位移,然后拉取一批消息进行消费,在下一次自动提交消费位移之前,消费者崩溃了,那么又得从上一次位移提交的地方重新开始消费,这样便发生了重复消费的现象(对于再均衡的情况同样适用)。我们可以通过减小位移提交的时间间隔来减小重复消息的窗口大小,但这样并不能避免重复消费的发送,而且也会使位移提交更加频繁。

Kafka生产者与消费者api示例 

 

  • 丢失消息

按照一般思维逻辑而言,自动提交是延时提交,重复消费可以理解,那么消息丢失又是在什么情形下会发生的呢?我们来看下图中的情形:

拉取线程不断地拉取消息并存入本地缓存,比如在BlockingQueue 中,另一个处理线程从缓存中读取消息并进行相应的逻辑处理。设目前进行到了第 y+l 次拉取,以及第m次位移提交的时候,也就是 x+6 之前的位移己经确认提交了,处理线程却还正在处理x+3的消息;此时如果处理线程发生了异常,待其恢复之后会从第m次位移提交处,也就是 x+6 的位置开始拉取消息,那么 x+3至x+6 之间的消息就没有得到相应的处理,这样便发生消息丢失的现象。

 Kafka生产者与消费者api示例

  Kafka生产者与消费者api示例

 手动提交消费者偏移量(调用kafka api)

 

自动位移提交的方式在正常情况下不会发生消息丢失或重复消费的现象,但是在编程的世界里异常无可避免;同时,自动位移提交也无法做到精确的位移管理。在 Kafka中还提供了手动位移提交的方式,这样可以使得开发人员对消费位移的管理控制更加灵活。

很多时候并不是说拉取到消息就算消费完成,而是需要将消息写入数据库、写入本地缓存,或者是更加复杂的业务处理。在这些场景下,所有的业务处理完成才能认为消息被成功消费;

手动的提交方式可以让开发人员根据程序的逻辑在合适的时机进行位移提交。开启手动提交功能的前提是消费者客户端参数 enable.auto.commit 配置为false ,示例如下:

props.put(ConsumerConf.ENABLE_AUTO_COMMIT_CONFIG, false); 

 

手动提交可以细分为同步提交和异步提交,对应于 KafkaConsumer 中的 commitSync()和

commitAsync()两种类型的方法。

  • 同步提交的方式

commitSync()方法的定义如下:

 

/**
 * 手动提交offset
 */
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> r : records) {
        //do something to process record.
    }
    consumer.commitSync();
}

 对于采用 commitSync()的无参方法,它提交消费位移的频率和拉取批次消息、处理批次消息的频率是一样的,如果想寻求更细粒度的、更精准的提交,那么就需要使用commitSync()的另一个有参方法,具体定义如下:

public void commitSync(final Map<TopicPartition,OffsetAndMetadata> offsets)

 示例代码如下:

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> r : records) {
        long offset = r.offset();
        //do something to process record.

        TopicPartition topicPartition = new TopicPartition(r.topic(), r.partition());
        consumer.commitSync(Collections.singletonMap(topicPartition,new OffsetAndMetadata(offset+1)));
    }
}

 提交的偏移量 = 消费完的record的偏移量 + 1

因为,__consumer_offsets中记录的消费偏移量,代表的是,消费者下一次要读取的位置!!!

  • 异步提交方式

异步提交的方式( commitAsync())在执行的时候消费者线程不会被阻塞;可能在提交消费位移的结果还未返回之前就开始了新一次的拉取。异步提交可以让消费者的性能得到一定的增强。 commitAsync方法有一个不同的重载方法,具体定义如下

 Kafka生产者与消费者api示例

 

/**
 * 异步提交offset
 */
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> r : records) {
        long offset = r.offset();

        //do something to process record.
        TopicPartition topicPartition = new TopicPartition(r.topic(), r.partition());
        consumer.commitSync(Collections.singletonMap(topicPartition,new OffsetAndMetadata(offset+1)));
        consumer.commitAsync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(offset + 1)), new OffsetCommitCallback() {
     @Override
     public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                if(e == null ){
                    System.out.println(map);
                }else{
                    System.out.println("error commit offset");
                }
            }
        });
    }
}

 手动提交位移(时机的选择)

 

  • 数据处理完成之前先提交偏移量

可能会发生漏处理的现象(数据丢失)反过来说,这种方式实现了: at most once的数据处理(传递)语义

  • 数据处理完成之后再提交偏移量

可能会发生重复处理的现象(数据重复)反过来说,这种方式实现了: at least once的数据处理(传递)语义当然,数据处理(传递)的理想语义是: exactly once(精确一次)Kafka也能做到exactly once(基于kafka的事务机制)

代码示例:

package com.doitedu;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

public class CommitOffsetByMyself {
    public static void main(String[] args) throws SQLException {

        //获取mysql的连接对象
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/football", "root", "123456");
        connection.setAutoCommit(false);
        PreparedStatement pps = connection.prepareStatement("insert into user values (?,?,?)");
        PreparedStatement pps_offset = connection.prepareStatement("insert into offset values (?,?) on duplicate key update offset = ?");

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //设置手动提交偏移量参数,需要将自动提交给关掉
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //设置从哪里开始消费
//        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        //设置组id
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group001");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //订阅主题
        consumer.subscribe(Arrays.asList("kafka2mysql"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                for (TopicPartition topicPartition : collection) {
                    try {
                        PreparedStatement get_offset = connection.prepareStatement("select offset from offset where topic_partition = ?");
                        String topic = topicPartition.topic();
                        int partition = topicPartition.partition();
                        get_offset.setString(1, topic + "_" + partition);
                        ResultSet resultSet = get_offset.executeQuery();
                        if (resultSet.next()){
                            int offset = resultSet.getInt(1);
                            System.out.println("发生了再均衡,被分配了分区消费权,并且查到了目标分区的偏移量"+partition+" , "+offset);
                            //拿到了offset后就可以定位消费了
                            consumer.seek(new TopicPartition(topic, partition), offset);
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        //拉去数据后写入到mysql
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
                for (ConsumerRecord<String, String> record : records) {
                    String data = record.value();
                    String[] arr = data.split(",");
                    String id = arr[0];
                    String name = arr[1];
                    String age = arr[2];

                    pps.setInt(1, Integer.parseInt(id));
                    pps.setString(2, name);
                    pps.setInt(3, Integer.parseInt(age));
                    pps.execute();

                    //埋个异常,看看是不是真的是这样
//                    if (Integer.parseInt(id) == 5) {
//                        throw new SQLException();
//                    }

                    long offset = record.offset();
                    int partition = record.partition();
                    String topic = record.topic();
                    pps_offset.setString(1, topic + "_" + partition);
                    pps_offset.setInt(2, (int) offset + 1);
                    pps_offset.setInt(3, (int) offset + 1);
                    pps_offset.execute();
                    //提交jdbc事务
                    connection.commit();
                }
            } catch (Exception e) {
                connection.rollback();
            }
        }
    }
}

 消费者提交偏移量方式的总结

consumer的消费位移提交方式:

  • 全自动

    • auto.offset.commit = true

    • 定时提交到consumer_offsets

  • 半自动

    • auto.offset.commit = false;

    • 然后手动触发提交 consumer.commitSync();

    • 提交到consumer_offsets

  • 全手动

    • auto.offset.commit = false;

    • 写自己的代码去把消费位移保存到你自己的地方mysql/zk/redis/

    • 提交到自己所涉及的存储;初始化时也需要自己去从自定义存储中查询到消费位移

 

 

到了这里,关于Kafka生产者与消费者api示例的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 笔记:配置多个kafka生产者和消费者

    如果只有一个kafka,那么使用自带的KafkaAutoConfiguration配置类即可,对应已有属性类KafkaProperties,属性前缀为spring.kafka.xxx; 本文记录配置多个kafka的情况,即在KafkaAutoConfiguration的基础上,自定义额外的kafka生产者和消费者。 适用场景:需要消费来源于不同kafka的消息、需要在不

    2024年02月15日
    浏览(53)
  • kafka生产者和消费者(python版)

    生产者 消费者 消费者中的组名主要用户针对主题的偏移量进行更改,也涉及到主题中分区的问题, kafka工具类 此工具类基本上拿过去就可以用 疑问 当消费者链接kafka时发现topic没有未读的消息怎样退出呢,默认是在一直等待,但是我期望没有要读的消息的时候直接退出即可

    2024年02月16日
    浏览(44)
  • Kafka官方生产者和消费者脚本简单使用

    怎样使用Kafka官方生产者和消费者脚本进行消费生产和消费?这里假设已经下载了kafka官方文件,并已经解压. 这就可以见到测试kafka对应topic了.

    2024年02月04日
    浏览(48)
  • Kafka:主题创建、分区修改查看、生产者、消费者

    1.创建主题 2.查看所有主题 3.查看详细主题 序号从0开始计算 Partition:分区数,该主题有3个分区 Replica:副本数,该主题有3个副本 Leader:副本数中的主的序号,生产消费的对象 1.修改分区数 修改的分区数量不可以小于或者等于当前主题分区的数量,否则会报错 在根目录kaf

    2024年02月11日
    浏览(87)
  • Kafka系列:查看Topic列表、消息消费情况、模拟生产者消费者

    执行topic删除命令时,出现提示 这条命令其实并不执行删除动作,仅仅是在zookeeper上标记该topic要被删除而已,同时也提醒用户一定要提前打开delete.topic.enable开关,否则删除动作是不会执行的。 解决办法: a)在server.properties中设置delete.topic.enable参数为ture b)如下操作: 1.登

    2023年04月26日
    浏览(58)
  • Linux安装Kafka,创建topic、生产者、消费者

    1.创建安装目录/usr/local/kafka mkdir /usr/local/kafka 2.进入安装包目录 cd /usr/local/kafka  3.下载安装包 wget https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz 4.解压安装包 tar -zxvf kafka_2.12-3.3.1.tgz 5.进入cd kafka_2.12-3.3.1目录 cd kafka_2.12-3.3.1/ 6.修改zookeeper配置 cat ./config/zookeeper.properties | grep

    2023年04月17日
    浏览(49)
  • 探究:kafka生产者/消费者与多线程安全

    目录 1. 多线程安全 1.1. 生产者是多线程安全的么? 1.1. 消费者是多线程安全的么? 2. 消费者规避多线程安全方案 2.1. 每个线程维护一个kafkaConsumer 2.2. [单/多]kafkaConsumer实例 + 多worker线程 2.3.方案优缺点对比         Kafka生产者是 线程安全 的,可以在多个线程中共享一个

    2023年04月26日
    浏览(92)
  • 【RabbitMQ实战】 03 SpringBoot RabbitMQ生产者和消费者示例

    上一节我们写了一段原生API来进行生产和消费的例子。实际上SpringBoot对原生RabbitMQ客户端做了二次封装,让我们使用API的代价更低。 依赖引入 RabbitMQ的配置如下 每个配置的具体含义,详见配置 代码说明 使用RabbitTemplate可以发送消息 这个Controller定义了一个发送的接口,调用

    2024年02月07日
    浏览(39)
  • Kafka 架构深度解析:生产者(Producer)和消费者(Consumer)

    Apache Kafka 作为分布式流处理平台,其架构中的生产者和消费者是核心组件,负责实现高效的消息生产和消费。本文将深入剖析 Kafka 架构中生产者和消费者的工作原理、核心概念以及高级功能。 1 发送消息到 Kafka Kafka 生产者负责将消息发布到指定的主题。以下是一个简单的生

    2024年02月03日
    浏览(47)
  • kafka配置大全broker、topic、生产者和消费者等配置介绍

    每个kafka broker中配置文件 server.properties 默认必须配置的属性如下: **bootstrap.servers** - 指定生产者客户端连接kafka集群所需的broker地址列表,格式为host1:port1,host2:port2,可以设置一个或多个。这里并非需要所有的broker地址,因为生产者会从给定的broker里寻找其它的broker。 **key

    2024年02月05日
    浏览(47)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包