Kafka 简介与安装

什么是 Kafka

Apache Kafka 是一个分布式流处理平台，具有以下三个核心能力：

消息队列：发布和订阅消息流
存储系统：持久化存储消息流
流处理：实时处理消息流

核心特性

高吞吐量：百万级 TPS
持久化：消息持久化到磁盘
分布式：天然支持集群部署
可扩展：水平扩展能力强
消息回溯：支持消息重新消费

核心概念

架构图

┌─────────────────────────────────────────────────────────────┐
│                      Kafka Cluster                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │   Broker 0  │  │   Broker 1  │  │   Broker 2  │         │
│  │  ┌───────┐  │  │  ┌───────┐  │  │  ┌───────┐  │         │
│  │  │Topic-A│  │  │  │Topic-A│  │  │  │Topic-A│  │         │
│  │  │ P0(L) │  │  │  │ P1(L) │  │  │  │ P2(L) │  │         │
│  │  │ P1(F) │  │  │  │ P2(F) │  │  │  │ P0(F) │  │         │
│  │  └───────┘  │  │  └───────┘  │  │  └───────┘  │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
└─────────────────────────────────────────────────────────────┘
         ↑                                       ↓
     Producer                               Consumer Group
                                           ┌─────────────┐
                                           │ Consumer 0  │
                                           │ Consumer 1  │
                                           │ Consumer 2  │
                                           └─────────────┘

核心术语

概念	说明
Broker	Kafka 服务器节点
Topic	消息主题，逻辑分类
Partition	分区，Topic 的物理划分
Replica	副本，分区的备份
Leader	分区的主副本，负责读写
Follower	分区的从副本，同步 Leader 数据
Producer	消息生产者
Consumer	消息消费者
Consumer Group	消费者组，多个消费者协作消费
Offset	消息在分区中的位置
ZooKeeper	协调服务（新版本可选）

分区与副本

Topic: orders (3 partitions, 2 replicas)

Broker 0          Broker 1          Broker 2
┌─────────┐      ┌─────────┐      ┌─────────┐
│ P0 (L)  │      │ P1 (L)  │      │ P2 (L)  │
│ P1 (F)  │      │ P2 (F)  │      │ P0 (F)  │
└─────────┘      └─────────┘      └─────────┘

L = Leader, F = Follower

消费者组

Topic: orders (3 partitions)

Consumer Group A:
  Consumer 1 → P0
  Consumer 2 → P1
  Consumer 3 → P2

Consumer Group B:
  Consumer 1 → P0, P1, P2

同一消费者组内，一个分区只能被一个消费者消费
不同消费者组之间互不影响

Docker 安装

docker-compose.yml

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

启动

docker-compose up -d

KRaft 模式（无 ZooKeeper）

Kafka 3.0+ 支持 KRaft 模式：

version: '3'
services:
  kafka:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

命令行工具

Topic 管理

# 进入容器
docker exec -it kafka bash

# 创建 Topic
kafka-topics --create \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --partitions 3 \
  --replication-factor 1

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

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

# 删除 Topic
kafka-topics --delete \
  --bootstrap-server localhost:9092 \
  --topic test-topic

生产消息

# 命令行生产者
kafka-console-producer \
  --bootstrap-server localhost:9092 \
  --topic test-topic

消费消息

# 命令行消费者
kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning

# 指定消费者组
kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --group my-group

消费者组管理

# 查看消费者组列表
kafka-consumer-groups --list --bootstrap-server localhost:9092

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

# 重置消费位移
kafka-consumer-groups --reset-offsets \
  --bootstrap-server localhost:9092 \
  --group my-group \
  --topic test-topic \
  --to-earliest \
  --execute

Java 客户端

添加依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

生产者

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringSerializer");
        
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = 
                    new ProducerRecord<>("test-topic", "key-" + i, "value-" + i);
                
                // 异步发送
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("发送成功: partition=%d, offset=%d%n",
                            metadata.partition(), metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                });
            }
        }
    }
}

消费者

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("test-topic"));
            
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息: partition=%d, offset=%d, key=%s, value=%s%n",
                        record.partition(), record.offset(), record.key(), record.value());
                }
            }
        }
    }
}

消息存储

Kafka 消息存储在日志文件中：

/kafka-logs/
└── test-topic-0/          # Topic-Partition
    ├── 00000000000000000000.log    # 日志文件
    ├── 00000000000000000000.index  # 偏移量索引
    └── 00000000000000000000.timeindex  # 时间索引

日志分段

默认每个分段 1GB
到达大小或时间限制时创建新分段
可配置日志保留策略

小结

Kafka 是高吞吐量的分布式消息系统
核心概念：Broker、Topic、Partition、Consumer Group
分区实现并行处理，副本实现高可用
支持 ZooKeeper 模式和 KRaft 模式
消息持久化到磁盘，支持消息回溯

云原生

数据库

消息队列

Kafka 简介与安装

什么是 Kafka

核心特性

核心概念

架构图

核心术语

分区与副本

消费者组

Docker 安装

docker-compose.yml

启动

KRaft 模式（无 ZooKeeper）

命令行工具

Topic 管理

生产消息

消费消息

消费者组管理

Java 客户端

添加依赖

生产者

消费者

消息存储

日志分段

小结

云原生

数据库

消息队列

​什么是 Kafka

​核心特性

​核心概念

​架构图

​核心术语

​分区与副本

​消费者组

​Docker 安装

​docker-compose.yml

​启动

​KRaft 模式（无 ZooKeeper）

​命令行工具

​Topic 管理

​生产消息

​消费消息

​消费者组管理

​Java 客户端

​添加依赖

​生产者

​消费者

​消息存储

​日志分段

​小结

什么是 Kafka

核心特性

核心概念

架构图

核心术语

分区与副本

消费者组

Docker 安装

docker-compose.yml

启动

KRaft 模式（无 ZooKeeper）

命令行工具

Topic 管理

生产消息

消费消息

消费者组管理

Java 客户端

添加依赖

生产者

消费者

消息存储

日志分段

小结