增加 spring boot kafka 示例

Author: YunaiV

lab-03/lab-03-kafka-demo-transaction/src/main/java/cn/iocoder/springboot/lab03/kafkademo/config/KafkaConfiguration.java

@@ -1,19 +1,19 @@
 package cn.iocoder.springboot.lab03.kafkademo.consumer;
 
-import cn.iocoder.springboot.lab03.kafkademo.message.Demo01Message;
+import cn.iocoder.springboot.lab03.kafkademo.message.Demo07Message;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
 @Component
-public class Demo01Consumer {
+public class Demo07Consumer {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
-    @KafkaListener(topics = Demo01Message.TOPIC,
-            groupId = "demo01-consumer-group-" + Demo01Message.TOPIC)
-    public void onMessage(Demo01Message message) {
+    @KafkaListener(topics = Demo07Message.TOPIC,
+            groupId = "demo07-consumer-group-" + Demo07Message.TOPIC)
+    public void onMessage(Demo07Message message) {
         logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
     }
 

lab-03/lab-03-kafka-demo-transaction/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo01AConsumer.java

@@ -1,18 +1,18 @@
 package cn.iocoder.springboot.lab03.kafkademo.message;
 
 /**
- * 示例 01 的 Message 消息
+ * 示例 07 的 Message 消息
  */
-public class Demo01Message {
+public class Demo07Message {
 
-    public static final String TOPIC = "DEMO_01";
+    public static final String TOPIC = "DEMO_07";
 
     /**
      * 编号
      */
     private Integer id;
 
-    public Demo01Message setId(Integer id) {
+    public Demo07Message setId(Integer id) {
         this.id = id;
         return this;
     }
@@ -23,7 +23,7 @@ public Integer getId() {
 
     @Override
     public String toString() {
-        return "Demo01Message{" +
+        return "Demo07Message{" +
                 "id=" + id +
                 '}';
     }

lab-03/lab-03-kafka-demo-transaction/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo04Consumer.java

@@ -0,0 +1,58 @@
+package cn.iocoder.springboot.lab03.kafkademo.producer;
+
+import cn.iocoder.springboot.lab03.kafkademo.message.Demo07Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import javax.annotation.Resource;
+import java.util.concurrent.ExecutionException;
+
+@Component
+public class Demo07Producer {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Resource
+    private KafkaTemplate<Object, Object> kafkaTemplate;
+
+    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
+        // 创建 Demo07Message 消息
+        Demo07Message message = new Demo07Message();
+        message.setId(id);
+        // 同步发送消息
+        return kafkaTemplate.send(Demo07Message.TOPIC, message).get();
+    }
+
+    public SendResult<Object, Object> syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException {
+        return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, SendResult<Object, Object>>() {
+
+            @Override
+            public SendResult<Object, Object> doInOperations(KafkaOperations<Object, Object> kafkaOperations) {
+                // 创建 Demo07Message 消息
+                SendResult<Object, Object> sendResult;
+                Demo07Message message = new Demo07Message();
+                message.setId(id);
+                try {
+//                    sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get();
+                    sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get();
+                    logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+
+                // 本地业务逻辑... biubiubiu
+                runner.run();
+
+                // 返回发送结果
+                return sendResult;
+            }
+
+        });
+    }
+
+}

lab-03/lab-03-kafka-demo-transaction/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo01Consumer.java renamed to lab-03/lab-03-kafka-demo-transaction/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo07Consumer.java

@@ -4,10 +4,11 @@ spring:
     bootstrap-servers: 127.0.0.1:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
     # Kafka Producer 配置项
     producer:
-      acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
+      acks: all # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
       retries: 3 # 发送失败时，重试发送的次数
       key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化
       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
+      transaction-id-prefix: demo. # 事务编号前缀
     # Kafka Consumer 配置项
     consumer:
       auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
@@ -18,6 +19,8 @@ spring:
           json:
             trusted:
               packages: cn.iocoder.springboot.lab03.kafkademo.message
+        isolation:
+          level: read_committed # 读取已提交的消息
     # Kafka Consumer Listener 监听器配置
     listener:
       missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错