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

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

article 2025/1/4 6:03:32

文章目录

一、以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.servers
	properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102
	:9092,hadoop103:9092");
	// 指定对应的 key 和 value 的序列化类型 key.serializer
	properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
	StringSerializer.class.getName());
	properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Strin
	gSerializer.class.getName());
	// 1 创建 kafka 生产者对象
	// "" hello
	KafkaProducer<String,  String>  kafkaProducer  =  new
	KafkaProducer<>(properties);
	// 2 发送数据
	for (int i = 0; i < 5; i++) {
	kafkaProducer.send(new
	ProducerRecord<>("first","atguigu"+i));
	}
	// 3 关闭资源
	kafkaProducer.close();
	}
}