花于陌上开个人站

第四章 Kafka入门基础 v1.0 ​

第一部分 Kafka概述 ​

一、简介 ​

Kafka是由LinkedIn开发的一个分布式流处理平台，具有高吞吐、低延迟、可持久化、分布式等特点，常用于日志收集、消息系统、用户行为追踪等场景。

二、核心概念 ​

概念说明
BrokerKafka服务节点，一个集群由多个Broker组成
Topic消息的主题/分类，逻辑上的消息容器
PartitionTopic的物理分区，一个Topic可有多个Partition，实现并行处理
Producer消息生产者，负责发送消息到Topic
Consumer消息消费者，从Topic读取消息
Consumer Group消费者组，组内消费者共同消费一个Topic，每条消息只被组内一个消费者消费
Replica副本，Partition的备份，保障数据高可用
Offset消费者在Partition中的消费位置

三、架构特点 ​

  • 高吞吐：单机支持每秒百万级消息
  • 持久化：消息持久化到磁盘，支持数据保留策略
  • 分布式：支持水平扩展，多副本保障高可用
  • 实时性：生产者发布消息后消费者可以立即消费

第二部分 安装与部署 ​

一、单机版搭建 ​

1.Docker安装 ​

bash
# 拉取Kafka镜像（使用Bitnami镜像，较官方镜像更易用）
docker pull bitnami/kafka:3.6

# 启动Zookeeper（Kafka依赖Zookeeper）
docker run -d \
  --name zookeeper \
  -p 2181:2181 \
  -e ALLOW_ANONYMOUS_LOGIN=yes \
  bitnami/zookeeper:latest

# 启动Kafka
docker run -d \
  --name kafka \
  -p 9092:9092 \
  -e KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181 \
  -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e ALLOW_PLAINTEXT_LISTENER=yes \
  --link zookeeper \
  bitnami/kafka:3.6

2.Docker Compose安装（推荐） ​

yaml
version: '3'
services:
  zookeeper:
    image: bitnami/zookeeper:latest
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

  kafka:
    image: bitnami/kafka:3.6
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
bash
docker-compose up -d

二、验证安装 ​

bash
# 进入Kafka容器
docker exec -it kafka bash

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

# 生产消息
kafka-console-producer.sh --topic test-topic \
  --bootstrap-server localhost:9092

# 另开终端消费消息
kafka-console-consumer.sh --topic test-topic \
  --bootstrap-server localhost:9092 --from-beginning

第三部分 基本操作 ​

一、Topic管理 ​

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

# 查看所有Topic
kafka-topics.sh --list --bootstrap-server localhost:9092

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

# 删除Topic
kafka-topics.sh --delete --topic my-topic \
  --bootstrap-server localhost:9092

二、消息生产与消费 ​

1.命令行方式 ​

bash
# 生产消息（控制台）
kafka-console-producer.sh --topic my-topic \
  --bootstrap-server localhost:9092
> Hello Kafka
> 第二条消息

# 消费消息（控制台，从最新开始）
kafka-console-consumer.sh --topic my-topic \
  --bootstrap-server localhost:9092

# 消费消息（从头开始）
kafka-console-consumer.sh --topic my-topic \
  --bootstrap-server localhost:9092 --from-beginning

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

2.Java客户端方式 ​

Maven依赖：

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

生产者示例：

java
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("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

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

        // 发送消息
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello Kafka");
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("消息发送成功: " + metadata.partition() + "-" + metadata.offset());
            } else {
                exception.printStackTrace();
            }
        });

        producer.close();
    }
}

消费者示例：

java
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("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("auto.offset.reset", "earliest");

        // 创建消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            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());
            }
        }
    }
}

三、消费者组管理 ​

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

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

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

# 重置消费位移到最新
kafka-consumer-groups.sh --group my-group \
  --topic my-topic \
  --bootstrap-server localhost:9092 \
  --reset-offsets --to-latest --execute

第四部分 Spring Boot整合Kafka ​

一、引入依赖 ​

xml
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

二、配置文件 ​

yaml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest

三、生产者 ​

java
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

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

四、消费者 ​

java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

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