大数据技术之KafkaKafka APIKafka监控Flume对接KafkaKafka面试题

Posted @从一到无穷大

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了大数据技术之KafkaKafka APIKafka监控Flume对接KafkaKafka面试题相关的知识，希望对你有一定的参考价值。

文章目录

1 Kafka API

1.1 Producer API

1.1.1 消息发送流程

Kafka 的 Producer 发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程：main 线程和 Sender 线程 ，以及一个线程共享变量 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka broker。

相关参数：
batch.size：只有数据积累到 batch.size 之后，sender 才会发送数据。
linger.ms：如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。

1.1.2 异步发送 API

1. 导入依赖

<dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.0</version>
        </dependency>
    </dependencies>

2. 编写代码
需要用到的类：
KafkaProducer：需要创建一个生产者对象，用来发送数据
ProducerConfig：获取所需的一系列配置参数
ProducerRecord：每条数据都要封装成一个 ProducerRecord 对象

（1）不带回调函数的 API

package com.Tom.producer;

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

import java.util.Properties;

public class MyProducer 
    public static void main(String[] args) 
        //1.创建Kafka生产者的配置信息
        Properties properties = new Properties();

        //2.指定连接的Kafka集群
        properties.put("bootstrap.servers", "hadoop102:9092");

        //3.ACK应答级别
        properties.put("acks", "all");

        //4.重试次数
        properties.put("retries", 3);

        //5.批次大小
        properties.put("batch.size", 16384);    // 16K

        //6.等待时间
        properties.put("linger.ms", 1);

        //7.RecordAccumulator缓冲区大小
        properties.put("buffer.memory", 33554432);    // 32M

        //8.Key, Value的序列化类
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //9.创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //10.发送数据
        for (int i = 0; i < 10; ++i)
            producer.send(new ProducerRecord<String, String>("first","Tom--" + i));
        

        //11.关闭资源
        producer.close();

（2）带回调函数的 API
回调函数会在 producer 收到 ack 时调用，为异步调用， 该方法有两个参数，分别是 RecordMetadata 和 Exception ，如果 Exception 为 null ，说明消息发送成功，如果 Exception 不为 null ，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

package com.Tom.producer;

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.Properties;

public class CallBackProducer 
    public static void main(String[] args) 

        // 1.创建配置信息
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 2.创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

//        ArrayList<String> list = new ArrayList<String>();
//        list.add("a");
//        list.add("b");
//        list.add("c");

        // 3.发送数据
        for (int i = 0; i < 10; ++i)
            producer.send(new ProducerRecord<String,String>("first","Tom","Tom--" + i), new Callback() 
                // 回调函数， 该方法会在Producer收到ack时调用，为异步调用
                public void onCompletion(RecordMetadata metadata, Exception exception) 
                    if (exception == null)
                        System.out.println(metadata.partition() + "--" + metadata.offset());
                     else 
                        exception.printStackTrace();
                    
                
            );
        

        // 4.关闭资源
        producer.close();

1.1.3 同步发送 API

同步发送的意思就是，一条消息发送之后，会阻塞当前线程， 直至返回 ack。
由于 send 方法返回的是一个 Future 对象，根据 Futrue 对象的特点，我们也可以实现同步发送的效果 ，只需在调用 Future 对象的 get 方法即可。

package com.Tom.producer;

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

import java.util.Properties;

public class MyProducer 
    public static void main(String[] args) 
        //1.创建Kafka生产者的配置信息
        Properties properties = new Properties();

        //2.指定连接的Kafka集群
        properties.put("bootstrap.servers", "hadoop102:9092");

        //3.ACK应答级别
        properties.put("acks", "all");

        //4.重试次数
        properties.put("retries", 3);

        //5.批次大小
        properties.put("batch.size", 16384);    // 16K

        //6.等待时间
        properties.put("linger.ms", 1);

        //7.RecordAccumulator缓冲区大小
        properties.put("buffer.memory", 33554432);    // 32M

        //8.Key, Value的序列化类
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //9.创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //10.发送数据
        for (int i = 0; i < 10; ++i)
            producer.send(new ProducerRecord<String, String>("first","Tom--" + i)).get();
        

        //11.关闭资源
        producer.close();

1.2 Consumer API

Consumer 消费数据时的可靠性是很容易保证的，因为数据在 Kafka 中是持久化的，故不用担心数据丢失问题。

由于 consumer 在消费过程中可能会出现断电宕机等故障， consumer恢复后，需要从故障前的位置的继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。

所以 offset 的维护是 Consumer 消费数据是必须考虑的问题。

1.2.1 自动提交 offset

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

以下为自动提交 offset 的代码：

package com.Tom.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 java.util.Arrays;
import java.util.Properties;

public class MyConsumer 
    public static void main(String[] args) 

        // 1 创建消费者配置信息
        Properties properties = new Properties();

        // 2 给配置信息赋值
        // 连接的集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // 开启自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 自动提交的延时
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // Key, Value的反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "bigdata0805");

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 订阅主题
        consumer.subscribe(Arrays.asList("test"));

        // 获取数据
        while(true)
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);

            // 解析并打印consumerRecords
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) 
                System.out.println(consumerRecord.key() + "--" + consumerRecord.value());

消费者重置 offset

package com.huxili.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 java.util.Arrays;
import java.util.Properties;

public class MyConsumer 
    public static void main(String[] args) 

        // 1 创建消费者配置信息
        Properties properties = new Properties();

        // 2 给配置信息赋值
        // 连接的集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // 开启自动提交
        // properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 自动提交的延时
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // Key, Value的反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "bigdata0805");

        // 重置消费者的offset
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 订阅主题
        consumer.subscribe(Arrays.asList("test"));

        // 获取数据
        while(true)
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);

            // 解析并打印consumerRecords
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) 
                System.out.println(consumerRecord.key() + "--" + consumerRecord.value());

1.2.2 手动提交 offset

虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的 开发人员难以把握 offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。

手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步提交） 。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交 ；不同点是 commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（ 由不可控因素导致，也会出现提交失败 ）；而 commitAsync 则没有失败重试机制，故有可能提交失败。

（1）同步提交 offset
由于同步提交 offset 有失败重试机制，故更加可靠 ，以下为同步提交 offset 的示例。

package com.Tom.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class CustomComsumer 

	public static void main(String[] args) 
		Properties props = new Properties();
		
		//Kafka集群
		props.put("bootstrap.servers", "hadoop102:9092");
		
		//消费者组，只要group.id相同，就属于同一个消费者组
		props.put("group.id", "test");
		
		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("first"));//消费者订阅主题
		
		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());
			
			
			//同步提交，当前线程会阻塞直到offset提交成功
			consumer.commitSync();

（2）异步提交 offset
虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会收到很大的影响。因此更多的情况下，会选用异步提交 offset的方式。
以下为

以上是关于大数据技术之KafkaKafka APIKafka监控Flume对接KafkaKafka面试题的主要内容，如果未能解决你的问题，请参考以下文章

大数据技术之KafkaKafka APIKafka监控Flume对接KafkaKafka面试题

大数据技术之KafkaKafka APIKafka监控Flume对接KafkaKafka面试题

大数据技术之KafkaKafka概述Kafka快速入门Kafka架构深入

大数据技术之KafkaKafka概述Kafka快速入门Kafka架构深入

大数据技术之KafkaKafka概述Kafka快速入门Kafka架构深入

大数据技术之KafkaKafka概述Kafka快速入门Kafka架构深入