引言

Kafka是高吞吐量的分布式消息队列。本文介绍Kafka的核心概念和应用。

一、核心概念

1
2
3
4
5
6
Producer（生产者）→ Topic（主题）→ Partition（分区）→ Consumer（消费者）

Topic: 消息类别
Partition: 物理分区
Consumer Group: 消费者组
Offset: 消息位移

二、生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 发送消息
ProducerRecord<String, String> record = 
    new ProducerRecord<>("my-topic", "key", "value");
producer.send(record);

// 异步发送with回调
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("Sent to partition: " + metadata.partition());
    }
});

三、消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

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());
    }
}

四、分区策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 轮询
// 默认策略，均匀分布

// Key Hash
// 相同key发送到同一分区
producer.send(new ProducerRecord<>("topic", "key1", "value"));

// 自定义分区器
public class MyPartitioner implements Partitioner {
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区逻辑
        return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);
    }
}

总结

Kafka核心：

  1. 高吞吐量：批量发送、压缩
  2. 持久化：消息存储在磁盘
  3. 分区：并行处理
  4. 消费者组：负载均衡

参考资料