学习笔记 | Kafka

news2024/9/24 11:33:39

一、概述

定义

1、Kafka传统定义：Kafka 是一个分布式的基于 发布/订阅模式 的消息队列（Message Queue） ，主要应用与大数据实时处理领域。

2、发布/订阅：消息的发送者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接受自己感兴趣的消息。

3、Kafka 最新定义：Kafka是一个开源的 分布式事件流平台 （Event Streaming Platfrom），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

消息队列的应用场景

传统的消息队列主要应用场景包括： 缓存/削峰、解耦和异步通信。

缓存/削峰

所有数据可以全部缓存到消息队列，服务器可以根据自己处理的性能按一定的频率去消息队列中取。

解耦

减少服务之间的直接调用，由消息队列充当中间者。

异步通信

一个业务可以将优化体验（发短信）的动作放到消息队列中，由专门的服务去处理，达到快速响应上游。

消息队列的俩种模式

1）点对点模式

消费者主动拉取数据，消息收到后清除数据。

2）发布/订阅模式

  • 一个队列可以有多个topic主题。（topic对消息进行分类，消费者可以自己需求拿消息）
  • 消费者消费数据之后，不删除数据。
  • 每个消费者相互独立，都可以拿到消费数据。

Kafka的基础架构

1、为方便扩展，并提高吞吐量，一个 Topic 分为多个 partition（分区）

2、配合分区的设计，提出了消费者组的概念，组内每个消费者并行消费，一个分区只能让一个消费者消费。

3、为了提高可用性，为每个 partition 增加诺干副本进行备份（分为leader 和 follower）消费者只找learder，当leader挂掉的时候，follower符合条件时会变成leader。

4、zookerper存储节点信息，有哪些副本。

二、入门

Kafka的基本命令

Topic命令

  • 查看有多少主题
 kafka-topics.sh --bootstrap-server 192.168.204.10:9092,192.168.204.10:9093 --list
  • 新增主题
 kafka-topics.sh --bootstrap-server 192.168.204.10:9092,192.168.204.10:9093 --topic first --create --partitions 1 --replication-factor 3
  • 查看主题详情
kafka-topics.sh --bootstrap-server 192.168.204.10:9092,192.168.204.10:9093 --topic first --describe
  • 修改主题

只能加不能减

kafka-topics.sh --bootstrap-server 192.168.204.10:9092,192.168.204.10:9093 --topic first --alter --partitions 3

命令行操作

  • 创建一个生产者
 kafka-console-producer.sh --bootstrap-server 192.168.204.10:9092 --topic second

  • 创建一个消费者
kafka-console-consumer.sh --bootstrap-server 192.168.204.10:9092 --topic second

可以查看到历史数据

kafka-console-consumer.sh --bootstrap-server 192.168.204.10:9092 --topic second --from-beginning

三、生产者

原理

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

异步发送

当main线程发送到RecordAccumulator之后就结束了，不管接下去的操作。

示例代码：

//配置参数
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.10:9092,192.168.204.10:9093");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

//创建KafkaProducer
KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<>(properties);
kafkaProducer.send(new ProducerRecord<>("second","hello"));

//释放资源
kafkaProducer.close();

回调异步发送

相对于异步发送，就是多了一个发送成功之后处理的函数。

示例代码：

//配置
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.10:9092,192.168.204.10:9093");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

//创客KafkaProducer
KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<>(properties);
kafkaProducer.send(new ProducerRecord<>("second", "hello"), (recordMetadata, e) -> {
    System.out.println(recordMetadata.toString());
    System.out.println("send success");
});

//释放资源
kafkaProducer.close();

同步发送

同步发送就是main线程需要等sender线程将双端队列中的数据发送出去才能继续往下面操作。

示例代码：

//配置参数
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.10:9092,192.168.204.10:9093");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

//创建KafkaProducer
KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<>(properties);
try {
    kafkaProducer.send(new ProducerRecord<>("second","hello")).get();
} catch (InterruptedException e) {
    e.printStackTrace();
} catch (ExecutionException e) {
    e.printStackTrace();
}

//释放资源
kafkaProducer.close();

分区

Kafka分区好处

1、便于合理使用存储资源，每个Partition 在一个Broker上存储，可以把海量数据按照分区切割成一块一块存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。

2、提高并行度，生产者可以以分区为单位发送数；消费者可以以分区为单位进行消费数据。

分区策略

自定义分区器

1、定义自己的分区器

package cn.swj.kafka;

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

import java.util.Map;

/**
 * @Author suweijie
 * @Date 2023/8/30 21:40
 * @Description: TODO
 * @Version 1.0
 */
public class MyPartitioner implements Partitioner {

    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        String msg = o1.toString();

        if(msg.contains("suweijie")) {
            return 1;
        }

        return 0;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

2、添加配置

//配置参数
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.10:9092,192.168.204.10:9093");
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,MyPartitioner.class.getName())

//创建KafkaProducer
KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<>(properties);
try {
    kafkaProducer.send(new ProducerRecord<>("second","hello")).get();
} catch (InterruptedException e) {
    e.printStackTrace();
} catch (ExecutionException e) {
    e.printStackTrace();
}

//释放资源
kafkaProducer.close();

提高生产者的吞吐量

batch.size: 批次的大小默认是16k（16384b） ，但是这个参数要跟linger.ms 配合才有用

linger.ms： 等待时间，修改为 5-100ms ，修改这个会造成数据的延迟。

RecordAccumulator： 双端队列的缓存区大小，修改为64m （33554432b）

compression.type ： 压缩snappy， none（默认）、gzip、snappy（用的比较多）、lz4、zstd

最佳实践：

batch.size = 32768
linger.ms = 5
buffer.memory = 33554432
compression.type = snappy

数据可靠性

应答ACKS

  • 0： 生产者发过来的数据，不需要等待数据落盘应答。
  • 1： 生产者发过来的数据，需要等待Leader收到之后应答。
  • -1（all）： 生产者发过来的数据，需要等Leader+ 和 isr 队列里面所有的节点收齐数据后应答。-1 和 all等价。

spring:
  kafka:
    bootstrap-servers: 192.168.204.10:9092,192.168.204.10:9093,192.168.204.10:9094
    consumer:
      group-id: 1
      value-deserializer: org.apache.kafka.common.serialization.StringSerializer
      key-deserializer: org.apache.kafka.common.serialization.StringSerializer

    producer:
      acks: -1  #ack机制  0 1 -1
      batch-size: 32768  #批次大小
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      compression-type: snappy  #数据压缩
      retries: 5  #重试次数
      buffer-memory: 33554432  #双端队列的缓冲区大小
      linger-ms: 5  # sender 等待时间

数据重复

幂等性特性

配置：

enable:
	idempotence: true  #开启幂等性  默认开启

但是Kafka挂掉之后会重新生成一个PID，所以也是有可能会产生重复数据。

生产者事务

开启事务、必须得开启幂等性

示例代码：

private void transaction() {
        //配置参数
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.10:9092,192.168.204.10:9093");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.LINGER_MS_CONFIG,5); //sender 发送的等待时间 ，当达到这个时间的时候Sender 会直接发
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   //开启幂等性，默认开启
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432); //设置双端队列的大小  64m
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,32768);  //批次的大小  32k ，当批次达到这个大小的时候，Sender会直接发送
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");  //数据的压缩方式
        properties.put(ProducerConfig.RETRIES_CONFIG,5);   //发送失败的重试次数
        properties.put(ProducerConfig.ACKS_CONFIG,-1); // acks的方式 -1 当leader 收到并且和isr 队列里面所有的节点同步才应答。
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"123");  //事务唯一id
        //自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());

        //创建KafkaProducer
        KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<>(properties);

        kafkaProducer.initTransactions();  //初始化事务

        kafkaProducer.beginTransaction();  //开启事务

        try {
            kafkaProducer.send(new ProducerRecord<>("second","hello"));
            kafkaProducer.commitTransaction();  //事务
        } catch (Exception e) {
            e.printStackTrace();
            kafkaProducer.abortTransaction();
        }

        //释放资源
        kafkaProducer.close();
    }

数据有序

