三十九、大数据技术之Kafka3.x（2）

🌻🌻 目录

一、Kafka 生产者
1.1 生产者消息发送流程
1.1.1 发送原理
1.1.2 生产者重要参数列表

1.2 异步发送API
1.2.1 普通异步发送
1.2.2 带回调函数的异步发送

1.3 同步发送 API
1.4 生产者分区
1.4.1 分区好处
1.4.2 生产者发送消息的分区策略
1.4.3 自定义分区器

1.5 生产经验——生产者如何提高吞吐量
1.6 生产经验——数据可靠性
1.7 生产经验——数据去重
1.7.1 数据传递语义
1.7.2 幂等性
1.7.3 生产者事务

1.8 生产经验——数据有序
1.9 生产经验——数据乱序

一、Kafka 生产者

1.1 生产者消息发送流程

1.1.1 发送原理

在消息发送的过程中，涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。

在这里插入图片描述

1.1.2 生产者重要参数列表

参数名称	描述
- -bootstrap-server	生产者连接集群所需的broker地址清单。例如hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置1个或者多个，中间用逗号隔开。注意这里并非需要所有的broker地址，因为生产者从给定的broker里查找到其他broker信息。
key.serializer和value.serializer	指定发送消息的key和value的序列化类型。一定要写全类名。
buffer.memory	RecordAccumulator缓冲区总大小，`默认32m`。
batch.size	缓冲区一批数据最大值，`默认16k`。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms	如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。单位ms，`默认值是0ms`，表示没有延迟。生产环境建议该值大小为5-100ms之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据，Leader收到数据后应答。-1（all）：生产者发送过来的数据，Leader+和isr队列里面的所有节点收齐数据后应答。`默认值是-1，-1和all是等价的。`
max.in.flight.requests.per.connection	允许最多没有返回ack的次数，`默认为5`，开启幂等性要保证该值是 1-5的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。`默认是int最大值，2147483647`。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是100ms。
enable.idempotence	是否开启幂等性，`默认true`，开启幂等性。
compression.type	生产者发送的所有数据的压缩方式。`默认是none`，也就是不压缩。支持压缩类型：`none、gzip、snappy、lz4和zstd。`

1.2 异步发送API

1.2.1 普通异步发送

1）需求：创建Kafka生产者，采用异步的方式发送到Kafka Broker

在这里插入图片描述

2）代码编写

（1）创建工程kafka

在这里插入图片描述

（2）导入依赖

在这里插入图片描述

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

（3）创建包名：com.gansu.kafka.producer

（4）编写不带回调函数的API代码 CustomProducer

在这里插入图片描述

