• Kafka (三) --------- Kafka 生产者

    目录

    一、 生产者消息发送流程

    1. 发送原理

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

    在这里插入图片描述

    2. 生产者重要参数列表

    参数名称描述
    bootstrap.servers生产者连接集群所需的 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 之间。
    acks0：生产者发送过来的数据，不需要等数据落盘应答。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。

    二、异步发送 API

    1. 普通异步发送

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

    在这里插入图片描述
    代码编写

    A、创建工程 kafka
    B、导入依赖

    <dependencies>
	<dependency>
		<groupId>org.apache.kafkagroupId>
		<artifactId>kafka-clientsartifactId>
		<version>3.0.0version>
	dependency>
dependencies>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    C、创建包名：com.fancy.kafka.producer

    D、编写不带回调函数的 API 代码

    package com.fancy.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class CustomProducer {
	public static void main(String[] args) throws InterruptedException {
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();
		// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
		// key,value 序列化（必须）：key.serializer，value.serializer
		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");
		// 3. 创建 kafka 生产者对象
		KafkaProducer<String, String> kafkaProducer = new
		KafkaProducer<String, String>(properties);
		// 4. 调用 send 方法,发送消息
		for (int i = 0; i < 5; i++) {
			kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));
		}
		// 5. 关闭资源
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    测试：
    ① 在 hadoop102 上开启 Kafka 消费者。

    [fancyry@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

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

    [fancyry@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
fancyry 0
fancyry 1

    • 1
    • 2
    • 3

    2. 带回调函数的异步发送

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

    发送流程：

    在这里插入图片描述

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

    package com.fancy.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CustomProducerCallback {
	public static void main(String[] args) throws InterruptedException {
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();
		// 2. 给 kafka 配置对象添加配置信息
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
		"hadoop102: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<String, String>(properties);
		// 4. 调用 send 方法,发送消息
		for (int i = 0; i < 5; i++) {
			// 添加回调
			kafkaProducer.send(new ProducerRecord<>("first", "fancyry " + 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(2);
		}
		// 5. 关闭资源
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40

    测试：
    ①在 hadoop102 上开启 Kafka 消费者。

    [fancyry@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

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

    [fancyry@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
fancyry 0
fancyry 1
fancyry 2
fancyry 3
fancyry 4

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

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

    主题：first->分区：0
    主题：first->分区：0
    主题：first->分区：1
    主题：first->分区：1
    主题：first->分区：1

    三、同步发送 API

    同步发送流程

    在这里插入图片描述

    package com.fancy.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducerSync {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();
		// 2. 给 kafka 配置对象添加配置信息
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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<String, String>(properties);
		// 4. 调用 send 方法,发送消息
		for (int i = 0; i < 10; i++) {
			// 异步发送 默认
			// kafkaProducer.send(new
			ProducerRecord<>("first","kafka" + i));
			// 同步发送
			kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();
		}
		// 5. 关闭资源
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    测试：
    ① 在 hadoop102 上开启 Kafka 消费者。

    [fancyry@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

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

    [fancyry@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

    四、生产者分区

    1. 分区好处

    在这里插入图片描述

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

    A、默认的分区器 DefaultPartitioner

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

    /**
* The default partitioning strategy:
* 
      
* 
    • If a partition is specified in the record, use it
* 
    • If no partition is specified but a key is present choose a
partition based on a hash of the key
* 
    • If no partition or key is present choose the sticky
partition that changes when the batch is full.
*
* See KIP-480 for details about sticky partitioning.
*/
public class DefaultPartitioner implements Partitioner {
	... ...
}

      • 1
      • 2
      • 3
      • 4
      • 5
      • 6
      • 7
      • 8
      • 9
      • 10
      • 11
      • 12
      • 13
      • 14

    在这里插入图片描述

    案例一

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

    package com.fancy.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CustomProducerCallbackPartitions {
	public static void main(String[] args) {
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();
		// 2. 给 kafka 配置对象添加配置信息
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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());
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
		for (int i = 0; i < 5; i++) {
			// 指定数据发送到 1 号分区，key 为空（IDEA 中 ctrl + p 查看参数）
			kafkaProducer.send(new ProducerRecord<>("first", 1, "","fancyry " + i), new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception e) {
					if (e == null){
						System.out.println(" 主 题 ： " +
						metadata.topic() + "->" + "分区：" + metadata.partition()
					);
					}else {
						e.printStackTrace();
					}
				}
			});
		}
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31

    测试：

    ① 在 hadoop102 上开启 Kafka 消费者。

    [fancyry@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

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

    [fancyry@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
fancyry 0
fancyry 1
fancyry 2
fancyry 3
fancyry 4

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

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

    主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1

    • 1
    • 2
    • 3
    • 4
    • 5

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

    package com.fancyry.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CustomProducerCallback {
	public static void main(String[] args) {
		Properties properties = new Properties();
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
		for (int i = 0; i < 5; i++) {
			// 依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，分别发往 1、2、0
			kafkaProducer.send(new ProducerRecord<>("first", "a","fancyry " + i), new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception e) {
					if (e == null){
						System.out.println(" 主 题 ： " + metadata.topic() + "->" + "分区：" + metadata.partition());
					} else {
						e.printStackTrace();
					}
				}
			});
		}
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    测试：
    ① key="a"时，在控制台查看结果。

    主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1

    • 1
    • 2
    • 3
    • 4
    • 5

    ② key="b"时，在控制台查看结果。

    主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2

    • 1
    • 2
    • 3
    • 4
    • 5

    ③ key="f"时，在控制台查看结果。

    主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0

    • 1
    • 2
    • 3
    • 4
    • 5

    3. 自定义分区器

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

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

    B、实现步骤

    定义类实现 Partitioner 接口。
    重写 partition()方法。

    package com.fancyry.kafka.producer;
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 {
/*** 返回信息对应的分区
* @param topic 主题
* @param key 消息的 key
* @param keyBytes 消息的 key 序列化后的字节数组
* @param value 消息的 value
* @param valueBytes 消息的 value 序列化后的字节数组
* @param cluster 集群元数据可以查看分区信息
* @return
*/
	@Override
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		// 获取消息
		String msgValue = value.toString();
		// 创建 partition
		int partition;
		// 判断消息是否包含 atguigu
		if (msgValue.contains("atguigu")){
			partition = 0;
		}else {
			partition = 1;
		}
	// 返回分区号
		return partition;
	}
	// 关闭资源
	@Override
	public void close() {
	}
	// 配置方法
	@Override
	public void configure(Map<String, ?> configs) {
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43

    使用分区器的方法，在生产者的配置中添加分区器参数。

    package com.fancy.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CustomProducerCallbackPartitions {
	public static void main(String[] args) throws InterruptedException {
	
		Properties properties = new Properties();
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		// 添加自定义分区器
		properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.fancy.kafka.producer.MyPartitioner");
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
		for (int i = 0; i < 5; i++) {
			kafkaProducer.send(new ProducerRecord<>("first", "fancy " + i), new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception e) {
					if (e == null){
						System.out.println(" 主 题 ： " + metadata.topic() + "->" + "分区：" + metadata.partition());
					} else {
						e.printStackTrace();
					}
				}
			});
		}
		kafkaProducer.close();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28

    测试
    ①在 hadoop102 上开启 Kafka 消费者。

    [fancyry@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

    • 1

    ②在 IDEA 控制台观察回调信息。
    主题：first->分区：0
    主题：first->分区：0
    主题：first->分区：0
    主题：first->分区：0
    主题：first->分区：0

  • 相关阅读:
    【雷丰阳-谷粒商城 】课程概述
    【学习】TensorFlow2环境配置
    虚拟DOM
    Vuex ——详细介绍
    【深度学习】CycleGAN开源项目学习笔记 | 完整流程 | 报错总结 | pytorch
    2核4g服务器能支持多少人访问?阿里云2核4g服务器在线人数
    Python 全栈系列178 单主机使用Docker搭建Mongo分片式集群
    我们这一代人的机会是什么？
    IDEA中maven无法下载依赖解决方案
    （154）Verilog HDL：设计一个选择器之mux2to1
  • 原文地址：https://blog.csdn.net/m0_51111980/article/details/126293150