同分区内消费者可以实现数据的有序消费，不同分区内消费者如何实现有序消费？TODO

数据乱序问题

产生的原因：

1、默认 broker 最多缓存5个请求

2、当sender一直在发送数据的时候，当有一条数据发送失败需要返回双端队列进行重发，就会产生数据乱序的问题。

解决方案：

1） kafka 在 1.x 版本之前确保单分区下数据有序需要增加以下配置：

max.in.flight.requests.per.connection = 1

1） kafka在 1.x 以及之后的版本确保单分区下的额数据有序，条件如下：

（1） 未开启幂等性

max.in.flight.requests.per.connection 设置为1

（2）开启幂等性

max.in.flight.requests.per.connection 设置小于5

原理：在kafka1.x 版本以后，启用幂等性后，kafka broker 会缓存producer 发来的最近5个request 的元数据，如果数据乱序会将乱序的数据保存在内存中，重新排序之后在落盘。

四、Broker

ZK存储

启动zkCli.sh:

docker exec -it zookeeper-server bash
#进入之后启动zkCli.sh
bin/zkCli.sh
ls /brokers/ids
get /brokers/topics/second/partitions/0/state 
get /controller

/brokes/ids : 记录有哪些节点

/brokers/topics/主题/patitions/0/state ： 记录着leader、isr队列

/controller ： 辅助选举leader

Broker工作原理

AR: kafka 分区中所有的副本统称

工作流程：

1） broker 启动会在zk中注册

2） controller 谁注册，谁说了算

3） 由选举出来的controller 监听 brokers 节点变化

4） Controller 决定 Leader 的选举

选举规则：

在isr队列中存活为前提，安装ARa中排在最前面的优先。例如 ar[1,0,2]、isr[1,0,2]，那么leader 就会按照1，0，2的顺序轮询。

5） 主broker的Controller，会将所有节点的信息上传到zk

6） 其他节点的controller 会去从zk同步相关信息下来。

7） 假设broker挂了

8） 监听到broker节点变化

9） 获取isr

10） 选举新的leader

11） 更新leader 以及 isr

新节点的服役以及退役(没听懂)

新节点服役

docker run  -d --name kafka3 \
--network kafka-net \
-p 9095:9095 \
-e  KAFKA_BROKER_ID=3 \
-e KAFKA_ZOOKEEPER_CONNECT=zookeeper-server:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.204.10:9095 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9095 \
-e TZ="Asia/Shanghai" \
wurstmeister/kafka:latest

查看在新节点是否有主题信息(指定这台broker的地址，查看是否有主题信息)

kafka-topics.sh --bootstrap-server 192.168.204.10:9094 --topic first --describe

服役新节点、正确退役旧节点

五、Kafka 副本

基本信息

1）Kafka 副本作用： 提高数据的可靠性。

2）Kafka默认的副本数为1，生产环境正常配置俩个，保证数据的可靠性；太多副本会增加磁盘的存储空间，增加网络上数据传输，降低效率。

3）Kafka 中副本分为： Leader 和 Follower。Kafka生产者只会把数据发送到Leader，然后Follower 自己去找Leader 同步。

4）Kafka 分区中的所有副本统称为AR（Assigned Replicas）。

AR = ISR + OSR

ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出ISR。该时间闽值由 replica.lagtime.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从ISR 中选举新的 Leader。

OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本。

Leader的选举流程

Follower的故障

Leader的故障

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

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

相关文章

企业网盘全方位解读：热门云存储工具的优势与适用场景

企业网盘全方位解读：热门云存储工具的优势与适用场景

企业网盘无疑是当下最热门的企业协同工具。什么是企业网盘&#xff1f;企业网盘与个人网盘又有什么不同呢&#xff1f;一文全方位解读企业网盘这一热门云存储工具。 什么是企业网盘 企业网盘为企业级文件存储、管理与共享平台&#xff0c;企业团队可以在企业网盘中存储企业文…
阅读更多...
【python】连上钉钉机器人每日推送

【python】连上钉钉机器人每日推送

