当前位置：首页 > news >正文

探秘Kafka源码：关键内容解析

news 2025/8/3 4:42:13

文章目录

一、以kafka-3.0.0为例
- 1.1安装 gradle
二、生产者源码
- 2.1源码主流程图
- 2.2 初始化
- 2.3生产者sender线程初始化
- 2.4 程序入口
- 2.5生产者 main 线程初始化
- 2.6 跳转到 KafkaProducer构造方法

一、以kafka-3.0.0为例

打开 IDEA，点击 File->Open…->源码包解压的位置。
在这里插入图片描述

1.1安装 gradle

Gradle是类似于 maven 的代码管理工具。Kafka基于它编译。

IDEA 自动帮你下载安装，下载的时间比较长（网络慢，需要 1 天时间，有 VPN 需要几分钟）。

二、生产者源码

2.1源码主流程图

在这里插入图片描述

2.2 初始化

在这里插入图片描述

2.3生产者sender线程初始化

在这里插入图片描述

2.4 程序入口

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class CustomProducer {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接集群 bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 指定对应的 key 和 value 的序列化类型 key.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 1 创建 kafka 生产者对象// "" helloKafkaProducer<String,  String>  kafkaProducer  =  newKafkaProducer<>(properties);// 2 发送数据for (int i = 0; i < 5; i++) {kafkaProducer.send(newProducerRecord<>("first","atguigu"+i));}// 3 关闭资源kafkaProducer.close();}
}