package com.gansu.kafka.producter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducter {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.102:9092");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 3; i++) {
            kafkaProducer.send(new ProducerRecord<String,String>("first","这个问题我找了好久才找到"+i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

问题：idea无法连接含有kafka的服务器linux,导致idea生产者生产了消息，服务器消费者无法接收

在这里插入图片描述

D:\develop\jdk\jdk\bin\java.exe -agentlib:jdwp=transport=dt_socket,address=127.0.0.1:14340,suspend=y,server=n -javaagent:C:\Users\Administrator\.IntelliJIdea2018.2\system\captureAgent\debugger-agent.jar=file:/C:/Users/Administrator/AppData/Local/Temp/capture8410.props -Dfile.encoding=UTF-8 -classpath "D:\develop\jdk\jdk\jre\lib\charsets.jar;D:\develop\jdk\jdk\jre\lib\deploy.jar;D:\develop\jdk\jdk\jre\lib\ext\access-bridge-
……
com.gansu.kafka.producter.CustomProducter
Connected to the target VM, address: '127.0.0.1:14340', transport: 'socket'
07:43:27.919 [main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values: 
	acks = -1
	batch.size = 16384
	bootstrap.servers = [linux-102:9092]
	buffer.memory = 33554432
……
	value.serializer = class org.apache.kafka.common.serialization.StringSerializer

07:43:30.396 [main] WARN org.apache.kafka.clients.ClientUtils - Couldn't resolve server linux-102:9092 from ……
Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:88)
	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:47)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:413)
	... 4 more
Disconnected from the target VM, address: '127.0.0.1:14340', transport: 'socket'
Process finished with exit code 1

解决：根据AI查询便知，需要在如下 /usr/local/kafka/config/server.properties 中增加配置

在这里插入图片描述

测试：

①在 linux-102上开启Kafka消费者。

./kafka-console-consumer.sh --bootstrap-server linux-102:9092 --topic first

②在IDEA中执行代码，观察linux-102控制台中是否接收到消息。

在这里插入图片描述

1.2.2 带回调函数的异步发送

回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。

在这里插入图片描述

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

copy上面的CustomProducer修改名为CustomProducerCallback再增加回调代码：

在这里插入图片描述

package com.gansu.kafka.producter;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducterCallbacck{

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.102:9092");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 3; i++) {
            kafkaProducer.send(new ProducerRecord<String,String>("first","aa"+i)
                 // 添加回调
                 , new Callback(){
                 // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                   if(exception == null){
                       // 没有异常,输出信息到控制台
                       System.out.println("主题是"+metadata.topic()+",分区是"+metadata.partition());
                   }else{
                       // 出现异常打印
                     exception.printStackTrace();
                   }
                }
            });
            // 延迟一会会看到数据发往不同分区
            Thread.sleep(10);
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

测试：

①在linux-102上开启Kafka消费者。

在这里插入图片描述

②在IDEA中执行代码，观察linux-102控制台中是否接收到消息。

③在IDEA控制台观察回调信息

在这里插入图片描述

1.3 同步发送 API

在这里插入图片描述

只需在异步发送的基础上，再调用一下get()方法即可。copy异步发送代码 CustomProducter，并修改名为 CustomProducterSycn。

在这里插入图片描述

package com.gansu.kafka.producter;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducterSycn {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.102:9092");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 3; i++) {
                                                                                    //增加get()抛出异常
            kafkaProducer.send(new ProducerRecord<String,String>("first","aa"+i)).get();
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

测试：

①在linux-102上开启Kafka消费者。

在这里插入图片描述

./kafka-console-consumer.sh --bootstrap-server linux-102:9092 --topic first

②在IDEA中执行代码，观察linux-102控制台中是否接收到消息。

在这里插入图片描述

1.4 生产者分区

1.4.1 分区好处

在这里插入图片描述

1.4.2 生产者发送消息的分区策略

1）默认的分区器 DefaultPartitioner

在IDEA中ctrl +n，全局查找DefaultPartitioner。

在这里插入图片描述

2）案例一

将数据发往指定partition的情况下，例如，将所有数据发往分区1中。

copy上面的回调 api CustomProducterCallback 并且修改名为 CustomProducterCallbackPartitions进行测试（可注掉线程睡眠）：

在这里插入图片描述

测试：

①在linux-102上开启Kafka消费者。

②在IDEA中执行代码，观察linux-102控制台中是否接收到消息。

③在IDEA控制台观察回调信息。

在这里插入图片描述

3）案例二

没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值。

在这里插入图片描述

1.4.3 自定义分区器

如果研发人员可以根据企业需求，自己重新实现分区器。

1）需求

例如我们实现一个分区器实现，发送过来的数据中如果包含root，就发往0号分区，不包含root，就发往1号分区。

2）实现步骤

(1) 定义分区器类 MyPartitioner 实现Partitioner接口。
(2) 重写partition()方法。

在这里插入图片描述

package com.gansu.kafka.producter;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;
/**
 * 1. 实现接口Partitioner
 * 2. 实现3个方法:partition,close,configure
 * 3. 编写partition方法,返回分区号
 */
public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        /**
         * 返回信息对应的分区
         * @param topic         主题
         * @param key           消息的key
         * @param keyBytes      消息的key序列化后的字节数组
         * @param value         消息的value
         * @param valueBytes    消息的value序列化后的字节数组
         * @param cluster       集群元数据可以查看分区信息
         * @return
         */
        // 获取消息
        String msgValue = value.toString();
        // 创建partitio
        int partition;
        // 判断消息是否包含Daniel
        if(msgValue.contains("Daniel")){

            partition = 0;
        }else

        partition = 1;
        // 返回分区号
        return partition;
    }
    // 关闭资源
    @Override
    public void close() {

    }
    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {

    }
}

