kafka学习笔记（三、生产者Producer使用及配置参数）

news2024/11/26 22:47:29

在这里插入图片描述

1.简介

1.1.producer介绍

生产者就是负责向kafka发送消息的应用程序。消息在通过send()方法发往broker的过程中，有可能需要经过拦截器(Interceptor)、序列化器(Serializer)和分区器(Partitioner)的一系列作用后才能被真正的发往broker。
demo:

public class KafkaClient {
	private static final String brokerList = "localhost:9092";
	private static final String tipic = "topic-test";
	
	public static Properties initConfig() {
		Properties props = new Properties();
		props.put("bootstrap.servers", brokerList);
		props.put("key.serializer", "org.apache.kafka.common.seralization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.seralization.StringSerializer");
		props.put("clinet.id", "producer.client.id.test");
		return props;
	} 
	
	public static void main(String[] args) {
		Properties props = initConfig();
		KafkaProducer<String, String> producer = new KafkaProducer<>(props);
		ProducerRecord<String, String> record = new ProducerRecord<>(topic, "kafka producer test.");
		try {
			producer.send(record);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

ProducerRecord类：

public class ProducerRecord<K, V> {
	private final String topic; // 主题
	private final Integer partition; // 分区号
	private final Headers headers; // 消息头部
	private final K key; // 键
	private final V value; // 值
	private final Long timestamp; // 消息的时间戳
}

  • ProducerRecord类中的key属性：
    key用来指定消息的键，不仅是附加消息还可以用来计算分区号而可以让消息发往特定的分区。

    • 同一个key的消息会被划分到同一分区
    • 有key的消息还可以支持日志压缩的功能

  • 必要参数：

    • bootstrap.servers：客户端连接kafka集群所属broker地址(host1:port1)，并非需要所有的broker地址，生产者会从给定的broker中查到其他broker信息，建议至少设置两个以上。
    • key.serializer和value.serializer：broker端接收的消息必须以字节数组(byte[])的形式存在。发往broker之前需要将消息中对应的key和value做相应的序列化操作来转换整字节数组。
  • KafkaProducer是线程安全的，可以在多个线程中共享单个KafkaProducer实例，也可以将KafkaProducer实例进行池化来供其他线程使用。
  • 消息发送有三种模式
    • 发后即忘(fire-and-forget)：producer.send(record);
    • 同步(sync)：producer.send(record).get();
    • 异步(async)：producer.send(record, callback); // 回调函数的调用可以保证分区有序。

1.2.生产者拦截器

用来在消息发送前做一些准备工作，比如按照某个规则过滤、修改消息内容等，也可以用来在发送回调逻辑中做一些定制化的需求。

实现： 需自定义实现org.apache.kafka.clinets.producer.ProducerInterception接口。接口中有三个方法：

public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record); // 对消息进行相应的定制化处理

// 在消息被应答或发送失败时调用，优先于Callback之前执行。此方法运行在Producer的IO线程中，所以
// 此方法中实现的代码逻辑越简单越好，否则会影响消息的发送速度。
public void onAcknowledgement(RecordMetadata metadata, Exception exception);

public void close();

KafkaProducer可以指定多个拦截器形成拦截链。拦截链会按照interceptor.classes参数配置的拦截器顺序来执行。

1.3.序列化

生产者需要用序列化器（Serializer）把对象转换成字节数组才能通过网络发送给kafka。而在对侧，消费者需要反序列化器（Deserializer）把从kafka中收到的字节数组转换成对应的对象。

客户端自带的序列化器都实现了org.apache.kafka.common.serialization.Serializer接口,此接口有三个方法：

public void configure(Map<String, ?> configs, boolean isKey); // 配置当前类，主要确定编码类型
public byte[] serialize(String topic, T data); // 将类型T的数据转换为byte[]
public void close(); // 关闭当前的序列化器，一般情况下为空方法

生产者使用的序列化器和消费者使用的反序列化器是需要一一对应的。

1.4.分区器

为消息分配分区。如果消息ProducerRecord中没有指定partition字段，那么就需要依赖分区器，根据key这个字段来计算partition的值。

kafka提供的默认分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner，它实现了org.apache.kafka.clients.producer.Partitioner接口，此接口中定义了两个方法：

// 用来计算分区号
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
public void close(); // 关闭分区器的时候用来回收一些资源

Partitioner接口还有一个父接口org.apache.kafka.common.Configurable,此接口只有一个方法：

void configure(Map<String, ?> configs); // 获取配置信息机初始化数据

kafka中除了使用默认的分区器进行分区外还可以使用自定义的分区器，只要实现Partitioner即可。

2.整体架构

在这里插入图片描述

2.1.RecordAccumulator讲解

主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源以便提升性能。（缓存大小通过生产者客户端参数buffer.memory配置，默认32MB）。

如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，此时KafkaProducer的send()方法调用要么被堵塞要么抛出异常，这个取决于参数max.block.ms的配置，默认为60秒。

  • 主线程发送过来的消息都会被追加到RecordAccumulator的某个双端队列（Deque）中，RecordAccumulator内部为每个分区都维护了一个双端队列，队列的内容就是ProducerBatch，即Deque<ProducerBatch>。

    注意： ProducerBatch中包含一个或多个ProducerRecord。ProducerRecord是生产者创建的消息，ProducerBatch是一个消息批次，ProducerRecord会被包含在ProducerBatch中，使字节的使用更加紧凑，也可以减少网络请求的次数以提高整体的吞吐量。

  • 消息写入缓存时，追加到Deque的尾部；Sender读取消息时，从Deque的头部读取。
  • Kafka生产者客户端通过java.io.ByteBuffer实现消息内存的创建和释放，不过频繁的创建和释放是比较耗费资源的，所以在RecordAccumulator内部存在一个BufferPool来实现ByteBuffer的复用，达到缓存高效利用的目的。

    BufferPool只针对特定大小（通过batch.size参数指定，默认16KB）的ByteBuffer进行管理，其他大小的ByteBuffer不会缓存到BufferPool中，我们可以适当的调大batch.size参数以便多缓存一些消息。

2.2.消息进入RecordAccumulator的逻辑

当一条ProducerRecord进入RecordAccumulator时：

  1. 先寻找与消息分区对应的Deque（没有则创建）；
  2. 再从这个Deque尾部获取一个ProducerBatch（没有则创建）；
  3. 查看ProducerBatch中是否可以写入这个ProducerRecord，可以则写入不可以则需新建ProducerBatch再写入；

    新建ProducerBatch时需评估这条消息是否超过batch.size参数的大小，如不超过则以batch.size参数大小来创建ProducerBatch（使用完这段内存区域后可以通过BufferPool的管理来复用），若超过则以评估的大小来创建（此段内存区域不会被复用）。

2.3.Sender步骤

Sender线程从RecordAccumulator中获取缓存的消息后：

  1. 将<分区, Deque<ProducerBatch>>转换为<Node, List<ProducerBatch>>，其中Node表示kafka集群的节点；

    对网络连接来说，生产者客户端是与具体的broker节点建立连接，不关心消息的分区；而KafkaProducer的应用逻辑则只关注往哪个分区发送那些消息，所以这里需要做一个应用逻辑层面到网络IO层面的转换。

  2. 将转换的<Node, List<ProducerBatch>>进一步封装成<Node, Request>的形式，这样就可以将Request发往各个Node了。

    这里的Request指kafka的各种协议请求，对于消息的发送而言就是指具体的ProducerRequest。

  3. 请求在从Sender线程发往kafka之前还会以Map<NodeId, Deque<Request>>形式保存到InFlightRequest中，以缓存已经发出去但还没有收到相应的请求。

    通过配置参数可以限制每个连接（客户端与Node之间的连接）最多缓存的请求数。此配置参数为：max.in.flight.requests.per.connection，默认为5，即每个连接最多只能缓存5个未响应的请求，超过此值之后就不能再向这个连接发送更多的请求了，除非有缓存的请求收到了响应。

2.4.元数据

值kafka集群的元数据，这些数据包括集群中有主题信息、主题上的分区信息、分区的leader副本信息、follower副本信息、副本的AR和ISR集合信息、集群的节点信息以及控制器节点信息等。

更新元数据信息的条件：

  • 当客户端中没有需要使用的元数据信息，比如没有指定的Topic信息；
  • 超过metadata.max.age.ms时间没有更新元数据信息，此配置默认为5分钟；

更新元数据信息的步骤：

  1. 元数据更新操作时在是在客户端内部进行的，对外部不可见。
  2. 当需要更新元数据信息时，会先挑选出leastLoadedNode，然后向这个Node发送MetadataRequest请求来获取具体的元数据信息。
  3. 这个跟新操作由Sender线程发起，在创建完MetadataRequest之后同样会存入InFlightRequests，之后的步骤和发送消息类似。

元数据信息Sender线程负责更新，主线程需要读取，所以这里的数据同步通过synchronized和final关键字来保障。

Producer配置参数

参数默认值含义
bootstarp.servers“”指定连接kafka集群的broker地址（可以只有部分broker地址）
key.serializer“”消息中key对应的序列化类，需实现org.apahce.kafka.common.serialization.Serialiaer
value.serializer“”消息中value对应的序列化类，需实现org.apahce.kafka.common.serialization.Serialiaer
client.id“”指定kafkaProducer对应的客户端id（用来标记消息是从哪个客户端发来的）
acks“1”指定分区中必须有多少个副本收到此条消息生产者才会认为这条消息是成功写入的。它涉及消息的可靠性和吞吐量之间的权衡。asks=1:leader成功写入则返回成功；acks=0:发送消息后不需要等待服务器的响应；scks=-1/all:ISK集合中所有副本成功写入才能收到服务器的成功响应。
buffer.memory32MB生产者客户端中用于缓存消息的缓冲区大小。详见本章节#2.1
batch.size16KB用于指定producerBatch可以复用的内存区域的大小。
max.request.size1MB限制生产者客户端能发送消息的最大值，一般不建议盲目的增大，因为此参数与broker端的message.max.bytes参数有联动。
retries0生产者重试次数
retry.backoff.ms100设定两次重试之间的时间间隔
metadata.max.age30000ms如果在这个时间内无数据没有更新的话会被强制更新
compression.type“none”指定消息的压缩方式，默认情况下消息不会被压缩，，该参数还可以配置为"gzip"、"snappy"和"l24"。消息压缩可极大的减少网络传输量，减低网络的IO，提高整体的性能，是一种以时间换空间的优化方式。
connections.max.idle.ms540000ms指定多久之后关闭闲置的连接
linger.ms0指定生产者发送producerBatch之前等待更过的producerRecord加入ProducerBatch的时间。生产者客户端会在producerBatch被填满或等待时间超过linger.ms值时发送出去。
receive.buffer.bytes32KB设置socket接收消息缓冲区（SO_RECBUF）的大小。如果设置为-1，则使用操作系统的默认值；如果Producer与kafka处于不同的机房，则可以适当调大这个参数值。
send.buffer.bytes128KB设置socket发送消息缓冲区的大小…
request.timeout.ms30000ms配置Producer等待请求响应的最大时间。注意此参数需要比broker端参数replica.lag.time.max.ms的值大，这样可以减少因客户端重试而引起的消息重复的概率。
max.block.ms60000用来控制kafkaProducer中send()方法和partitionsFor()方法的阻塞时间。当生产者的送缓冲区已满或者没有可用的元数据时，这些方法就会阻塞。
partitioner.class~Defaultpartitioner用来指定分区器，需要实现org.apache.kafka.clients.producer.partitioner
enable.idempotencefalse是否开启幂等性功能。所谓幂等简单说就是对接口的多次调用所产生的结果和调用一次是一致的。生产者在进行重试的时候有可能会重复的写入消息，而使用kafka的幂等性功能之后就可以避免这种情况。
interceptor.classes“”用来设定生产者拦截器，需实现org.apache.kafka.clients.producer.ProducerInterceptor接口。
max.in.fligh.request5限制客户端与Node之间的连接最多缓存的请求数。
per.connection.transactional.idnull设置事务id，必须唯一。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1650803.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【NodeMCU实时天气时钟温湿度项目 4】通过NTPClient库获取实时网络时间并显示在TFT屏幕上

【NodeMCU实时天气时钟温湿度项目 4】通过NTPClient库获取实时网络时间并显示在TFT屏幕上

今天是【实时天气时钟温湿度项目】第四专题&#xff0c;主要内容是&#xff1a;学习导入NTPClient库&#xff0c;通过这个库获取实时网络时间&#xff0c;显示在1.3寸TFT液晶屏幕上。此前三个专题&#xff0c;请选择查看以下链接。 第一专题内容&#xff0c;请参考 【N…
阅读更多...
使用Docker安装MySQL5.7.36

使用Docker安装MySQL5.7.36

拉取镜像并查看 docker pull mysql:5.7.36拉取成功后查看&#xff08;非必须&#xff09; docker images创建并设置宿主机 mysql 配置文件目录和数据文件目录 创建相关文件夹将容器中的mysql数据保存到本地&#xff0c;这样即使容器被删除&#xff0c;数据也不会丢失。 mkd…
阅读更多...
牛客NC142 最长重复子串【中等 字符串 Java/Go】

牛客NC142 最长重复子串【中等 字符串 Java/Go】

题目 题目链接&#xff1a; https://www.nowcoder.com/practice/4fe306a84f084c249e4afad5edf889cc 思路 注意&#xff1a;题目给的时间复杂度是O(N^2)那么直接套用双重循环&#xff1a;外层循环i为假定起始重复子串的初始位置&#xff0c;内层循环的j为假定重复子串的结束位置…
阅读更多...
每日一题5：Pandas-修改列

每日一题5：Pandas-修改列

一、每日一题 一家公司决定增加员工的薪水。 编写一个解决方案&#xff0c;将每个员工的薪水乘以2来 修改 salary 列。 返回结果格式如下示例所示。 解答&#xff1a; import pandas as pddef modifySalaryColumn(employees: pd.DataFrame) -> pd.DataFrame:employees.loc[…
阅读更多...
人工神经网络初步

人工神经网络初步

大家好&#xff0c;这里是七七&#xff0c;由于各种比赛的缘故&#xff0c;使用了很多人工神经网络模型。但是很多的原理都不是很明白&#xff0c;就导致了不能灵活地运用┭┮﹏┭┮。为此&#xff0c;去看了些人工神经网络原理书&#xff0c;写下此专题。 在进入正文之前要先…
阅读更多...
Delta lake with Java--数据增删改查

Delta lake with Java--数据增删改查

之前写的关于spark sql 操作delta lake表的&#xff0c;总觉得有点混乱&#xff0c;今天用Java结合真实的数据来进行一次数据的CRUD操作&#xff0c;所涉及的数据来源于Delta lake up and running配套的 GitGitHub - benniehaelen/delta-lake-up-and-running: Companion reposi…
阅读更多...
Mac idea gradle解决异常: SSL peer shut down incorrectly

Mac idea gradle解决异常: SSL peer shut down incorrectly

系统&#xff1a;mac 软件&#xff1a;idea 解决异常: SSL peer shut down incorrectly 查看有没有安装 gradle -v安装 根据项目gradle提示安装版本 brew install gradle7idea的配置 在settings搜索gradle&#xff0c;配置Local installation&#xff0c;选择自己的安装目录…
阅读更多...
Unity Shader中获取像素点深度信息

Unity Shader中获取像素点深度信息

1.顶点着色器中对深度进行计算 v2f vert(appdata v) {v2f o;o.pos UnityObjectToClipPos(v.vertex);o.uv TRANSFORM_TEX(v.uv, _MainTex);o.depth (o.pos.z / o.pos.w 1.0) * 0.5; // Normalize depth to [0, 1]return o; }但是达不到预期&#xff0c;最后返回的值一直大于…
阅读更多...
连通“数据”，让制造变“聪明”

连通“数据”，让制造变“聪明”

说起数据智能&#xff0c;你第一时间想到的是什么呢&#xff1f;是科技感十足的智慧城市&#xff1f;还是炫酷的人工智能景象&#xff1f; 数据作为企业的战略资产越来越受到重视&#xff0c;从最初的数据协助业务协同&#xff0c;转化为数据驱动业务&#xff0c;数据驱动运营…
阅读更多...
Hive两代命令行客户端（Hive、Beeline）

Hive两代命令行客户端（Hive、Beeline）

Hive命令行客户端 Hive有两个主要的客户端工具&#xff0c;分别是旧版的Hive CLI&#xff08;Command Line Interface&#xff09;和新版的Beeline。 Hive CLI&#xff1a; Hive CLI 是 Hive 最早期的命令行客户端工具&#xff0c;它使用 JDBC 连接到 Hive 服务器&#xff0c;…
阅读更多...
栈的2道面试题【有效的括号】【用栈实现队列】

栈的2道面试题【有效的括号】【用栈实现队列】

栈的面试题&#xff1a; 1.有效的括号 题目&#xff1a; 有效的括号 给定一个只包括 (&#xff0c;)&#xff0c;{&#xff0c;}&#xff0c;[&#xff0c;] 的字符串 s &#xff0c;判断字符串是否有效。 有效字符串需满足&#xff1a; 左括号必须用相同类型的右括号闭合…
阅读更多...
深入剖析Tomcat(七) 日志记录器

深入剖析Tomcat(七) 日志记录器

在看原书第六章之前&#xff0c;一直觉得Tomcat记日志的架构可能是个“有点东西”的东西。在看了第六章之后呢&#xff0c;额… 就这&#xff1f;不甘心的我又翻了翻logback与新版tomcat的源码&#xff0c;额…&#xff0c;日志架构原来也没那么神秘。本篇文章先过一遍原书内容…
阅读更多...
为什么会查询不到DNS信息？怎么排查？

为什么会查询不到DNS信息？怎么排查？

DNS&#xff08;域名系统&#xff09;是将域名转换为相应 IP 地址的关键系统。查询 DNS 信息具有重要作用&#xff0c;通过查询 DNS 信息&#xff0c;我们可以知道域名对应的 IP 地址&#xff0c;这是最主要的信息&#xff0c;使设备能与目标服务器进行通信&#xff1b;其次是域…
阅读更多...
网络 IO 模式

网络 IO 模式

同步 IO 与异步 IO 同步 IO 和异步 IO 是关于数据读写方式的两种不同模式。 同步 IO 是指在程序读写数据时&#xff0c;需要等待操作完成后才能继续执行后面的程序。这种模式下&#xff0c;当程序使用阻塞式 IO 时&#xff0c;会一直等待IO操作完成&#xff0c;程序会暂停执行…
阅读更多...
Python中设计注册登录代码

Python中设计注册登录代码

import hashlib import json import os import sys # user interface 用户是界面 UI """ 用户登录系统 1.注册 2.登陆 0.退出 """ # 读取users.bin def load(path): return json.load(open(path, "rt")) # 保存user.bin def save(dic…
阅读更多...
Autosar NvM配置-手动配置Nvblock及使用-基于ETAS软件

Autosar NvM配置-手动配置Nvblock及使用-基于ETAS软件

文章目录 前言NvDataInterfaceNvBlockNvM配置SWC配置RTE Mapping使用生成的接口操作NVM总结前言 NVM作为存储协议栈中最顶层的模块，是必须要掌握的。目前项目基本使用MCU带的Dflash模块，使用Fee模拟eeprom。在项目前期阶段，应该充分讨论需要存储的内容，包括应用数据，诊断…
阅读更多...
ETLCloud工具怎么实现多流SQL实时运算？

ETLCloud工具怎么实现多流SQL实时运算？

多流SQL实时运算的特点和应用场景 多流SQL实时运算是一种先进的数据处理技术&#xff0c;它在大数据处理领域中扮演着至关重要的角色&#xff0c;尤其是在需要对多个数据流进行实时分析和处理的应用场景中。该技术结合了SQL&#xff08;结构化查询语言&#xff09;的易用性和流…
阅读更多...
15.计算机网络

15.计算机网络

1.物理层的互联设备 中继器 和 集线器 2.集线器可以看做特殊的多路中继器 集线器 不可以做到自动寻址的功能 3.数据链路层 网桥 和 交换机 4.交换机是多端口网桥 5.网络层 路由器 6.应用层 网关 7.广播域 网络层 可以形成多个广播域 冲突域 网络层数据链路层 可以形成多个冲突域…
阅读更多...
matlab 基于拉依达检验法（3σ准则） 实现多类别多参数的批量异常样本检验 V2.0

matlab 基于拉依达检验法（3σ准则） 实现多类别多参数的批量异常样本检验 V2.0

简介 拉依达检验法&#xff08;3σ准则&#xff09;是一种统计学方法&#xff0c;用于检测数据中的异常值。这种方法基于正态分布的特性来确定数据点是否可能是异常值。以下是关于拉依达检验法&#xff08;3σ准则&#xff09;的详细介绍&#xff1a; 基本原理&#xff1a; 拉…
阅读更多...
分布式锁概述

分布式锁概述

什么是分布式锁 分布式锁是一种在分布式计算环境中用于同步访问共享资源的机制。它的主要目的是在一个分布式系统中&#xff0c;当多个进程或服务需要同时访问同一个资源时&#xff0c;确保任一时刻只有一个进程或服务能够执行涉及该资源的关键操作。这类似于传统单体应用中的…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023