springboot - kafka实践

Kafka是一个开源的分布式流处理平台，由Apache软件基金会开发和维护。它是一种高性能、可持久化、可扩展的消息队列系统，常用于解决大规模数据传输和处理的问题。

以下是Kafka的一些核心概念和主要特点：

1. 消息和主题：Kafka基于发布订阅模式，消息被发布到一个或多个主题（Topic）中。每条消息都有一个唯一的标识符称为偏移量（Offset）。

2. 分区和副本：每个主题可以被分为一个或多个分区（Partition），每个分区在不同的Kafka节点上存储副本（Replica）。分区和副本提供了高可用性和容错性。

3. 生产者和消费者：生产者（Producer）将消息发布到主题，消费者（Consumer）从主题订阅并消费消息。

4. 流处理：Kafka支持流处理，即可以实时处理和转换数据流，支持复杂的流处理逻辑。

5. 可扩展性：Kafka具有良好的可扩展性，可以通过添加更多的节点来增加处理能力和存储容量。

6. 持久化：Kafka使用日志式存储结构，将所有的消息持久化到磁盘上，从而保证数据的可靠性和持久性。

Kafka在很多场景中被广泛使用，例如实时日志收集、事件驱动架构、微服务通信、大数据处理等。它具有高吞吐量、低延迟、可靠性强的特点，并且提供了一系列的客户端库和工具来简化使用和集成。

一、引入依赖

1     <dependency>
2       <groupId>org.springframework.kafka</groupId>
3       <artifactId>spring-kafka</artifactId>
4     </dependency>

二、生产消息

 1 import org.apache.kafka.clients.producer.KafkaProducer;
 2 import org.apache.kafka.clients.producer.ProducerRecord;
 3 import org.apache.kafka.common.serialization.StringSerializer;
 4 
 5 import java.util.Properties;
 6 
 7 /**
 8  * @Classname KafkaProducer
 9  * @Created by Michael
10  * @Date 2023/7/19
11  * @Description 消息生产者
12  */
13 public class MessageProducer {
14   public static void main(String[] args) {
15     Properties properties = new Properties();
16     properties.setProperty("bootstrap.servers","192.168.3.100:9092");// kafka部署服务器
17     properties.setProperty("key.serializer", StringSerializer.class.getName());
18     properties.setProperty("value.serializer", StringSerializer.class.getName());
19     KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
20 
21     //准备消息
22     String topic = "message"; // 主题
23     Integer partition = 0; // 指定分区
24     long timeMillis = System.currentTimeMillis(); // 毫秒值 15分钟
25     String key = "key-message"; // key
26     String value = "value-this is kafka demo -- "+timeMillis; // value
27 
28     ProducerRecord<String, String> record
29             = new ProducerRecord<>(topic, partition, timeMillis, key, value);
30 
31     //生产消息
32     producer.send(record);
33     producer.close();
34   }
35 
36 }

三、消费消息

 1 import org.apache.kafka.clients.consumer.ConsumerRecords;
 2 import org.apache.kafka.clients.consumer.KafkaConsumer;
 3 import org.apache.kafka.common.serialization.StringDeserializer;
 4 
 5 import java.util.Arrays;
 6 import java.util.Properties;
 7 
 8 /**
 9  * @Classname MessageConsumer
10  * @Created by Michael
11  * @Date 2023/7/19
12  * @Description 消息消费者
13  */
14 public class MessageConsumer {
15   public static void main(String[] args) {
16     Properties properties = new Properties();
17     properties.setProperty("bootstrap.servers","192.168.3.100:9092");
18     properties.setProperty("group.id", "group-1");
19     properties.setProperty("key.deserializer", StringDeserializer.class.getName());
20     properties.setProperty("value.deserializer", StringDeserializer.class.getName());
21 
22     //创建kafka消费者
23     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
24     //订阅kafka主题
25     consumer.subscribe(Arrays.asList("message"));
26 
27     //持续消费消息
28     ConsumerRecords<String, String> messages;
29 //    while (true) {
30       messages = consumer.poll(1000);//每一秒拉取依次消息
31       if (messages != null && messages.count() > 0) {
32         messages.forEach((record) -> {
33           System.out.printf("========offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
34         });
35 //      }
36     }
37 
38   }
39 }

四、测试结果

运行producer发布消息到kafka，consumer因为订阅了消息，所以从kafka中拉取到消息内容

 注意，demo没做特殊处理，所以这里的消息时可以被重复消费的