(3) 使用分区器的方法，在生产者的配置中（CustomProducterCallbackPartitions）添加分区器参数。(关联自定义分区器)

在这里插入图片描述

package com.gansu.kafka.producter;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducterCallbackPartitions {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.102:9092");

        //关联添加自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.gansu.kafka.producter.MyPartitioner");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 3; i++) {
            // 指定数据发送到1号分区，key为空（IDEA中ctrl + p查看参数）
            // 依次指定key值为a,b,f ，数据key的hash值与3个分区求余，分别发往1、2、0
            kafkaProducer.send(new ProducerRecord<String,String>("first","Daniel-xiaojin"+i)
                 // 添加回调
                 , new Callback(){
                 // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                   if(exception == null){
                       // 没有异常,输出信息到控制台
                       System.out.println("主题是"+metadata.topic()+",分区是"+metadata.partition());
                   }else{
                       // 出现异常打印
                     exception.printStackTrace();
                   }
                }
            });
            // 延迟一会会看到数据发往不同分区
           // Thread.sleep(10);
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

(4) 测试

① 在linux-102上开启Kafka消费者。

./kafka-console-consumer.sh --bootstrap-server linux-102:9092 --topic first

② 在IDEA控制台观察回调信息。

包含Daniel

在这里插入图片描述

不包含Daniel

在这里插入图片描述

1.5 生产经验——生产者如何提高吞吐量

在这里插入图片描述

创建类 CustomProducerParameters

在这里插入图片描述

package com.gansu.kafka.producter;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerParameters {

    public static void main(String[] args) {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.102:9092");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
       // batch.size：批次大小，默认16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        // linger.ms：等待时间，默认0
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
        // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
        properties.put(ProducerConfig.RECEIVE_BUFFER_CONFIG,33554432);
        // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i <3 ; i++) {

            kafkaProducer.send(new ProducerRecord<String,String>("first","Daniel-sys"));

        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

测试：

① 在linux-102上开启Kafka消费者。

② 在IDEA中执行代码，观察linux-102控制台中是否接收到消息。

在这里插入图片描述

1.6 生产经验——数据可靠性

0）回顾发送流程

发送流程

在这里插入图片描述

1）ack应答原理

ack 应答级别

在这里插入图片描述

2）代码配置（复制前面 API类 CustomProducter 修改名为 CustomProducterAcks即可）

在这里插入图片描述

package com.gansu.kafka.producter;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import javax.rmi.PortableRemoteObject;
import java.util.Properties;

public class CustomProducterAcks {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.102:9092");
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //设置acks  打开官网 直接CTRL+F 搜索 ack 即可查看，下面配置海量数据下，还是区别很大的
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        // 重试次数retries，默认是int最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG,3);

        
        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 3; i++) {
            kafkaProducer.send(new ProducerRecord<String,String>("first","aa"+i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

官网搜索可以查看

在这里插入图片描述

1.7 生产经验——数据去重

1.7.1 数据传递语义

在这里插入图片描述

1.7.2 幂等性

1）幂等性原理

在这里插入图片描述

2）如何使用幂等性

开启参数enable.idempotence 默认为true，false关闭。

1.7.3 生产者事务

1）Kafka事务原理

在这里插入图片描述

2）Kafka的事务一共有如下5个API

// 1初始化事务
void initTransactions();

// 2开启事务
void beginTransaction() throws ProducerFencedException;

// 3在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                              String consumerGroupId) throws ProducerFencedException;

// 4提交事务
void commitTransaction() throws ProducerFencedException;

// 5放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;