APP

海量资源，尽在掌握

搜索
当前位置：首页 > IT技术 > Windows编程 > 正文

Kafka快速入门系列(10) | Kafka的Consumer API操作
2021-09-02 21:13:30

  本篇博主带来的是Kafka的Consumer API操作。

  Consumer消费数据时的可靠性是很容易保证的，因为数据在Kafka中是持久化的，故不用担心数据丢失问题。
  由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。
  所以offset的维护是Consumer消费数据是必须考虑的问题。

1. 手动提交offset
  • 1. 导入依赖
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>

    </build>


    <dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.2</version>
    </dependency>
    </dependencies>
  • 2. 编写代码

需要用到的类：
KafkaConsumer：需要创建一个消费者对象，用来消费数据
ConsumerConfig：获取所需的一系列配置参数
ConsuemrRecord：每条数据都要封装成一个ConsumerRecord对象

package com.buwenbuhuo.kafka.consumer;

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.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author 卜温不火
 * @create 2020-05-06 23:22
 * com.buwenbuhuo.kafka.consumer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop002:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        props.put(ConsumerConfig.GROUP_ID_CONFIG, "bigData-0507");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,3000);

        //1.创建1个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("first"));

        //2.调用poll
     try {

         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records) {
             System.out.println("record = " + record);

             }
         }
     }finally {
         consumer.close();
     }
    }
}
  • 3. 结果呈现

Kafka快速入门系列(10) | Kafka的Consumer API操作_数据

  • 4. 代码分析

  手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次poll的一批数据最高的偏移量提交；不同点是，commitSync会失败重试，一直到提交成功（如果由于不可恢复原因导致，也会提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败。

  • 5. 此为异步提交代码
package com.buwenbuhuo.kafka.consumer;

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.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author 卜温不火
 * @create 2020-05-06 23:22
 * com.buwenbuhuo.kafka.consumer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop002:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        props.put(ConsumerConfig.GROUP_ID_CONFIG, "bigData-0507");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);


        //1.创建1个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("second"));

        //2.调用poll
     try {

         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records) {
             System.out.println("record = " + record);
             }
             // 异步提交
             consumer.commitAsync();
         }
     }finally {
         consumer.close();
     }
    }
}
  • 6. 此为同步提交代码
package com.buwenbuhuo.kafka.consumer;

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.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author 卜温不火
 * @create 2020-05-06 23:22
 * com.buwenbuhuo.kafka.consumer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop002:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        props.put(ConsumerConfig.GROUP_ID_CONFIG, "bigData-0507");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);


        //1.创建1个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("second"));

        //2.调用poll
     try {

         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records) {
             System.out.println("record = " + record);
             }
             // 同步提交
             consumer.commitSync();
         }
     }finally {
         consumer.close();
     }
    }
}
  • 7. 结果图

Kafka快速入门系列(10) | Kafka的Consumer API操作_kafka_02

2. 自动提交offset

  为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。
自动提交offset的相关参数：
enable.auto.commit：是否开启自动提交offset功能
auto.commit.interval.ms：自动提交offset的时间间隔

  • 1. 代码
package com.buwenbuhuo.kafka.consumer;

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.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author 卜温不火
 * @create 2020-05-06 23:22
 * com.buwenbuhuo.kafka.consumer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop002:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        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("second"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}
  • 2. 运行结果
    Kafka快速入门系列(10) | Kafka的Consumer API操作_数据_03
3. 自己维护offset
  • 1. 代码
package com.buwenbuhuo.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
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 java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

/**
 * @author 卜温不火
 * @create 2020-05-07 15:16
 * com.buwenbuhuo.kafka.consumer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomOffsetConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop002:9092");
        props.put("group.id", "test");//消费者组，只要group.id相同，就属于同一个消费者组
        props.put("enable.auto.commit", "false");//自动提交offset
        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("second"), new ConsumerRebalanceListener() {

            //提交当前负责的分区的offset
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("==========回收的分区===========");
                for (TopicPartition partition : partitions){
                    System.out.println("partition = " + partition);   // 啥也没干不写东西也可以
                }

            }

            //定位新分配的分区的offset
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("========重新得到的分区===========");
                for (TopicPartition partition : partitions) {
                    Long offset = getPartitionOffset(partition);
                    consumer.seek(partition,offset);
                }
            }
        });


        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {

                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                commitOffset(topicPartition,record.offset()+1);
            }
        }
    }

    private static void commitOffset(TopicPartition topicPartition, long l) {

    }

    private static Long getPartitionOffset(TopicPartition partition) {
        return null;
    }

}
  • 2. 结果
    Kafka快速入门系列(10) | Kafka的Consumer API操作_kafka_04

  本次的分享就到这里了,

Kafka快速入门系列(10) | Kafka的Consumer API操作_大数据_05

   看 完 就 赞 ， 养 成 习 惯 ！ ！ ！ color{#FF0000}{看完就赞，养成习惯！！！} 看完就赞，养成习惯！！！^ _ ^ ❤️ ❤️ ❤️
  码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！

本文摘自 ：https://blog.51cto.com/u

开通会员，享受整站包年服务立即开通 >
推荐内容
IT技术
Web编程 Windows编程 编程语言 数据库 移动平台 系统服务 微信平台 其他技术
科技百科
可穿戴设备 电子工程 电力 智能汽车网 光学 智能硬件 环保 风电 机器人 安防 云计算 物联网 3D打印 传感器 锂电 储能 智能电网 新材料 太阳能光伏 激光 医疗科技 氢能 工控 照明 显示 智慧城市 半导体照明 光通讯 人工智能 新能源汽车 电源 仪器仪表 通信 VR 智能家居