新增：主题退订

src/com/zdemo/IConsumer.java

@@ -18,4 +18,10 @@ public interface IConsumer<T extends Event> {
     void addEventType(EventType... eventType);
 
     void accept(String topic, String record);
+
+    /**
+     * 取消订阅
+     * @param topic
+     */
+    void unsubscribe(String topic);
 }

src/com/zdemo/kafak/KafakConsumer.java

@@ -35,18 +35,14 @@ public abstract class KafakConsumer extends AbstractConsumer implements IConsume
 
     public abstract String getGroupid();
 
-    private final LinkedBlockingQueue<EventType> queue = new LinkedBlockingQueue<>();
+    private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
 
     @Override
     public void addEventType(EventType... eventTypes) {
         super.addEventType(eventTypes);
-        try {
+
-            for (EventType eventType : eventTypes) {
+        // 增加变更标记
-                queue.put(eventType);
+        queue.add(() -> logger.info("KafakConsumer starting..."));
-            }
-        } catch (InterruptedException e) {
-            logger.log(Level.WARNING, "", e);
-        }
     }
 
     @Override
@@ -58,8 +54,6 @@ public abstract class KafakConsumer extends AbstractConsumer implements IConsume
             props = new Properties();
             props.load(fis);
 
-            if (logger.isLoggable(Level.INFO)) logger.info(getGroupid() + " consumer started!");
-
             new Thread(() -> {
                 try {
                     props.put("group.id", getGroupid());
@@ -78,9 +72,12 @@ public abstract class KafakConsumer extends AbstractConsumer implements IConsume
                             }
                         });
 
-                        // 动态新增订阅
+                        if (!queue.isEmpty()) {
-                        while (!queue.isEmpty()) {
+                            Runnable runnable;
-                            queue.clear();
+                            while ((runnable = queue.poll()) != null) {
+                                runnable.run();
+                            }
+
                             consumer.unsubscribe();
                             consumer.subscribe(getTopics());
                         }
@@ -95,5 +92,8 @@ public abstract class KafakConsumer extends AbstractConsumer implements IConsume
         }
     }
 
-
+    @Override
+    public void unsubscribe(String topic) {
+        queue.add(() -> eventMap.remove(topic)); // 加入延时执行队列（下一次订阅变更检查周期执行）
+    }
 }

src/com/zdemo/pulsar/PulsarConsumer.java

@@ -22,18 +22,14 @@ public abstract class PulsarConsumer extends AbstractConsumer implements IConsum
 
     public abstract String getGroupid();
 
-    private final LinkedBlockingQueue<EventType> queue = new LinkedBlockingQueue<>();
+    private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
 
     @Override
     public void addEventType(EventType... eventTypes) {
         super.addEventType(eventTypes);
-        try {
+
-            for (EventType eventType : eventTypes) {
+        // 增加变更标记
-                queue.put(eventType);
+        queue.add(() -> logger.info("PulsarConsumer add new topic!"));
-            }
-        } catch (InterruptedException e) {
-            logger.log(Level.WARNING, "", e);
-        }
     }
 
     @Override
@@ -41,22 +37,21 @@ public abstract class PulsarConsumer extends AbstractConsumer implements IConsum
         if (!preInit()) {
             return;
         }
+        queue.add(() -> logger.info("PulsarConsumer starting ..."));
         new Thread(() -> {
             try {
                 client = PulsarClient.builder()
                         .serviceUrl(serviceurl)
                         .build();
 
-                consumer = client.newConsumer()
-                        .topics(new ArrayList<>(getTopics()))
-                        .subscriptionName(getGroupid())
-                        .subscriptionType(SubscriptionType.Shared)
-                        .subscribe();
-
                 while (true) {
                     // 动态新增订阅
-                    while (!queue.isEmpty()) {
+                    if (!queue.isEmpty()) {
-                        queue.clear();
+                        Runnable runnable;
+                        while ((runnable = queue.poll()) != null) {
+                            runnable.run();
+                        }
+
                         consumer.unsubscribe();
                         consumer = client.newConsumer()
                                 .topics(new ArrayList<>(getTopics()))
@@ -90,7 +85,7 @@ public abstract class PulsarConsumer extends AbstractConsumer implements IConsum
     }
 
     @Override
-    public void destroy(AnyValue config) {
+    public void unsubscribe(String topic) {
 
     }
 }

src/com/zdemo/redis/RedisConsumer.java

@@ -101,4 +101,14 @@ public abstract class RedisConsumer extends AbstractConsumer implements IConsume
             }
         }
     }
+
+    @Override
+    public void unsubscribe(String topic) {
+        try {
+            writer.write("UNSUBSCRIBE " + topic + "\r\n");
+            writer.flush();
+        } catch (IOException e) {
+            logger.log(Level.WARNING, "", e);
+        }
+    }
 }