增加 kafka 示例

README.md

@@ -43,6 +43,7 @@
 ## 消息队列
 
 * [《芋道 Spring Boot 消息队列 RocketMQ 入门》](http://www.iocoder.cn/Spring-Boot/RocketMQ/?github) 对应 [lab-31](https://github.com/YunaiV/SpringBoot-Labs/tree/master/lab-31) 。
+* [《芋道 Spring Boot 消息队列 Kafka 入门》](http://www.iocoder.cn/Spring-Boot/Kafka/?github) 对应 [lab-03](https://github.com/YunaiV/SpringBoot-Labs/tree/master/lab-03) 。
 
 ## 性能测试
 
@@ -78,6 +79,10 @@ RabbitMQ 示例，已经废弃
 
 TODOs
 
+# lab-06
+
+# lab-07
+
 # lab-08
 
 简单测试，Tomcat + Redis 提供最小接口，看看 QPS 。

lab-03/lab-03-kafka-native/src/main/java/cn/iocoder/springboot/lab03/kafkademo/ConsumerMain.java

@@ -0,0 +1,54 @@
+package cn.iocoder.springboot.lab03.kafkademo;
+
+import org.apache.kafka.clients.consumer.Consumer;
+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;
+
+public class ConsumerMain {
+
+    private static Consumer<String, String> createConsumer() {
+        // 设置 Producer 的属性
+        Properties properties = new Properties();
+        properties.put("bootstrap.servers", "127.0.0.1:9092"); // 设置 Broker 的地址
+        properties.put("group.id", "demo-consumer-group"); // 消费者分组
+        properties.put("auto.offset.reset", "earliest"); // 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
+        properties.put("enable.auto.commit", true); // 是否自动提交消费进度
+        properties.put("auto.commit.interval.ms", "1000"); // 自动提交消费进度频率
+        properties.put("key.deserializer", StringDeserializer.class.getName()); // 消息的 key 的反序列化方式
+        properties.put("value.deserializer", StringDeserializer.class.getName()); // 消息的 value 的反序列化方式
+
+        // 创建 KafkaProducer 对象
+        // 因为我们消息的 key 和 value 都使用 String 类型，所以创建的 Producer 是 <String, String> 的泛型。
+        return new KafkaConsumer<>(properties);
+    }
+
+    public static void main(String[] args) {
+        // 创建 KafkaConsumer 对象
+        Consumer<String, String> consumer = createConsumer();
+
+        // 订阅消息
+        consumer.subscribe(Collections.singleton("TestTopic"));
+
+        // 拉取消息
+        while (true) {
+            // 拉取消息。如果拉取不到消息，阻塞等待最多 10 秒，或者等待拉取到消息。
+            ConsumerRecords records = consumer.poll(Duration.ofSeconds(10));
+            // 遍历处理消息
+            records.forEach(new java.util.function.Consumer<ConsumerRecord>() {
+
+                @Override
+                public void accept(ConsumerRecord record) {
+                    System.out.println(record.key() + "\t" + record.value());
+                }
+
+            });
+        }
+    }
+
+}