使用Python向钉钉机器人发送消息 导入必要的库 导入json库用于处理JSON数据&#xff0c;time库用于获取当前时间&#xff0c;requests库用于发送HTTP请求。 定义send_ding_message函数 该函数接收一个消息作为参数&#xff0c;并通过POST请求发送给钉钉机器人。请求的URL和头部…
阅读更多...
阿里云服务器可用区是什么？

阿里云服务器可用区是什么？

阿里云服务器地域和可用区怎么选择&#xff1f;地域是指云服务器所在物理数据中心的位置&#xff0c;地域选择就近选择&#xff0c;访客距离地域所在城市越近网络延迟越低&#xff0c;速度就越快&#xff1b;可用区是指同一个地域下&#xff0c;网络和电力相互独立的区域&#…
阅读更多...
JS加密/解密之常见的JS代码加密

JS加密/解密之常见的JS代码加密

在软件开发领域&#xff0c;混淆&#xff08;JS&#xff09;是一种常见的技术&#xff0c;通过改变代码结构、命名和逻辑&#xff0c;增加代码的复杂性&#xff0c;使其对于逆向工程者变得更加困难。然而&#xff0c;有时候开发者可能需要解开混淆&#xff0c;以便理解、维护或…
阅读更多...
jupyter如何更改默认保存路径

jupyter如何更改默认保存路径

jupyter更改默认路径 jupyter默认路径在‘c\用户\Administrator’下&#xff0c;很不方便。 接下来看如何更改默认路径: 1、找到Anaconda Prompt,打开 2、 输入conda activate env1&#xff0c;其中env1为自己创建的环境&#xff0c;如果不知道怎么创建&#xff0c;按照下面…
阅读更多...
技术学习周刊第 1 期

技术学习周刊第 1 期

2018 年参与过 1 年的 ARTS 打卡&#xff0c;也因为打卡有幸加入了 MegaEase 能与皓哥&#xff08;左耳朵耗子&#xff09;共事。时过境迁&#xff0c;皓哥已经不在了&#xff0c;自己的学习梳理习惯也荒废了一段时间。 2024 年没给自己定具体的目标&#xff0c;只要求自己好好…
阅读更多...
大模型迎来“AppStore时刻”，OpenAI给2024的新想象

大模型迎来“AppStore时刻”，OpenAI给2024的新想象

一夜之间&#xff0c;OpenAI公布了多个重磅消息&#xff0c;引发市场关注。 钛媒体App 1月5日消息&#xff0c;今晨&#xff0c;OpenAI公司向所有GPT开发者们发布一封邮件称&#xff0c;下周将上线自定义的“GPT Store”商店&#xff0c;这有望推动ChatGPT开发者生态不断完善。…
阅读更多...
java基础-给个一键三联呗^_^哈哈

java基础-给个一键三联呗^_^哈哈

文章目录 1.注释修改注释字体三种注释方式 2.标识符和关键字3.数据类型4.类型转换5. 变量、常量、作用域6.基本运算符7.自增自减运算符、初识Math类8.逻辑运算符、位运算符9.三元运算符及小结10.包机制11.JavaDoc生成文档 1.注释 修改注释字体 打开设置Settings 三种注释方…
阅读更多...
IDEA 每次新建工程都要重新配置 Maven的解决方案

IDEA 每次新建工程都要重新配置 Maven的解决方案

文章目录 IDEA 每次新建工程都要重新配置 Maven 解决方案一、选择 File -> New Projects Setup -> Settingsfor New Projects…二、选择 Build,Execution,Deployment -> Build Tools -> Maven IDEA 每次新建工程都要重新配置 Maven 解决方案 DEA 每次新建工程都要…
阅读更多...
lv14 ioctl、printk及多个此设备支持 6

lv14 ioctl、printk及多个此设备支持 6

1 ioctl操作实现 对相应设备做指定的控制操作&#xff08;各种属性的设置获取等等&#xff09; long xxx_ioctl (struct file *filp, unsigned int cmd, unsigned long arg); 功能&#xff1a;对相应设备做指定的控制操作&#xff08;各种属性的设置获取等等&#xff09; 参数…
阅读更多...
关于vite的glob坑

关于vite的glob坑

