kafka权威指南学习

**Kafka权威指南**

**前言**

Apache Kafka 是一个分布式流数据平台，用于处理高吞吐量的日志、事件驱动的应用程序以及实时数据管道。它以其高性能、高可用性和易于扩展的特点而闻名。Kafka 的设计初衷是作为一个分布式日志服务，但随着时间的推移，它已经演变为一个更广泛的流数据平台。

**什么是Kafka**

Kafka 是一个基于主题-分区-副本（Topic-Partition-Replica）的分布式消息队列系统。它允许你将数据写入到多个主题中，每个主题可以被分成多个分区，分区中的数据会被复制到多个副本上，以实现高可用性和负载均衡。

**Kafka的组件**

1. **Broker**: Kafka 的核心组件是 Broker，它负责存储和传输消息。每个 Broker 可以作为一个独立的实例，也可以作为一个集群的一部分。
2. **Topic**: Topic 是一个逻辑上的概念，用于组织相关的消息。每个 Topic 可以被分成多个分区。
3. **Partition**: Partition 是一个物理上的概念，用于存储和传输消息。每个 Partition 中的数据会被复制到多个副本上。
4. **Replica**: Replica 是一个 Broker 上存储的 Partition 的副本。

**Kafka的工作流程**

1. **生产者**: 生产者是向 Kafka 写入消息的客户端。它可以将消息写入到指定的 Topic 中。
2. **Broker**: 当生产者向 Kafka 写入消息时，Broker 会接收并存储这些消息。
3. **分区和副本**: Broker 会将消息写入到指定的 Partition 中，并复制到多个副本上，以实现高可用性和负载均衡。
4. **消费者**: 消费者是从 Kafka读取消息的客户端。它可以订阅指定的 Topic 并接收消息。

**Kafka的配置**

1. **broker.id**: 每个 Broker 需要一个唯一的 ID，以区分不同的 Broker。
2. **listeners**: 指定 Broker 接收连接的 IP 和端口号。
3. **zookeeper.connect**: 指定 ZooKeeper 的连接信息，用于管理 Kafka 集群。
4. **num.partitions**: 指定每个 Topic 的 Partition 数量。

**Kafka的命令行工具**

1. **kafka-console-producer**: 用于向 Kafka 写入消息的生产者。
2. **kafka-console-consumer**: 用于从 Kafka读取消息的消费者。
3. **kafka-run-class**: 用于运行 Kafka 的 Java 类。

**Kafka的API**

1. **Producer API**: 用于向 Kafka 写入消息的生产者 API。
2. **Consumer API**: 用于从 Kafka读取消息的消费者 API。
3. **Admin API**: 用于管理 Kafka 集群的 Admin API。

**Kafka的示例代码**

### 生产者

javaProperties 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 producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "Hello, World!", "This is a message from the producer."));

javaProperties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
while (true) {
 ConsumerRecords records = consumer.poll(100);
 for (ConsumerRecord record : records) {
 System.out.println(record.value());
 }
}

javaAdminClient adminClient = AdminClient.create(props);
List topics = Arrays.asList(
 new NewTopic("my-topic",1, (short)1)
);

adminClient.createTopics(topics);