Kafka使用教程

小蓝 • 2024-12-31 11:49 • 编程

一、Kafka简介

Apache Kafka是一个基于发布/订阅模式的分布式流处理平台，由LinkedIn开发，以高吞吐量、可扩展性、持久性等特点而闻名。它主要由Kafka Broker、Producer、Consumer三个部分组成。

二、Kafka安装

1、安装Java环境（Kafka需要Java 8或更高版本）；
2、下载Kafka二进制文件，并解压；
3、配置Kafka环境变量；
4、启动Zookeeper：Kafka使用Zookeeper来维护集群状态；
5、启动Kafka Server。

三、Kafka Producer

Kafka Producer是将数据发布到Kafka Broker的组件。下面是Java版本的Kafka Producer示例代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
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);
for (int i = 0; i < 100; i++)
    producer.send(new ProducerRecord("my-topic", Integer.toString(i), Integer.toString(i)));

producer.close();

以上代码会将0到99的整数按顺序发送到名为“my-topic”的topic中。

四、Kafka Consumer

Kafka Consumer是从Kafka Broker上订阅数据的组件。下面是Java版本的Kafka Consumer示例代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
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(Collections.singletonList("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", record.offset(), record.key(), record.value());
    }
}

consumer.close();

以上代码会消费“my-topic”这个topic的所有消息，并打印出消息的offset、key和value。

五、Kafka Stream

Kafka Stream是一个轻量级的库，它将源数据流转换为另一个数据流，并可以进行聚合、过滤、分组等操作。下面是一个简单的Kafka Stream示例代码：

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream("my-input-topic");
source.filter((key, value) -> value.contains("kafka"))
        .mapValues(value -> value.toUpperCase())
        .to("my-output-topic");

KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();

以上代码将消费“my-input-topic”这个topic上的消息，如果消息内容中包含“kafka”，则将消息转换为大写字母，并发布到“my-output-topic”这个topic上。

六、Kafka Connect

Kafka Connect是一个用于将数据从外部系统导入到Kafka或导出到外部系统的工具。它支持多种数据源和数据目标，例如MySQL、Elasticsearch、HDFS等。以下是一个简单的Kafka Connect配置文件示例：

name=my-source-task 
connector.class=FileStreamSource
file=/path/to/file.txt
topic=my-topic

以上配置会将“/path/to/file.txt”这个文件的内容导入到Kafka的“my-topic”这个topic中。

七、总结

本文对Kafka的各个组件进行了介绍，并给出了Java代码示例。希望本文可以帮助读者快速上手Kafka，并在实践中体验到Kafka的优越性能和可扩展性。

原创文章，作者：小蓝，如若转载，请注明出处：https://www.506064.com/n/303418.html