kafka学习-生产者
目录
1、消息生产流程
2、生产者常见参数配置
3、序列化器
基本概念
自定义序列化器
4、分区器
默认分区规则
自定义分区器
5、生产者拦截器
作用
自定义拦截器
6、生产者原理解析
1、消息生产流程

2、生产者常见参数配置

3、序列化器
基本概念
- 在Kafka中保存的数据都是字节数组。
- 消息发送前,需要将消息序列化为字节数组进行发送。
- 生产者通过key.serializer和value.serializer指定key和value的序列化器。
- Kafka使用org.apache.kafka.common.serialization.Serializer接口定义序列化器。
- Kafka已实现的序列化器有:ByteArraySerializer、ByteBufferSerializer、BytesSerializer、DoubleSerializer、FloatSerializer、IntegerSerializer、StringSerializer、LongSerializer、ShortSerializer。
自定义序列化器
实现org.apache.kafka.common.serialization.Serializer<T>接口,并实现其中的serializer方法。
@Data
public class User {private Integer userId;private String username;
}public class UserSerializer implements Serializer<User> {@Overridepublic void configure(Map<String, ?> configs, boolean isKey) {// do nothing}@Overridepublic byte[] serialize(String topic, User data) {try {// 如果数据是null,则返回nullif (data == null) return null;Integer userId = data.getUserId();String username = data.getUsername();int length = 0;byte[] bytes = null;if (null != username) {bytes = username.getBytes("utf-8");length = bytes.length;}// 第一个4字节存储userId的值// 第二个4字节存储username字节数组的长度int值// 第三个length长度,存储username序列化之后的字节数组ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);buffer.putInt(userId);buffer.putInt(length);buffer.put(bytes);return buffer.array();} catch (UnsupportedEncodingException e) {throw new SerializationException("序列化数据异常");}}@Overridepublic void close() {// do nothing}
}
4、分区器
默认分区规则
KafkaProducer.partition();DefaultPartitioner.partition();
- 如果record提供了分区号,则使⽤record提供的分区号
- 如果record没有提供分区号,则使⽤key的序列化后的值的hash值对分区数量取模
- 如果record没有提供分区号,也没有提供key,则使⽤轮询的⽅式分配分区号。
自定义分区器
实现org.apache.kafka.clients.producer.Partitioner接口,并实现其中的partition方法。
在生产者参数中通过配置partitioner.class指定自定义分区器。
/*** 自定义分区器*/
public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 此处可以计算分区的数字。// 我们直接指定为2return 2;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}
5、生产者拦截器
作用
在发送消息前,或者在执行回调逻辑前,对消息做一些定制化的处理,比如修改消息,打印消息日志等。此外,Producer允许设置多个拦截器从而形成一条拦截器链,Producer将按照指定顺序调用它们。
自定义拦截器
自定义拦截器实现org.apache.kafka.clients.producer.ProducerInterceptor接口,并实现其中的onSend()、onAcknowledgement()、close()接口。其中:
- onSend(ProducerRecord):Producer 确保在消息被序列化前调⽤该⽅法。⽤户可以在该⽅法中对消息做任何操作,但最好保证不要修 改消息所属的topic和分区,否则会影响⽬标分区的计算。
- onAcknowledgement(RecordMetadata, Exception):该⽅法会在消息被应答之前或消息发送失败时调⽤, 并且通常都是在Producer回调逻辑触发之前。
- close:关闭Interceptor,主要⽤于执⾏⼀些资源清理⼯作。
在生产者参数中通过配置ProducerConfig.INTERCEPTOR_CLASSES_CONFIG指定自定义拦截器。
public class Interceptor<KEY, VALUE> implements ProducerInterceptor<KEY, VALUE> {private static final Logger LOGGER = LoggerFactory.getLogger(InterceptorTwo.class);@Overridepublic ProducerRecord<KEY, VALUE> onSend(ProducerRecord<KEY, VALUE> record) {System.out.println("拦截器---go");// 此处根据业务需要对相关的数据作修改String topic = record.topic();Integer partition = record.partition();Long timestamp = record.timestamp();KEY key = record.key();VALUE value = record.value();Headers headers = record.headers();// 添加消息头headers.add("interceptor", "interceptor".getBytes());ProducerRecord<KEY, VALUE> newRecord = new ProducerRecord<KEY, VALUE>(topic, partition, timestamp, key, value, headers);return newRecord;}@Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {System.out.println("拦截器---back");if (exception != null) {// 如果发生异常,记录在日志中LOGGER.error(exception.getMessage());}}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}
6、生产者原理解析



以上内容为个人学习理解,如有问题,欢迎在评论区指出。
部分内容截取自网络,如有侵权,联系作者删除。
相关文章:
kafka学习-生产者
目录 1、消息生产流程 2、生产者常见参数配置 3、序列化器 基本概念 自定义序列化器 4、分区器 默认分区规则 自定义分区器 5、生产者拦截器 作用 自定义拦截器 6、生产者原理解析 1、消息生产流程 2、生产者常见参数配置 3、序列化器 基本概念 在Kafka中保存的数…...
【Python】设计模式
设计模式分为三种类型,共23类。 创建型模式:单例模式、抽象工厂模式、建造者模式、工厂模式、原型模式。结构型模式:适配器模式、桥接模式、装饰模式、组合模式、外观模式、享元模式、代理模式。行为型模式:模版方法模式、命令模…...
C++ 数字
C 数字 通常,当我们需要用到数字时,我们会使用原始的数据类型,如 int、short、long、float 和 double 等等。这些用于数字的数据类型,其可能的值和数值范围,我们已经在 C 数据类型一章中讨论过。 C 定义数字 我们已…...
code阶段——gitgitlab安装
在code阶段,我们需要将不同版本的代码存储到一个仓库中,常见的版本控制工具就是SVN或者Git,这里我们采用Git作为版本控制工具,GitLab作为远程仓库。 Git安装 https://git-scm.com/(傻瓜式安装) GitLab安…...
C 风格文件输入/输出---无格式输入/输出
C 标准库的 C I/O 子集实现 C 风格流输入/输出操作。 <cstdio> 头文件提供通用文件支持并提供有窄和多字节字符输入/输出能力的函数,而 <cwchar>头文件提供有宽字符输入/输出能力的函数。 无格式输入/输出 从文件流获取字符 std::fgetc, std::getc …...
Spring-MVC的文件上传下载,及插件的使用(让项目开发更节省时间)
目录 一、概述 ( 1 ) 介绍 ( 2 ) 讲述 二、上传 三、下载 四、jrebel的使用 五、多文件上传 给我们带来什么收获 一、概述 ( 1 ) 介绍 Spring MVC的文件上传下载是指在Spring MVC框架中实现文件的上传和下载功能。文件上传是指将本地计算机上的文件上传到服务器端…...
算法 数据结构 递归冒泡算法 java冒泡算法 优化递归冒泡 数据结构(九)
使用递归算法实现冒泡: package com.nami.algorithm.study.day06;import java.util.Arrays;/*** beyond u self and trust u self.** Author: lbc* Date: 2023-09-05 15:36* email: 594599620qq.com* Description: keep coding*/ public class BubbleSort2 {// p…...
【计算机视觉 | 目标检测】目标检测常用数据集及其介绍(十五)
文章目录 一、STN PLAD (STN Power Line Assets Dataset)二、Satlas三、Street Dataset四、UAVVaste五、UDA-CH (Unsupervised Domain Adaptation on Cultural Heritage)六、USB (Universal-Scale Object Detection Benchmark)七、VEDAI (Vehicle Detection in Aerial Imagery)…...
洛谷P8814:解密 ← CSP-J 2022 复赛第2题
【题目来源】https://www.luogu.com.cn/problem/P8814https://www.acwing.com/problem/content/4732/【题目描述】 给定一个正整数 k,有 k 次询问,每次给定三个正整数 ni,ei,di,求两个正整数 pi,qi…...
Flutter实现CombineExecutor进行多个异步分组监听,监听第一个异步执行的开始和最后一个异步执行结束时机。
1.场景 我们在调用接口时,很多时候会同时调用多个接口,接口都是异步执行,我们很难知道调用的多个接口哪个会最后执行完成,我们有时候需要对最后一个接口执行完成的时机监听,所以基于该需求,设计了CombineE…...
2023 年最新Java 毕业设计选题题目参考,500道 Java 毕业设计题目,值得收藏
大家好,我是程序员徐师兄,最近有很多同学咨询,说毕业设计了,不知道选怎么题目好,有哪些是想需要注意的。 确实毕设选题实际上对很多同学来说一个大坑, 每年挖坑给自己跳的人太多太多,选题选得好…...
Mac电脑其他文件占用超过一大半的内存如何清理?
mac的存储空间时不时会提示内存已满,查看内存占用比例最大的居然是「其他文件」,「其他文件」是Mac无法识别的格式文件或应用插件扩展等等...如果你想要给Mac做一次彻底的磁盘空间清理,首当其冲可先对「其他文件」下手,那么我们该…...
geopandas 笔记: datasets 数据集
geopandas 自带的几个数据集 1 世界各个国家 import geopandas as gpd import pandas as pdpd.set_option(display.max_rows,None) gpd.read_file(gpd.datasets.get_path(naturalearth_lowres)) pop_est人口数量continent国家所在的大陆name国家的名称iso_a3国家的三个字母的…...
长胜证券:三大拐点共振 看好智能驾驶新一轮行情
摘要 【长胜证券:三大拐点共振 看好智能驾驭新一轮行情】长胜证券研报指出,全球共振,国内智驾商场正迎来三大拐点:1)技能上,“BEV Transformer数据闭环”新架构2023年开端上车,使得不依靠高精地…...
AIGC专栏5——EasyPhoto AI写真照片生成器 sd-webui插件介绍、安装与使用
AIGC专栏5——EasyPhoto AI写真照片生成器 插件安装与使用 学习前言源码下载地址技术原理储备(SD/Control/Lora)StableDiffusionControlNetLora EasyPhoto插件简介EasyPhoto插件安装安装方式一:Webui界面安装 (需要良好的网络&…...
【Python程序设计】 工厂模式【07/8】
一、说明 我们探索数据工程中使用的设计模式 - 软件设计中常见问题的可重用解决方案。 以下文章是有关 Python 数据工程系列文章的一部分,旨在帮助数据工程师、数据科学家、数据分析师、机器学习工程师或其他刚接触 Python 的人掌握基础知识。 迄今为止,…...
PHP8的多维数组-PHP8知识详解
今天分享的是php8的数组中的多维数组,主要内容有:多维数组的概念、创建和输出二维数组、创建和输出三维数组。 1、多维数组的概念 多维数组是包含一个或多个数组的数组。在多维数组中,主数组中的每一个元素也可以是一个数组,子数…...
【【STM32--28--IO引脚的复用功能】】
STM32–28–IO引脚的复用功能 STM32的IO复用功能 何为复用? 我们先了解一下何为通用 IO端口的输入或输出是由GPIO外设控制,我们称之为通用 复用: IO端口的输入或者是输出是由其他非GPIO外设控制就像经常说的USART 由 DR寄存器进行输出 STM32的IO复用功…...
CodeJock Active-X / COM v22.1.0 Crack
CodeJock Active-X / COM v22.1.0--这个支持 Unicode 啦, Unicode Unicode 创建专业应用程序,其中包含一整套高度可定制的用户界面组件,包括 Visual Studio 风格的对接窗格和 Office 风格的功能区、工具栏和菜单,为您的应用程序…...
mac通过docker搭建elasticsearch:8.9.2以及kibana:8.9.2
1.elasticsearch.yml配置修改: cluster.name: "docker-cluster" network.host: 0.0.0.0 http.port: 9200 #discovery.seed_hosts: ["172.17.0.2"]#----------------------- BEGIN SECURITY AUTO CONFIGURATION ----------------------- # # T…...
生成xcframework
打包 XCFramework 的方法 XCFramework 是苹果推出的一种多平台二进制分发格式,可以包含多个架构和平台的代码。打包 XCFramework 通常用于分发库或框架。 使用 Xcode 命令行工具打包 通过 xcodebuild 命令可以打包 XCFramework。确保项目已经配置好需要支持的平台…...
论文解读:交大港大上海AI Lab开源论文 | 宇树机器人多姿态起立控制强化学习框架(二)
HoST框架核心实现方法详解 - 论文深度解读(第二部分) 《Learning Humanoid Standing-up Control across Diverse Postures》 系列文章: 论文深度解读 + 算法与代码分析(二) 作者机构: 上海AI Lab, 上海交通大学, 香港大学, 浙江大学, 香港中文大学 论文主题: 人形机器人…...
解锁数据库简洁之道:FastAPI与SQLModel实战指南
在构建现代Web应用程序时,与数据库的交互无疑是核心环节。虽然传统的数据库操作方式(如直接编写SQL语句与psycopg2交互)赋予了我们精细的控制权,但在面对日益复杂的业务逻辑和快速迭代的需求时,这种方式的开发效率和可…...
2024年赣州旅游投资集团社会招聘笔试真
2024年赣州旅游投资集团社会招聘笔试真 题 ( 满 分 1 0 0 分 时 间 1 2 0 分 钟 ) 一、单选题(每题只有一个正确答案,答错、不答或多答均不得分) 1.纪要的特点不包括()。 A.概括重点 B.指导传达 C. 客观纪实 D.有言必录 【答案】: D 2.1864年,()预言了电磁波的存在,并指出…...
实现弹窗随键盘上移居中
实现弹窗随键盘上移的核心思路 在Android中,可以通过监听键盘的显示和隐藏事件,动态调整弹窗的位置。关键点在于获取键盘高度,并计算剩余屏幕空间以重新定位弹窗。 // 在Activity或Fragment中设置键盘监听 val rootView findViewById<V…...
C# 求圆面积的程序(Program to find area of a circle)
给定半径r,求圆的面积。圆的面积应精确到小数点后5位。 例子: 输入:r 5 输出:78.53982 解释:由于面积 PI * r * r 3.14159265358979323846 * 5 * 5 78.53982,因为我们只保留小数点后 5 位数字。 输…...
接口自动化测试:HttpRunner基础
相关文档 HttpRunner V3.x中文文档 HttpRunner 用户指南 使用HttpRunner 3.x实现接口自动化测试 HttpRunner介绍 HttpRunner 是一个开源的 API 测试工具,支持 HTTP(S)/HTTP2/WebSocket/RPC 等网络协议,涵盖接口测试、性能测试、数字体验监测等测试类型…...
Ubuntu Cursor升级成v1.0
0. 当前版本低 使用当前 Cursor v0.50时 GitHub Copilot Chat 打不开,快捷键也不好用,当看到 Cursor 升级后,还是蛮高兴的 1. 下载 Cursor 下载地址:https://www.cursor.com/cn/downloads 点击下载 Linux (x64) ,…...
Docker拉取MySQL后数据库连接失败的解决方案
在使用Docker部署MySQL时,拉取并启动容器后,有时可能会遇到数据库连接失败的问题。这种问题可能由多种原因导致,包括配置错误、网络设置问题、权限问题等。本文将分析可能的原因,并提供解决方案。 一、确认MySQL容器的运行状态 …...
《信号与系统》第 6 章 信号与系统的时域和频域特性
目录 6.0 引言 6.1 傅里叶变换的模和相位表示 6.2 线性时不变系统频率响应的模和相位表示 6.2.1 线性与非线性相位 6.2.2 群时延 6.2.3 对数模和相位图 6.3 理想频率选择性滤波器的时域特性 6.4 非理想滤波器的时域和频域特性讨论 6.5 一阶与二阶连续时间系统 6.5.1 …...
