📌 第一部分：Apache Kafka 概览与定位

1.1 定义与全称

Apache Kafka 是由 LinkedIn 于 2011 年开源、后成为 Apache 顶级项目的分布式消息队列系统，由 Jay Kreps、Jun Rao 和 Neha Narkhede 创建。Kafka 是 实时数据流处理的事实标准。

1.2 核心定位

Kafka 的核心定位是 高吞吐量的分布式消息系统。它提供了：

• 高吞吐量（单机百万级消息/秒）
• 低延迟（毫秒级）
• 持久化存储（基于磁盘，可配置）
• 分布式架构（水平扩展）
• 分区和复制（高可用）
• 发布-订阅模式
• 消息顺序保证
• Exactly-Once 语义
• Kafka Streams（流处理）
• Connector（数据集成）

1.3 主要应用领域

• 日志收集： 集中式日志管理
• 实时监控： 系统指标和告警
• 数据管道： ETL 数据流
• 流处理： 实时计算和分析
• 事件溯源： 事件驱动架构
• 微服务通信： 服务间异步通信
• 用户行为追踪： 点击流分析
• IoT 数据采集： 物联网设备数据

1.4 知名案例

• LinkedIn： Kafka 发源地，处理万亿级消息
• Netflix： 使用 Kafka 处理实时数据
• Uber： 使用 Kafka 处理出行数据
• Twitter： 使用 Kafka 处理推文流
• 阿里巴巴： 使用 Kafka 处理双 11 数据
• 腾讯： 使用 Kafka 处理游戏日志
• 字节跳动： 使用 Kafka 处理用户行为

---

📜 第二部分：Apache Kafka 的历史与发展演进

2.1 诞生背景（2011年）

LinkedIn 在 2010 年面临活动流数据处理挑战，需要一种高吞吐、低延迟的消息系统。Jay Kreps、Jun Rao 和 Neha Narkhede 开发了 Kafka，2011 年开源，2012 年成为 Apache 顶级项目。

2.2 关键版本里程碑

• Kafka 0.7（2011年）： 首次开源
• Kafka 0.8（2013年）： 副本机制（高可用）
• Kafka 0.9（2015年）： 新的消费者 API
• Kafka 0.10（2016年）： Kafka Streams 稳定版
• Kafka 0.11（2017年）： Exactly-Once 语义
• Kafka 2.0（2018年）： 性能优化
• Kafka 2.3（2019年）： Raft 共识协议
• Kafka 2.7（2020年）： 性能提升
• Kafka 3.0（2021年）： 重大重构——移除 ZooKeeper 依赖
• Kafka 3.4（2023年）： KRaft 稳定版
• Kafka 3.7（2024年）： 最新版本

2.3 核心概念

• Producer（生产者）： 发送消息的应用
• Consumer（消费者）： 接收消息的应用
• Topic（主题）： 消息的逻辑分类
• Partition（分区）： Topic 的物理分片
• Broker（代理）： Kafka 服务节点
• Consumer Group： 消费者的逻辑分组
• Offset（偏移量）： 消息在分区中的位置
• Replication（复制）： 数据副本
• ISR（In-Sync Replica）： 同步副本集

---

⚙️ 第三部分：核心操作

3.1 基础命令

# 创建 Topic
kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --topic my-topic \
    --partitions 3 \
    --replication-factor 2

# 查看 Topic 列表
kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看 Topic 详情
kafka-topics.sh --describe \
    --bootstrap-server localhost:9092 \
    --topic my-topic

# 生产者发送消息
kafka-console-producer.sh \
    --broker-list localhost:9092 \
    --topic my-topic

# 消费者消费消息
kafka-console-consumer.sh \
    --bootstrap-server localhost:9092 \
    --topic my-topic \
    --from-beginning

# 查看消费者组
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --list

# 查看消费者组详情
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group my-group \
    --describe

3.2 Java 生产者示例

// Kafka 生产者 Java 示例
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // 配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");  // 等待所有副本确认
        props.put("retries", 3);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);

        // 创建生产者
        Producer producer = new KafkaProducer<>(props);

        // 发送消息
        for (int i = 0; i < 100; i++) {
            ProducerRecord record =
                new ProducerRecord<>("my-topic", "key-" + i, "message-" + i);

            producer.send(record, (metadata, exception) -> {
                if (exception == null) {
                    System.out.printf("发送成功: offset=%d, partition=%d%n",
                        metadata.offset(), metadata.partition());
                } else {
                    exception.printStackTrace();
                }
            });
        }

        producer.close();
    }
}

3.3 Java 消费者示例

// Kafka 消费者 Java 示例
import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");

        // 创建消费者
        KafkaConsumer consumer = new KafkaConsumer<>(props);

        // 订阅 Topic
        consumer.subscribe(Arrays.asList("my-topic"));

        // 消费消息
        try {
            while (true) {
                ConsumerRecords records =
                    consumer.poll(Duration.ofMillis(1000));

                for (ConsumerRecord record : records) {
                    System.out.printf("收到消息: offset=%d, key=%s, value=%s%n",
                        record.offset(), record.key(), record.value());

                    // 手动提交偏移量
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }
}

3.4 Python 示例

# 使用 kafka-python 库
from kafka import KafkaProducer, KafkaConsumer
import json

# 生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',
    retries=3
)

# 发送消息
producer.send('my-topic', {'id': 1, 'name': 'Alice'})
producer.flush()

# 消费者
consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=False
)

# 消费消息
for msg in consumer:
    print(f"收到消息: key={msg.key}, value={msg.value}")
    consumer.commit()

3.5 Spring Boot 集成

// Spring Boot Kafka 配置
@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory producerFactory() {
        Map config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConsumerFactory consumerFactory() {
        Map config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public KafkaListenerContainerFactory kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

// 使用
@Service
public class KafkaService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send("my-topic", message);
    }

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void listen(String message) {
        System.out.println("收到消息: " + message);
    }
}

3.6 Kafka Streams 流处理

// Kafka Streams 示例
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.util.Properties;

public class WordCountStream {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();

        // 读取输入流
        KStream textLines = builder.stream("input-topic");

        // 单词计数
        KTable wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((key, word) -> word)
            .count();

        // 输出到 Topic
        wordCounts.toStream().to("output-topic");

        // 启动流处理
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

---

⚖️ 第四部分：Kafka vs RabbitMQ vs Pulsar

对比项	Kafka	RabbitMQ	Pulsar
吞吐量	极高	高	极高
延迟	低	极低	低
持久化	✅ 磁盘	✅ 可选	✅ 磁盘
消息顺序	分区内保证	队列内保证	分区内保证
流处理	✅ Kafka Streams	❌	✅ Pulsar Functions
适用场景	大数据流	通用消息	云原生流

---

🧠 第五部分：学习建议

推荐学习资源

• Kafka 官方文档： kafka.apache.org
• 《Kafka 权威指南》—— 全面入门
• 《Kafka Streams 实战》—— 流处理实战
• Confluent 文档： docs.confluent.io

---

🎯 总结升华