Kafka 详细介绍

1. 概述

Apache Kafka 是一个分布式流处理平台，由 LinkedIn 开发并开源，主要用于构建实时数据管道和流应用。它具备高吞吐量、低延迟、高可扩展性和持久性，广泛应用于日志收集、消息系统、流处理等场景。

2. 核心概念

• Producer: 生产者，负责将消息发布到 Kafka 的 Topic。
• Consumer: 消费者，从 Topic 订阅并消费消息。
• Broker: Kafka 服务器，存储消息并处理客户端请求。
• Topic: 消息类别，生产者发布消息到 Topic，消费者从中订阅。
• Partition: Topic 的分区，每个 Partition 是一个有序、不可变的消息序列。
• Offset: 消息在 Partition 中的唯一标识，消费者通过 Offset 跟踪消费进度。
• Consumer Group: 消费者组，组内消费者共同消费一个 Topic，每条消息只被组内一个消费者处理。

3. 架构

• Broker 集群: 多个 Broker 组成集群，提供高可用性和负载均衡。
• Zookeeper: 管理 Broker 和 Topic 的元数据，协调集群。
• Producer/Consumer API: 提供生产者和消费者的编程接口。

4. 特点

• 高吞吐量: 支持每秒百万级消息处理。
• 低延迟: 消息传递延迟在毫秒级。
• 持久性: 消息持久化到磁盘，支持数据保留策略。
• 可扩展性: 通过增加 Broker 和 Partition 实现水平扩展。
• 容错性: 通过副本机制保证数据不丢失。

5. 使用场景

• 日志收集: 集中收集分布式系统日志。
• 消息系统: 解耦生产者和消费者，实现异步通信。
• 流处理: 实时处理数据流，如点击流分析、实时推荐。

6. 安装与配置

1. 下载 Kafka: 从 Apache Kafka 官网 下载。
2. 启动 Zookeeper:

   bin/zookeeper-server-start.sh config/zookeeper.properties
   

3. 启动 Kafka Broker:

   bin/kafka-server-start.sh config/server.properties
   

4. 创建 Topic:

   bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   

5. 启动 Producer:

   bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092
   

6. 启动 Consumer:

   bin/kafka-console-consumer.sh --topic test --bootstrap-server localhost:9092 --from-beginning
   

7. 常用命令

• 查看 Topic 列表:

  bin/kafka-topics.sh --list --bootstrap-server localhost:9092
  

• 查看 Topic 详情:

  bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092
  

• 删除 Topic:

  bin/kafka-topics.sh --delete --topic test --bootstrap-server localhost:9092
  

8. 客户端编程

• Java Producer:

  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);
  producer.send(new ProducerRecord<>("test", "key", "value"));
  producer.close();
  

• Java Consumer:

  Properties props = new Properties();
  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", "test-group");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");Consumer<String, String> consumer = new KafkaConsumer<>(props);
  consumer.subscribe(Arrays.asList("test"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}
  }
  

9. 生态系统

• Kafka Connect: 用于与其他系统集成，实现数据导入导出。
• Kafka Streams: 用于构建流处理应用。
• KSQL: 用于流数据的 SQL 查询。

10. 参考资料

• Apache Kafka 官方文档
• 《Kafka 权威指南》

通过这些内容，你可以全面了解 Kafka 的基本概念、架构、使用方法及其生态系统。