cache 1

Author: cosmoswei

pom.xml

@@ -91,6 +91,12 @@
             <artifactId>netty-all</artifactId>
             <version>4.1.109.Final</version>
         </dependency>
+        <dependency>
+            <groupId>org.reflections</groupId>
+            <artifactId>reflections</artifactId>
+            <version>0.10.2</version>
+        </dependency>
+
     </dependencies>
 
 </project>

src/main/java/com/wei/seqMq/ConsumerInfo.java

@@ -17,4 +17,11 @@ public class ConsumerInfo {
      */
     private String consumerName;
 
+    public ConsumerInfo(String streamName, String groupName) {
+        this.streamName = streamName;
+        this.groupName = groupName;
+    }
+
+    public ConsumerInfo() {
+    }
 }

src/main/java/com/wei/seqMq/ListenerTask.java

@@ -27,17 +27,21 @@ public void run() {
     public void listen() {
         log.info("start listen stream message...");
         while (canLoop()) {
-            Map<StreamMessageId, Map<String, String>> streamMessageIdMapMap = readEventsFromStreamTtl(consumerInfo, Duration.ofSeconds(5));
-            log.info("listen msg = {}", streamMessageIdMapMap);
-            if (null == streamMessageIdMapMap) {
-                continue;
-            }
-            for (StreamMessageId streamMessageId : streamMessageIdMapMap.keySet()) {
-                Map<String, String> stringMap = streamMessageIdMapMap.get(streamMessageId);
-                if (stringMap == null) {
+            try {
+                Map<StreamMessageId, Map<String, String>> streamMessageIdMapMap = readEventsFromStreamTtl(consumerInfo, Duration.ofSeconds(10));
+                log.info("listen msg = {}", streamMessageIdMapMap);
+                if (null == streamMessageIdMapMap) {
                     continue;
                 }
-                handlerListen(consumerInfo, streamMessageId, stringMap);
+                for (StreamMessageId streamMessageId : streamMessageIdMapMap.keySet()) {
+                    Map<String, String> stringMap = streamMessageIdMapMap.get(streamMessageId);
+                    if (stringMap == null) {
+                        continue;
+                    }
+                    handlerListen(consumerInfo, streamMessageId, stringMap);
+                }
+            } catch (Throwable e) {
+                log.error("listen throw exception cause = {}", e.getMessage(), e.getCause());
             }
         }
     }

src/main/java/com/wei/seqMq/MyStreamListener.java

@@ -3,9 +3,14 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
+@StreamConsumer(
+        groupName = "SeqMessageQueue",
+        streamName = "SeqGroup",
+        autoAck = true
+)
 class MyStreamListener implements StreamListener<SeqMessage> {
 
-    boolean actAck = true;
+    private boolean actAck = true;
 
     @Override
     public void onMessage(SeqMessage message) {
@@ -16,4 +21,8 @@ public void onMessage(SeqMessage message) {
     public boolean autoAck() {
         return actAck;
     }
+
+    public void setActAck(boolean actAck) {
+        this.actAck = actAck;
+    }
 }

src/main/java/com/wei/seqMq/SeqMessageQueue.java

@@ -75,10 +75,16 @@ public static void main(String[] args) throws InterruptedException {
         }
         queue.processPending(streamKey, streamGroup);
 
-        MyStreamListener streamListener = new MyStreamListener();
 
+        // 创建容器实例
+        StreamMessageListenerContainer container = StreamMessageListenerContainer.create();
+
+        // 自动扫描并注册监听器（指定扫描包路径）
+        StreamConsumerRegistrar.register(container, "com.wei.seqMq");
         StreamMessageListenerContainer streamMessageListenerContainer = StreamMessageListenerContainer.create();
-        streamMessageListenerContainer.receiveAutoAck(consumerInfo, streamListener, true);
+//        MyStreamListener streamListener = new MyStreamListener();
+//        streamListener.setActAck(true);
+//        streamMessageListenerContainer.receiveAutoAck(consumerInfo, streamListener, true);
         streamMessageListenerContainer.start();
         Thread.sleep(60 * 1000);
         queue.shutdown();

src/main/java/com/wei/seqMq/StreamConsumer.java

@@ -0,0 +1,16 @@
+package com.wei.seqMq;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface StreamConsumer {
+    String streamName();      // 流键
+
+    String groupName();  // 消费者组
+
+    boolean autoAck() default true; // 是否自动确认
+}

src/main/java/com/wei/seqMq/StreamConsumerRegistrar.java

@@ -0,0 +1,56 @@
+package com.wei.seqMq;
+
+import org.reflections.Reflections;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Set;
+
+public class StreamConsumerRegistrar {
+
+    public static void register(StreamMessageListenerContainer container, String... basePackages) {
+        Reflections reflections = new Reflections(basePackages);
+        Set<Class<?>> classes = reflections.getTypesAnnotatedWith(StreamConsumer.class);
+        for (Class<?> clazz : classes) {
+            if (StreamListener.class.isAssignableFrom(clazz)) {
+                try {
+                    StreamConsumer annot = clazz.getAnnotation(StreamConsumer.class);
+                    StreamListener<?> listener = (StreamListener<?>) clazz.getDeclaredConstructor().newInstance();
+                    // 设置 autoAck 属性
+                    setAutoAckProperty(listener, annot.autoAck());
+                    // 构建 ConsumerInfo
+                    ConsumerInfo consumerInfo = new ConsumerInfo(annot.streamName(), annot.groupName());
+                    // 注册到容器
+                    if (annot.autoAck()) {
+                        container.receiveAutoAck(consumerInfo, listener);
+                    } else {
+                        container.receive(consumerInfo, listener, false);
+                    }
+                } catch (InstantiationException | IllegalAccessException |
+                         NoSuchMethodException | InvocationTargetException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    private static void setAutoAckProperty(StreamListener<?> listener, boolean autoAck) {
+        try {
+            // 尝试调用 setActAck 方法
+            Method setMethod = listener.getClass().getMethod("setActAck", boolean.class);
+            setMethod.invoke(listener, autoAck);
+        } catch (NoSuchMethodException e) {
+            // 方法不存在，尝试直接设置字段
+            try {
+                Field field = listener.getClass().getDeclaredField("actAck");
+                field.setAccessible(true);
+                field.setBoolean(listener, autoAck);
+            } catch (NoSuchFieldException | IllegalAccessException ex) {
+                throw new RuntimeException("Failed to set autoAck for listener: " + listener.getClass(), ex);
+            }
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException("Error setting autoAck property", e);
+        }
+    }
+}

src/main/java/com/wei/seqMq/StreamMessageListenerContainer.java

@@ -16,7 +16,7 @@ default void receive(ConsumerInfo consumerInfo, StreamListener streamListener) {
         this.receive(consumerInfo, streamListener, false);
     }
 
-    default void receiveAutoAck(ConsumerInfo consumerInfo, StreamListener streamListener, boolean autoAck) {
-        this.receive(consumerInfo, streamListener, autoAck);
+    default void receiveAutoAck(ConsumerInfo consumerInfo, StreamListener streamListener) {
+        this.receive(consumerInfo, streamListener, true);
     }
 }