修改：消费监听使用专门线程

src/com/zdemo/kafak/KafakConsumer.java

@@ -16,7 +16,6 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Properties;
-import java.util.concurrent.CompletableFuture;
 
 /**
  * 消费
@@ -33,7 +32,7 @@ public abstract class KafakConsumer<T extends Event> implements IConsumer<T>, Se
 
     @Override
     public void init(AnyValue config) {
-        CompletableFuture.runAsync(() -> {
+        new Thread(() -> {
             try (FileInputStream fis = new FileInputStream(new File(APP_HOME, "conf/kafak.properties"));) {
                 Properties props = new Properties();
                 props.load(fis);
@@ -59,6 +58,6 @@ public abstract class KafakConsumer<T extends Event> implements IConsumer<T>, Se
             } catch (IOException e) {
                 e.printStackTrace();
             }
-        });
+        }).start();
     }
 }

src/com/zdemo/kafak/KafakProducer.java

@@ -50,7 +50,6 @@ public class KafakProducer<T extends Event> implements IProducer<T>, Service {
     @Override
     public void send(T... t) {
         for (T x : t) {
-            logger.finest("send message: " + JsonConvert.root().convertTo(x));
             producer.send(new ProducerRecord(x.getTopic(), JsonConvert.root().convertTo(x)));
         }
     }

src/com/zdemo/redis/RedisConsumer.java

@@ -11,7 +11,6 @@ import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
 import java.net.InetSocketAddress;
 import java.net.Socket;
-import java.util.concurrent.CompletableFuture;
 
 public abstract class RedisConsumer<T extends Event> implements IConsumer<T>, Service {
 
@@ -28,7 +27,7 @@ public abstract class RedisConsumer<T extends Event> implements IConsumer<T>, Se
 
     @Override
     public void init(AnyValue config) {
-        CompletableFuture.runAsync(() -> {
+        new Thread(() -> {
             try {
                 Socket client = new Socket();
                 client.connect(new InetSocketAddress(host, port));
@@ -72,6 +71,6 @@ public abstract class RedisConsumer<T extends Event> implements IConsumer<T>, Se
             } catch (Exception e) {
                 e.printStackTrace();
             }
-        });
+        }).start();
     }
 }

test/com/zdemo/test/AppTest.java

@@ -5,8 +5,6 @@ import com.zdemo.redis.RedisProducer;
 import org.junit.Test;
 import org.redkale.boot.Application;
 
-import java.util.Map;
-
 /**
  * 消息发布订阅测试
  */

test/com/zdemo/test/MyConsumer.java

@@ -1,14 +1,14 @@
 package com.zdemo.test;
 
 import com.zdemo.Event;
-import com.zdemo.redis.RedisConsumer;
+import com.zdemo.kafak.KafakConsumer;
 import org.redkale.convert.json.JsonConvert;
 import org.redkale.util.TypeToken;
 
 import java.util.Collection;
 import java.util.List;
 
-public class MyConsumer extends RedisConsumer<Event<String>> {
+public class MyConsumer extends KafakConsumer<Event<String>> {
 
     public String getGroupid() {
         return "group-test"; //quest、user、im、live
@@ -16,7 +16,7 @@ public class MyConsumer extends RedisConsumer<Event<String>> {
 
     @Override
     public Collection<String> getSubscribes() {
-        return List.of("a", "b", "c");
+        return List.of("a", "b", "c", "vis-log");
     }
 
     @Override