update 优化 完善kafka案例

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/consumer/KafkaNormalConsumer.java

@@ -14,11 +14,11 @@ import org.springframework.stereotype.Component;
 public class KafkaNormalConsumer {
 
     //默认获取最后一条消息
-    @KafkaListener(topics = "test-topic",groupId = "demo")
-    public void timiKafka(ConsumerRecord record){
+    @KafkaListener(topics = "test-topic", groupId = "test-group-id")
+    public void timiKafka(ConsumerRecord<String, String> record) {
         Object key = record.key();
         Object value = record.value();
-        log.info("【消费者】received the message key {}，value：{}",key,value);
+        log.info("【消费者】received the message key {}，value：{}", key, value);
     }
 
 }

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/producer/KafkaNormalProducer.java

@@ -5,8 +5,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
 
-import java.util.concurrent.CompletableFuture;
-
 /**
  * @author xbhog
  * @date 2024/05/19 18:02
@@ -16,10 +14,9 @@ import java.util.concurrent.CompletableFuture;
 public class KafkaNormalProducer {
 
     @Autowired
-    private KafkaTemplate kafkaTemplate;
+    private KafkaTemplate<String, String> kafkaTemplate;
 
     public void sendKafkaMsg() {
-        CompletableFuture send = kafkaTemplate.send("test-topic", "hello", "kafkaTest");
-        send.join();
+        kafkaTemplate.send("test-topic", "hello", "kafkaTest");
     }
 }

ruoyi-example/ruoyi-test-mq/src/main/resources/application.yml

@@ -24,8 +24,12 @@ spring:
 spring:
   kafka:
     bootstrap-servers: localhost:9092
+    consumer:
+      group-id: test-group-id # 消费者组ID
+      auto-offset-reset: earliest # 当没有偏移量或偏移量无效时，从何处开始消费
     producer:
-      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
 
 --- # rocketmq 配置
 rocketmq: