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…...
微信小程序之bind和catch
这两个呢,都是绑定事件用的,具体使用有些小区别。 官方文档: 事件冒泡处理不同 bind:绑定的事件会向上冒泡,即触发当前组件的事件后,还会继续触发父组件的相同事件。例如,有一个子视图绑定了b…...
云计算——弹性云计算器(ECS)
弹性云服务器:ECS 概述 云计算重构了ICT系统,云计算平台厂商推出使得厂家能够主要关注应用管理而非平台管理的云平台,包含如下主要概念。 ECS(Elastic Cloud Server):即弹性云服务器,是云计算…...
【OSG学习笔记】Day 18: 碰撞检测与物理交互
物理引擎(Physics Engine) 物理引擎 是一种通过计算机模拟物理规律(如力学、碰撞、重力、流体动力学等)的软件工具或库。 它的核心目标是在虚拟环境中逼真地模拟物体的运动和交互,广泛应用于 游戏开发、动画制作、虚…...
树莓派超全系列教程文档--(62)使用rpicam-app通过网络流式传输视频
使用rpicam-app通过网络流式传输视频 使用 rpicam-app 通过网络流式传输视频UDPTCPRTSPlibavGStreamerRTPlibcamerasrc GStreamer 元素 文章来源: http://raspberry.dns8844.cn/documentation 原文网址 使用 rpicam-app 通过网络流式传输视频 本节介绍来自 rpica…...
《Qt C++ 与 OpenCV:解锁视频播放程序设计的奥秘》
引言:探索视频播放程序设计之旅 在当今数字化时代,多媒体应用已渗透到我们生活的方方面面,从日常的视频娱乐到专业的视频监控、视频会议系统,视频播放程序作为多媒体应用的核心组成部分,扮演着至关重要的角色。无论是在个人电脑、移动设备还是智能电视等平台上,用户都期望…...
Qt Widget类解析与代码注释
#include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this); }Widget::~Widget() {delete ui; }//解释这串代码,写上注释 当然可以!这段代码是 Qt …...
解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错
出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上,所以报错,到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本,cu、torch、cp 的版本一定要对…...
微信小程序云开发平台MySQL的连接方式
注:微信小程序云开发平台指的是腾讯云开发 先给结论:微信小程序云开发平台的MySQL,无法通过获取数据库连接信息的方式进行连接,连接只能通过云开发的SDK连接,具体要参考官方文档: 为什么? 因为…...
SAP学习笔记 - 开发26 - 前端Fiori开发 OData V2 和 V4 的差异 (Deepseek整理)
上一章用到了V2 的概念,其实 Fiori当中还有 V4,咱们这一章来总结一下 V2 和 V4。 SAP学习笔记 - 开发25 - 前端Fiori开发 Remote OData Service(使用远端Odata服务),代理中间件(ui5-middleware-simpleproxy)-CSDN博客…...
【7色560页】职场可视化逻辑图高级数据分析PPT模版
7种色调职场工作汇报PPT,橙蓝、黑红、红蓝、蓝橙灰、浅蓝、浅绿、深蓝七种色调模版 【7色560页】职场可视化逻辑图高级数据分析PPT模版:职场可视化逻辑图分析PPT模版https://pan.quark.cn/s/78aeabbd92d1...