我先展示一段代码&#xff1a; /*** function 根据pages路径动态生成路由* param {Array} 基础路由*/ export default function (routes) {const modules import.meta.glob("../pages/**/page.js", { eager: true, import: "default" });const comps im…
阅读更多...
CSS3渐变属性详解

CSS3渐变属性详解

渐变属性 线性渐变 概念&#xff1a;线性渐变&#xff0c;指的是在一条直线上进行的渐变。在线性渐变过程中&#xff0c;起始颜色会沿着一条直线按顺序过渡到结束颜色 语法&#xff1a; background:linear-gradient(渐变角度&#xff0c;开始颜色&#xff0c;结束颜色);渐变…
阅读更多...
循环队列的队空队满情况

循环队列的队空队满情况

有题目&#xff1a; 循环队列放在一维数组A[0....M-1]中&#xff0c;end1指向队头元素&#xff0c;end2指向队尾元素的后一个位置。假设队列两端均可进行入队和出队操作&#xff0c;队列中最多能容纳M-1个元素。初始时为空。下列判断队空和队满的条件中&#xff0c;正确的是 …
阅读更多...
移动通信原理与关键技术学习（第四代蜂窝移动通信系统）

移动通信原理与关键技术学习（第四代蜂窝移动通信系统）

前言&#xff1a;LTE 标准于2008 年底完成了第一个版本3GPP Release 8的制定工作。另一方面&#xff0c;ITU 于2007 年召开了世界无线电会议WRC07&#xff0c;开始了B3G 频谱的分配&#xff0c;并于2008 年完成了IMT-2000&#xff08;即3G&#xff09;系统的演进——IMT-Advanc…
阅读更多...
进程与计划任务管理

进程与计划任务管理

目录 一、进程 1.进程相关概念 2.判断线程 3.进程的命令 ps命令 top命令 pstree命令 kill与killall命令 二、计划任务 1.一次性执行任务 2.定时性周期任务 一、进程 1.进程相关概念 程序&#xff1a;保存在硬盘等介质中的可执行的代码。 进程&#xff1a;正在运行…
阅读更多...
C++学习笔记（二十四）：c++ this

C++学习笔记（二十四）：c++ this

this指针在c中较为常用。this是一个指向当前对象实例的指针&#xff0c;通过this指针&#xff0c;可以访问该类的成员函数。示例如下&#xff1a;this指针主要的使用场景是在类内部调用类外部的函数&#xff0c;该函数传递的参数是调用该函数的类对象&#xff0c;代码示例如下&…
阅读更多...
关于整形提升

关于整形提升

个人主页&#xff1a;Lei宝啊 愿所有美好如期而遇 什么是整型提升&#xff1f; 在C语言的整型算数运算总是至少以int类型来进行的&#xff0c;当表达式中有char&#xff0c;byte&#xff0c;short类型的操作数时&#xff0c;他们在被使用前需要被转换成int类型&#xff0c;…
阅读更多...
归并排序例题——逆序对的数量

归并排序例题——逆序对的数量

做道简单一点的题巩固一下 归并排序实现步骤 将整个区间 [l, r] 划分为 [l, mid] 和 [mid1, r]。 递归排序 [l, mid] 和 [mid1, r]。 将左右两个有序序列合并为一个有序序列。 题目描述 给定一个长度为 n 的整数数列&#xff0c;请计算数列中的逆序对的数量。 逆序对的定义…
阅读更多...
linux 02 vmware的快照，文件管理

linux 02 vmware的快照，文件管理

01.快照 使用快照&#xff1a; 同时的快照管理器&#xff1a; 如果想要返回快照&#xff0c;选择要选择的快照&#xff0c;跳转 02. 文件管理&#xff1a; cd 02.touch 2. mkdir 文件夹 mkdir -p 文件夹 &#xff08;创建之前没有的上级文件夹&#xff09;
阅读更多...
HTML5+CSS3小实例：人物介绍卡片2.0

HTML5+CSS3小实例：人物介绍卡片2.0

实例：人物介绍卡片2.0 技术栈：HTML+CSS 效果： 源码： 【HTML】 <!DOCTYPE html> <html lang="zh-CN"> <head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023