1、kafka简介
Kafka 是一种流处理平台，主要用于处理大量数据流，如实时事件、日志文件和传感器数据等。Kafka的目的是实现高吞吐量、低延迟和高可用性的数据处理。
Kafka提供了一个高度可扩展的架构，可以轻松地添加和删除节点，并且能够处理数百亿条消息/分区。Kafka的消息可以容错，即使某个节点失败，消息也会在集群中的其他节点上得到处理。
2、Kafka 的使用场景包括：
实时数据处理：
Kafka 非常适合处理实时事件，例如实时交易、实时搜索结果和实时推文等。Kafka 可以将数据快速地发布和订阅，从而实现实时处理。
日志文件处理：
Kafka 可以处理大量的日志文件，例如 Web 服务器日志、数据库日志和操作系统日志等。Kafka可以将日志文件的数据快速地发布和订阅，并且提供了多种聚合和分析日志数据的方法，例如使用 Apache Storm 或 ApacheFlink。
传感器数据处理：
Kafka 可以处理来自传感器的数据，例如温度、湿度和气压等传感器数据。Kafka可以将传感器数据快速地发布和订阅，并且可以将数据发送到分布式处理系统，例如 Apache Hadoop 或 ApacheSpark，进行处理。
流处理：
Kafka 可以作为流处理系统，例如 Apache Nifi 或 Apache Beam 的底层存储系统。Kafka可以将数据流快速地发布和订阅，并且可以支持多种流处理模式，例如按时间排序、字段过滤和路由规则。
分布式消息队列：
Kafka 可以作为分布式消息队列，用于多个应用程序之间的数据传输。Kafka提供了多种消息传输模式，例如点对点模式、多主节点模式和发布/订阅模式。
数据备份：
Kafka 可以作为数据备份系统，用于备份数据到多个节点上。Kafka 可以将数据发布到多个主题中，从而实现数据备份。
3、简单使用
Kafka 的使用非常简单，可以使用多种语言来编写客户端库和消费者。以下是使用 Java 客户端库的示例：
首先，您需要在项目中添加 Kafka 依赖项：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
然后，您需要编写一个生产者，以将消息发布到指定的主题中：
package com.yinfeng.test.demo.kafka;
import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* @author admin
* @date 2023/7/2 19:02
* @description
*/
public class KafkaProducerDemo {
@SneakyThrows
public static void main(String[] args) {
Properties props = new Properties();
// Kafka 集群地址
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<String, String> producer = new KafkaProducer<>(props);
// 发送3条消息
for (int i = 0; i < 3; i++) {
ProducerRecord<String, String> record1 = new ProducerRecord<>("test", "key"+i, "hello"+i);
producer.send(record1, (metadata, exception) -> {
System.out.println("消息发送成功 topic="+metadata.topic()+", msg=>" + record1.value());
});
}
// kafka异步发送，延时等待执行完成
Thread.sleep(5000);
}
}
执行之后可在控制台看到3条消息已发送
最后，您需要编写一个消费者，以从指定的主题中接收消息：
package com.yinfeng.test.demo.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* @author admin
* @date 2023/7/2 19:02
* @description
*/
public class KafkaConsumerDemo {
public static void main(String[] args) {
Properties props = new Properties();
// Kafka 集群地址
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my_group");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "true");
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, new StringDeserializer(), new StringDeserializer());
consumer.subscribe(Collections.singleton("test"));
// 循环拉取消息
while (true){
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.println("Received message: " + record.value());
}
}
}
}
以上示例只是一个简单的例子，Kafka 还提供了更多的功能和配置选项，例如订阅多个主题、设置消息过滤器和消息压缩等。
4、总结
总的来说，Kafka 是一个非常强大的数据处理平台，可以用于实时数据处理、日志文件处理、传感器数据处理和流处理等场景。其使用简单、功能丰富，并且可以扩展到数百亿条消息/分区，适用于各种大规模的数据处理场景。
使用 Kafka 需要一定的学习和配置，但是一旦您熟悉了其使用方法，Kafka 将会成为您的一个得力工具，可以提高您的工作效率并为您的业务增添价值。 如果您对使用 Kafka 有疑问或遇到问题，可以查看 Kafka 的官方文档和社区，或者使用 Kafka 提供的丰富的客户端库和文档，包括 Apache Kafka、Apache Confluent Kafka 和 Confluent Kubernetes 等。
总之，Kafka 是一种高性能、可扩展、易于使用的数据处理平台，可以广泛应用于实时数据处理、日志文件处理、传感器数据处理和流处理等领域。
