新增 redission + flink 实现消息总线 模块

tr-dependencies/pom.xml

@@ -46,8 +46,11 @@
         <!--websocket-->
         <tio-websocket.version>3.6.0.v20200315-RELEASE</tio-websocket.version>
 
-        <!--事件主线-->
-        <eventbus.version>3.3.1</eventbus.version>
+        <!--字节码增强-->
+        <javassist.version>3.25.0-GA</javassist.version>
+
+        <!--flink版本-->
+        <flink.version>1.16.1</flink.version>
     </properties>
 
 
@@ -67,6 +70,12 @@
                 <groupId>org.redisson</groupId>
                 <artifactId>redisson-spring-boot-starter</artifactId>
                 <version>${redisson.version}</version>
+                <exclusions>
+                    <exclusion>
+                        <groupId>org.springframework.boot</groupId>
+                        <artifactId>spring-boot-starter-logging</artifactId>
+                    </exclusion>
+                </exclusions>
             </dependency>
 
             <!-- Sa-Token 权限认证，在线文档：https://sa-token.cc -->
@@ -211,13 +220,6 @@
                 <version>${tio-websocket.version}</version>
             </dependency>
 
-            <!--事件主线-->
-            <dependency>
-                <groupId>org.greenrobot</groupId>
-                <artifactId>eventbus-java</artifactId>
-                <version>${eventbus.version}</version>
-            </dependency>
-
 
             <!--业务组件-->
             <dependency>
@@ -232,6 +234,13 @@
                 <version>${revision}</version>
             </dependency>
 
+            <dependency>
+                <groupId>cn.tr</groupId>
+                <artifactId>tr-spring-boot-starter-plugin-flink</artifactId>
+                <version>${revision}</version>
+            </dependency>
+
+
             <dependency>
                 <groupId>cn.tr</groupId>
                 <artifactId>tr-spring-boot-starter-plugin-mybatis</artifactId>
@@ -253,6 +262,12 @@
                 <version>${revision}</version>
             </dependency>
 
+            <!--消息总线-->
+            <dependency>
+                <groupId>cn.tr</groupId>
+                <artifactId>tr-spring-boot-starter-plugin-eventbus</artifactId>
+                <version>${revision}</version>
+            </dependency>
 
             <!--字典插件-->
             <dependency>
@@ -260,6 +275,28 @@
                 <artifactId>tr-spring-boot-starter-plugin-dict</artifactId>
                 <version>${revision}</version>
             </dependency>
+
+            <!--字节码增强技术-->
+            <dependency>
+                <groupId>org.javassist</groupId>
+                <artifactId>javassist</artifactId>
+                <version>${javassist.version}</version>
+            </dependency>
+
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-streaming-java</artifactId>
+                <version>1.16.1</version>
+            </dependency>
+
+
+            <!--flink 组件-->
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-clients</artifactId>
+                <version>${flink.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 </project>

tr-framework/pom.xml

@@ -51,6 +51,11 @@
             <artifactId>spring-boot-starter-web</artifactId>
             <scope>provided</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.javassist</groupId>
+            <artifactId>javassist</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

tr-framework/src/main/java/cn/tr/core/utils/Proxy.java

@@ -0,0 +1,164 @@
+package cn.tr.core.utils;
+
+import cn.hutool.core.util.StrUtil;
+import javassist.*;
+import javassist.bytecode.AnnotationsAttribute;
+import javassist.bytecode.ConstPool;
+import javassist.bytecode.annotation.*;
+import lombok.Getter;
+import lombok.SneakyThrows;
+import org.springframework.util.ClassUtils;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+/**
+ * @author zhouhao
+ * @since 3.0
+ */
+public class Proxy<I> {
+    private static final AtomicLong counter = new AtomicLong(1);
+
+    public final CtClass ctClass;
+    @Getter
+    private final Class<I> superClass;
+    @Getter
+    private final String className;
+    @Getter
+    private final String classFullName;
+
+    private Class<I> targetClass;
+
+    @SneakyThrows
+    public static <I> Proxy<I> create(Class<I> superClass, String... classPathString) {
+        return new Proxy<>(superClass, classPathString);
+    }
+
+    public Proxy(Class<I> superClass, String... classPathString) throws Exception {
+        if (superClass == null) {
+            throw new NullPointerException("superClass can not be null");
+        }
+
+        className = superClass.getSimpleName() + "FastBeanCopier" + counter.getAndAdd(1);
+        classFullName = "cn.tr.plugin.eventbus"+ "." + className;
+
+        this.superClass = superClass;
+        ClassPool classPool = ClassPool.getDefault();
+
+        classPool.insertClassPath(new ClassClassPath(this.getClass()));
+        classPool.insertClassPath(new LoaderClassPath(ClassUtils.getDefaultClassLoader()));
+
+        if (classPathString != null) {
+            for (String path : classPathString) {
+                classPool.insertClassPath(path);
+            }
+        }
+
+
+        ctClass = classPool.makeClass(classFullName);
+        if (superClass != Object.class) {
+            if (superClass.isInterface()) {
+                ctClass.setInterfaces(new CtClass[]{classPool.get(superClass.getName())});
+            } else {
+                ctClass.setSuperclass(classPool.get(superClass.getName()));
+            }
+        }
+        addConstructor("public " + className + "(){}");
+    }
+
+    public Proxy<I> addMethod(String code) {
+        return handleException(() -> ctClass.addMethod(CtNewMethod.make(code, ctClass)));
+    }
+
+    public Proxy<I> addConstructor(String code) {
+        return handleException(() -> ctClass.addConstructor(CtNewConstructor.make(code, ctClass)));
+    }
+
+    public Proxy<I> addField(String code) {
+        return addField(code, null);
+    }
+
+    public Proxy<I> addField(String code, Class<? extends java.lang.annotation.Annotation> annotation) {
+        return addField(code, annotation, null);
+    }
+
+    @SuppressWarnings("all")
+    public static MemberValue createMemberValue(Object value, ConstPool constPool) {
+        MemberValue memberValue = null;
+        if (value instanceof Integer) {
+            memberValue = new IntegerMemberValue(constPool, ((Integer) value));
+        } else if (value instanceof Boolean) {
+            memberValue = new BooleanMemberValue((Boolean) value, constPool);
+        } else if (value instanceof Long) {
+            memberValue = new LongMemberValue((Long) value, constPool);
+        } else if (value instanceof String) {
+            memberValue = new StringMemberValue((String) value, constPool);
+        } else if (value instanceof Class) {
+            memberValue = new ClassMemberValue(((Class) value).getName(), constPool);
+        } else if (value instanceof Object[]) {
+            Object[] arr = ((Object[]) value);
+            ArrayMemberValue arrayMemberValue = new ArrayMemberValue(new ClassMemberValue(arr[0].getClass().getName(), constPool), constPool);
+            arrayMemberValue.setValue(Arrays.stream(arr)
+                    .map(o -> createMemberValue(o, constPool))
+                    .toArray(MemberValue[]::new));
+            memberValue = arrayMemberValue;
+
+        }
+        return memberValue;
+    }
+
+    public Proxy<I> custom(Consumer<CtClass> ctClassConsumer) {
+        ctClassConsumer.accept(ctClass);
+        return this;
+    }
+
+    @SneakyThrows
+    public Proxy<I> addField(String code, Class<? extends java.lang.annotation.Annotation> annotation, Map<String, Object> annotationProperties) {
+        return handleException(() -> {
+            CtField ctField = CtField.make(code, ctClass);
+            if (null != annotation) {
+                ConstPool constPool = ctClass.getClassFile().getConstPool();
+                AnnotationsAttribute attributeInfo = new AnnotationsAttribute(constPool, AnnotationsAttribute.visibleTag);
+                Annotation ann = new javassist.bytecode.annotation.Annotation(annotation.getName(), constPool);
+                if (null != annotationProperties) {
+                    annotationProperties.forEach((key, value) -> {
+                        MemberValue memberValue = createMemberValue(value, constPool);
+                        if (memberValue != null) {
+                            ann.addMemberValue(key, memberValue);
+                        }
+                    });
+                }
+                attributeInfo.addAnnotation(ann);
+                ctField.getFieldInfo().addAttribute(attributeInfo);
+            }
+            ctClass.addField(ctField);
+        });
+    }
+
+    @SneakyThrows
+    private Proxy<I> handleException(Task task) {
+        task.run();
+        return this;
+    }
+
+
+    @SneakyThrows
+    public I newInstance() {
+        return getTargetClass().newInstance();
+    }
+
+    @SneakyThrows
+    public Class<I> getTargetClass() {
+        if (targetClass == null) {
+            targetClass = (Class<I>) ctClass.toClass(ClassUtils.getDefaultClassLoader(), null);
+        }
+        return targetClass;
+    }
+
+    interface Task {
+        void run() throws Exception;
+    }
+}

tr-plugins/pom.xml

@@ -29,6 +29,9 @@
         <module>tr-spring-boot-starter-plugin-websocket</module>
         <module>tr-spring-boot-starter-plugin-sms</module>
         <module>tr-spring-boot-starter-plugin-eventbus</module>
+        <module>tr-spring-boot-starter-plugin-mq</module>
+        <module>tr-spring-boot-starter-plugin-flink</module>
+        <module>tr-spring-boot-starter-plugin-desensitize</module>
     </modules>
 
 </project>

tr-plugins/tr-spring-boot-starter-plugin-banner/pom.xml

@@ -19,6 +19,7 @@
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>
+            <scope>provided</scope>
         </dependency>
 
         <dependency>

tr-plugins/tr-spring-boot-starter-plugin-biz-excel/src/test/java/cn/tr/plugin/excel/ExcelTest.java

@@ -3,7 +3,6 @@ package cn.tr.plugin.excel;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.util.RandomUtil;
 import cn.tr.core.enums.IEnum;
-import cn.tr.plugin.dict.annotation.Dict;
 import cn.tr.plugin.dict.bo.DictBO;
 import cn.tr.plugin.dict.config.cache.DictManager;
 import cn.tr.plugin.excel.annotation.Excel;
@@ -18,7 +17,6 @@ import org.junit.jupiter.api.Test;
 import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
 
 import java.io.File;
-import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.util.ArrayList;
 import java.util.List;

tr-plugins/tr-spring-boot-starter-plugin-desensitize/pom.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>tr-plugins</artifactId>
+        <groupId>cn.tr</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <version>${revision}</version>
+    <artifactId>tr-spring-boot-starter-plugin-desensitize</artifactId>
+
+
+    <description>支持脱敏操作</description>
+</project>

tr-plugins/tr-spring-boot-starter-plugin-dict/pom.xml

@@ -30,6 +30,7 @@
         <dependency>
             <groupId>cn.tr</groupId>
             <artifactId>tr-spring-boot-starter-plugin-test</artifactId>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -47,13 +48,13 @@
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>
-            <scope>test</scope>
+            <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-redis</artifactId>
-            <scope>test</scope>
+            <scope>provided</scope>
         </dependency>
     </dependencies>
 </project>

tr-plugins/tr-spring-boot-starter-plugin-dict/src/main/java/cn/tr/plugin/dict/annotation/Dict.java

@@ -24,14 +24,11 @@ import java.lang.annotation.Target;
 @JsonDeserialize(using = DictJacksonDeserializer.class)
 @JsonSerialize(using = DictJacksonSerializer.class)
 public @interface Dict {
-    @AliasFor("dictCode")
-    String value() default "";
     /**
      * 字典编码
      * @return
      */
-    @AliasFor("value")
-    String dictCode()default "";
+    String value() default "";
 
     /**
      * 字典描述

tr-plugins/tr-spring-boot-starter-plugin-dict/src/main/java/cn/tr/plugin/dict/bo/DictBO.java

@@ -42,20 +42,14 @@ public class DictBO implements Serializable {
     }
 
     public DictBO(String value, String label, String color) {
-        this.value = value;
-        this.label = label;
-        this.color = color;
-        this.sort = DictConstant.DICT_SORT_DEFAULT;
+        this(value, label,color,DictConstant.DICT_SORT_DEFAULT);
     }
 
     public DictBO(String value, String label) {
-        this.value = value;
-        this.label = label;
-        this.color = color;
-        this.sort = DictConstant.DICT_SORT_DEFAULT;
+        this(value, label,null,DictConstant.DICT_SORT_DEFAULT);
     }
 
     public DictBO() {
-        this.sort = DictConstant.DICT_SORT_DEFAULT;
+        this(null, null,null,DictConstant.DICT_SORT_DEFAULT);
     }
 }

tr-plugins/tr-spring-boot-starter-plugin-dict/src/main/java/cn/tr/plugin/dict/config/jackson/DictJacksonSerializer.java

@@ -1,5 +1,6 @@
 package cn.tr.plugin.dict.config.jackson;
 
+import cn.hutool.core.annotation.AnnotationUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.extra.spring.SpringUtil;
 import cn.tr.plugin.dict.annotation.Dict;
@@ -13,6 +14,7 @@ import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.ser.ContextualSerializer;
 import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
+import org.springframework.core.annotation.AnnotationUtils;
 
 import java.io.IOException;
 
@@ -30,7 +32,7 @@ public class DictJacksonSerializer extends JsonSerializer<String> implements Con
     @Override
     public void serialize(String value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
         DictManager dictManager = SpringUtil.getBean(DictManager.class);
-        DictBO dictBO = dictManager.lookByDictCode(dict.dictCode(), value);
+        DictBO dictBO = dictManager.lookByDictCode(dict.value(), value);
         gen.writeObject(dictBO);
     }
 
@@ -41,7 +43,7 @@ public class DictJacksonSerializer extends JsonSerializer<String> implements Con
             // 为空直接跳过
             if (ObjectUtil.equals(beanProperty.getType().getRawClass(), String.class)) {
                 // 非 String 类直接跳过
-                Dict dict = beanProperty.getAnnotation(Dict.class);
+                dict = beanProperty.getAnnotation(Dict.class);
                 if (dict == null) {
                     dict = beanProperty.getContextAnnotation(Dict.class);
                 }

tr-plugins/tr-spring-boot-starter-plugin-dict/src/test/java/cn/tr/plugin/dict/config/jackson/DictJacksonSerializerTest.java

@@ -69,7 +69,7 @@ public class DictJacksonSerializerTest extends BaseMockitoUnitTest {
     @Data
     @NoArgsConstructor
     static class User{
-        @Dict(dictCode = "gender")
+        @Dict("gender")
         String gender;
     }
 }

tr-plugins/tr-spring-boot-starter-plugin-eventbus/pom.xml

@@ -21,20 +21,32 @@
             <artifactId>tr-framework</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>cn.tr</groupId>
+            <artifactId>tr-spring-boot-starter-plugin-flink</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <dependency>
             <groupId>org.redisson</groupId>
             <artifactId>redisson-spring-boot-starter</artifactId>
-            <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>cn.tr</groupId>
             <artifactId>tr-spring-boot-starter-plugin-test</artifactId>
+            <scope>test</scope>
         </dependency>
 
-        <dependency>
-            <groupId>org.greenrobot</groupId>
-            <artifactId>eventbus-java</artifactId>
-        </dependency>
     </dependencies>
 </project>

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/TrEventBusAutoConfiguration.java

@@ -1,22 +1,64 @@
 package cn.tr.plugin.eventbus;
 
-import org.greenrobot.eventbus.EventBus;
-import org.springframework.context.annotation.Bean;
 
+import cn.tr.plugin.eventbus.config.EventBus;
+import cn.tr.plugin.eventbus.config.SubscribeListenerAnnotationBeanPostProcessor;
+import cn.tr.plugin.eventbus.config.core.AbstractEvenBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.core.EventBusDataSourceEmitter;
+import cn.tr.plugin.eventbus.config.core.LocalEventBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.core.ShareEventBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.redission.RedissionEventBus;
+import cn.tr.plugin.eventbus.config.redission.RedissionEventBusDataSourceEmitter;
+import cn.tr.plugin.eventbus.urils.EventBusStrategy;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.redisson.api.RedissonClient;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import java.util.*;
 /**
  * @ClassName : TrEventBusAutoConfiguration
  * @Description : 消息总线自动装配
  * @Author : LF
  * @Date: 2023年03月15日
  */
-
 public class TrEventBusAutoConfiguration {
+    /**
+     * 本地订阅处理
+     */
     @Bean
-    public EventBus eventBus(){
-        return EventBus.builder()
-                .logNoSubscriberMessages(false)
-                .sendNoSubscriberEvent(false)
-                .throwSubscriberException(true)
-                .build();
+    public LocalEventBusDataRichSourceFunction localEventBusDataRichSourceFunction(){
+        return new LocalEventBusDataRichSourceFunction();
+    }
+
+    /**
+     * 共享订阅处理
+     */
+    @Bean
+    public ShareEventBusDataRichSourceFunction shareEventBusDataRichSourceFunction(){
+        return new ShareEventBusDataRichSourceFunction();
+    }
+
+    @Bean
+    public SubscribeListenerAnnotationBeanPostProcessor subscribeListenerAnnotationBeanPostProcessor(EventBus eventBus, ObjectMapper objectMapper){
+        EventBusStrategy.tr.objectMapperSupplier=()->objectMapper;
+        return new SubscribeListenerAnnotationBeanPostProcessor(eventBus);
+    }
+
+    //todo 可根据配置进行配置
+    @Configuration
+    public static class RedissionEventBusAutoConfiguration{
+        @Bean
+        public EventBus redissionEventBus(StreamExecutionEnvironment ec,RedissonClient redissonClient,List<AbstractEvenBusDataRichSourceFunction> functions) throws Exception {
+            EventBusStrategy.tr.redissonClientSupplier=()->redissonClient;
+            return new RedissionEventBus(redissonClient,ec,functions);
+        }
+
+        @Bean(initMethod = "emitter")
+        public EventBusDataSourceEmitter redissionEventBusDataSourceEmitter(List<AbstractEvenBusDataRichSourceFunction> functions){
+            return new RedissionEventBusDataSourceEmitter(functions);
+        }
+
     }
 }

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/annotation/Subscribe.java

@@ -0,0 +1,26 @@
+package cn.tr.plugin.eventbus.annotation;
+
+import cn.tr.plugin.eventbus.bean.Subscription;
+import org.springframework.core.annotation.AliasFor;
+import java.lang.annotation.*;
+
+/**
+ * @ClassName : Subscribe
+ * @Description : 方法参数个数必须为0或1，方法参数类型不包含数组类型，如果必须传递数组类型，请创建一个对象进行包装
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+@Target({ ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface Subscribe {
+
+    /**
+     * 订阅主题
+     */
+    String[] value() default {};
+
+    String id() default "";
+
+    Subscription.Feature feature() default Subscription.Feature.local;
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/bean/Subscription.java

@@ -0,0 +1,141 @@
+package cn.tr.plugin.eventbus.bean;
+
+import cn.tr.plugin.eventbus.urils.TopicUtils;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.springframework.util.Assert;
+
+import java.io.Serializable;
+import java.util.*;
+import java.util.stream.Collectors;
+
+@AllArgsConstructor
+@Getter
+public class Subscription implements Serializable {
+    private static final long serialVersionUID = -6849794470754667710L;
+
+    public static final Feature[] DEFAULT_FEATURES = Subscription.Feature.values();
+
+    /**
+     * 订阅者标识
+     */
+    private final String subscriber;
+
+    /**
+     * 订阅主题,主题以/分割,如: /device/TS-01/09012/message 支持通配符 /device/**
+     */
+    private final String[] topics;
+
+    /**
+     * 订阅特性
+     */
+    private final Feature feature;
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        Subscription that = (Subscription) o;
+        return getSubscriber().equals(that.getSubscriber()) &&
+                Arrays.equals(getTopics(), that.getTopics()) &&
+                getFeature() == that.getFeature();
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hash(getSubscriber(), getFeature());
+        result = 31 * result + Arrays.hashCode(getTopics());
+        return result;
+    }
+
+    public static Subscription of(String subscriber, String topic, Feature feature) {
+        return of(subscriber,new String[]{topic},feature);
+    }
+    public static Subscription of(String subscriber, String[] topics, Feature features) {
+
+        return Subscription
+                .builder()
+                .subscriberId(subscriber)
+                .topics(topics)
+                .build();
+    }
+
+
+
+
+    @AllArgsConstructor
+    @Getter
+    public enum Feature {
+
+        //如果相同的订阅者,只有一个订阅者收到消息
+        shared("shared"),
+        //订阅本地消息
+        local("订阅本地消息");
+
+        private final String text;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        //订阅者标识
+        private String subscriber;
+
+        //订阅主题,主题以/分割,如: /device/TS-01/09012/message 支持通配符 /device/**
+        private final Set<String> topics = new HashSet<>();
+
+        //订阅特性
+        private Feature feature;
+
+
+        public Builder randomSubscriberId() {
+            return subscriberId(UUID.randomUUID().toString());
+        }
+
+        public Builder subscriberId(String id) {
+            this.subscriber = id;
+            return this;
+        }
+
+        public Builder topics(String... topics) {
+            return topics(Arrays.asList(topics));
+        }
+
+        public Builder topics(Collection<String> topics) {
+            this.topics.addAll(topics.stream()
+                                     .flatMap(topic -> TopicUtils.expand(topic).stream())
+                                     .collect(Collectors.toSet()));
+            return this;
+        }
+
+        public Builder feature(Feature feature) {
+            this.feature=feature;
+            return this;
+        }
+
+        public Builder justLocal() {
+            return feature(Feature.local);
+        }
+
+
+        public Builder local() {
+            return feature(Feature.local);
+        }
+
+        public Builder shared() {
+            return feature(Feature.shared);
+        }
+
+        public Subscription build() {
+            if (feature==null) {
+                local();
+            }
+            Assert.notEmpty(topics, "topic cannot be empty");
+//            Assert.hasText(subscriber, "subscriber cannot be empty");
+            return new Subscription(subscriber, topics.toArray(new String[0]), feature);
+        }
+
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/bean/TopicPayload.java

@@ -0,0 +1,28 @@
+package cn.tr.plugin.eventbus.bean;
+
+import cn.hutool.core.util.IdUtil;
+import cn.tr.core.utils.JsonUtils;
+import lombok.Data;
+import java.io.Serializable;
+
+/**
+ * @ClassName : TopicPayload
+ * @Description : 主题消息
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+@Data
+public class TopicPayload implements Serializable {
+    private static final long serialVersionUID = -3950884003355115963L;
+    private long id;
+    private String payload;
+    private long createTime;
+    private String topic;
+
+    public TopicPayload(Object payload, String topic) {
+        this.id = IdUtil.getSnowflakeNextId();
+        this.payload = JsonUtils.toJsonString(payload);
+        this.createTime = System.currentTimeMillis();
+        this.topic = topic;
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/EventBus.java

@@ -0,0 +1,105 @@
+package cn.tr.plugin.eventbus.config;
+import cn.hutool.core.util.ReflectUtil;
+import cn.tr.plugin.eventbus.proxy.ConsumerWrapper;
+import cn.tr.plugin.eventbus.proxy.ProxyMessageListener;
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import org.apache.commons.compress.utils.Sets;
+
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * @ClassName : EventBus
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+
+public interface EventBus {
+
+    /**
+     * 订阅主题，由代理类消费
+     * @param topic
+     * @param messageConsumer
+     * @param <T>
+     */
+    default  <T> void subscribe(Subscription topic, Consumer<String> messageConsumer){
+        subscribe(Sets.newHashSet(topic),messageConsumer);
+    };
+
+    /**
+     * 订阅主题，由代理类消费
+     * @param topic
+     * @param messageListener
+     * @param <T>
+     */
+    default  <T> void subscribe(Subscription topic, ProxyMessageListener messageListener){
+        subscribe(Sets.newHashSet(topic),messageListener);
+    };
+
+    /**
+     * 订阅主题
+     * @param topics 主题，支持正则
+     * @param messageConsumer 所订阅的主题的消费者
+     * @param <T>
+     */
+    default  <T> void subscribe(Set<Subscription> topics, Consumer<String> messageConsumer){
+        ProxyMessageListener proxyMessageListener = new ProxyMessageListener(new ConsumerWrapper(messageConsumer),
+                ReflectUtil.getMethod(ConsumerWrapper.class, "accept", String.class));
+        subscribe(topics,proxyMessageListener);
+    };
+
+    /**
+     * 订阅主题，由代理类消费
+     * @param topics
+     * @param messageListener
+     * @param <T>
+     */
+    public <T> void subscribe(Set<Subscription> topics, ProxyMessageListener messageListener);
+
+    /**
+     * 向主题中发布消息
+     * @param topics 发布主题
+     * @param message 消息
+     * @param <T>
+     * @return
+     */
+    default  <T> Long publish(Set<String> topics, T message){
+        return publish(topics,()->message);
+    };
+
+    default  <T> Long publish(String topic, T message){
+        return publish(Sets.newHashSet(topic),()->message);
+    };
+
+    default  <T> Long publish(String topic, Supplier<T> supplierMsg){
+        return publish(Sets.newHashSet(topic),supplierMsg);
+    };
+
+    /**
+     * 向主题中发布消息
+     * @param topics 发布主题
+     * @param supplierMsg 消息产生者
+     * @param <T>
+     * @return
+     */
+    default  <T> Long publish(Set<String> topics, Supplier<T> supplierMsg){
+        return topics.stream()
+                .map(topic->new TopicPayload(supplierMsg.get(),topic))
+                .map(this::publish)
+                .mapToLong(Long::intValue)
+                .sum();
+    };
+
+    /**
+     * 向主题中发布消息
+     * @param msgs 发布消息
+     * @param <T>
+     * @return
+     */
+    public <T> Long publish(TopicPayload msgs);
+
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/SubscribeListenerAnnotationBeanPostProcessor.java

@@ -0,0 +1,199 @@
+package cn.tr.plugin.eventbus.config;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.tr.core.utils.JsonUtils;
+import cn.tr.plugin.eventbus.annotation.Subscribe;
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.proxy.ProxyMessageListener;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.annotation.MergedAnnotation;
+import org.springframework.core.annotation.MergedAnnotations;
+import org.springframework.util.ReflectionUtils;
+
+import java.lang.reflect.AnnotatedElement;
+import java.util.*;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * @ClassName : SubscribeListenerAnnotationBeanPostProcessor
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+@Slf4j
+public class SubscribeListenerAnnotationBeanPostProcessor implements BeanPostProcessor {
+    private final ConcurrentMap<Class<?>, TypeMetadata> typeCache = new ConcurrentHashMap<>();
+
+    /**
+     * 订阅者集合
+     */
+    private final static Set<Subscription> subscriptionSet=new HashSet<>();
+
+    private final EventBus eventBus;
+
+    public SubscribeListenerAnnotationBeanPostProcessor(EventBus eventBus) {
+        this.eventBus = eventBus;
+    }
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        Class<?> targetClass = AopUtils.getTargetClass(bean);
+        final TypeMetadata metadata = this.typeCache.computeIfAbsent(targetClass, this::buildMetadata);
+        for (ListenerMethod lm : metadata.listenerMethods) {
+            for (Subscribe subscribe : lm.annotations) {
+                if(subscribe.value().length==0){
+                    continue;
+                }
+                processSubscribe(targetClass,subscribe, lm.method, bean, beanName);
+            }
+        }
+        return bean;
+    }
+
+    /**
+     * 处理订阅者
+     * @param objClass 对象class
+     * @param subscribe 订阅属性
+     * @param method 订阅方法
+     * @param bean 订阅bean
+     * @param beanName 订阅bean名称
+     */
+    protected void processSubscribe(Class objClass,Subscribe subscribe,Method method,Object bean,String beanName){
+        checkProxy(method, bean);
+        Optional<Class<?>> isArray = Stream.of(method.getParameterTypes()).filter(this::returnsMany).findAny();
+        if(isArray.isPresent()){
+            throw new UnsupportedOperationException("The parameter of class :"+ objClass.getName()+"method : "+method.getName()+"   cannot be an array");
+        }
+        ProxyMessageListener proxyMessageListener = new ProxyMessageListener(bean, method);
+        Class<? extends Class> aClass = objClass.getClass();
+        Class<?>[] parameterTypes = method.getParameterTypes();
+        String id =subscribe.id();
+        if(StrUtil.isEmpty(id)){
+            id= aClass.getName() + method.getName() + (parameterTypes.length == 0 ? "" : parameterTypes[0].getName());
+        }
+        Subscription subscription = Subscription.builder()
+                .feature(subscribe.feature())
+                .topics(subscribe.value())
+                .subscriberId(id)
+                .build();
+        if (subscriptionSet.contains(subscription)) {
+            log.warn("subscription : [{}] has exist ,con't repeat register", JsonUtils.toJsonString(subscription));
+        }else {
+            eventBus.subscribe(subscription,proxyMessageListener);
+            subscriptionSet.add(subscription);
+        }
+
+    }
+
+
+    /**
+     * 判断返回值类型是否是集合或者数组类型
+     * @param returnType 类型
+     * @return 是否是集合或者数组类型
+     */
+    private boolean returnsMany(Class<?> returnType) {
+        //判断返回类型是否是集合类型
+        boolean isCollection = Collection.class.isAssignableFrom(returnType);
+        //判断返回类型是否是数组类型
+        boolean isArray = returnType.isArray();
+        return isCollection || isArray;
+    }
+
+    private Method checkProxy(Method methodArg, Object bean) {
+        Method method = methodArg;
+        if (AopUtils.isJdkDynamicProxy(bean)) {
+            try {
+                // Found a @RabbitListener method on the target class for this JDK proxy ->
+                // is it also present on the proxy itself?
+                method = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
+                Class<?>[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces();
+                for (Class<?> iface : proxiedInterfaces) {
+                    try {
+                        method = iface.getMethod(method.getName(), method.getParameterTypes());
+                        break;
+                    }
+                    catch (@SuppressWarnings("unused") NoSuchMethodException noMethod) {
+                    }
+                }
+            }
+            catch (SecurityException ex) {
+                ReflectionUtils.handleReflectionException(ex);
+            }
+            catch (NoSuchMethodException ex) {
+                throw new IllegalStateException(String.format(
+                        "@Subscribe method '%s' found on bean target class '%s', " +
+                                "but not found in any interface(s) for a bean JDK proxy. Either " +
+                                "pull the method up to an interface or switch to subclass (CGLIB) " +
+                                "proxies by setting proxy-target-class/proxyTargetClass " +
+                                "attribute to 'true'", method.getName(), method.getDeclaringClass().getSimpleName()), ex);
+            }
+        }
+        return method;
+    }
+
+    private TypeMetadata buildMetadata(Class<?> targetClass){
+        final List<ListenerMethod> methods = new ArrayList<>();
+        //找到类中所有带有 @Subscribe 注解的方法
+        ReflectionUtils.doWithMethods(targetClass,method->{
+            List<Subscribe> methodSubscribes = findSubscribeAnnotation(method);
+            if (CollectionUtil.isNotEmpty(methodSubscribes)) {
+                methods.add(new ListenerMethod(method,methodSubscribes.toArray(new Subscribe[CollectionUtil.size(methodSubscribes)])));
+            }
+        }, ReflectionUtils.USER_DECLARED_METHODS);
+
+        return new TypeMetadata(methods.toArray(new ListenerMethod[methods.size()]));
+    }
+
+
+    private List<Subscribe> findSubscribeAnnotation(AnnotatedElement element){
+        return MergedAnnotations.from(element, MergedAnnotations.SearchStrategy.TYPE_HIERARCHY)
+                .stream(Subscribe.class)
+                .map(MergedAnnotation::synthesize)
+                .collect(Collectors.toList());
+    }
+
+
+    private static class TypeMetadata{
+        final ListenerMethod[] listenerMethods;
+
+        static final TypeMetadata EMPTY = new TypeMetadata();
+
+        private TypeMetadata() {
+            this.listenerMethods = new SubscribeListenerAnnotationBeanPostProcessor.ListenerMethod[0];
+        }
+
+        TypeMetadata(ListenerMethod[] methods) {
+            this.listenerMethods = methods;
+        }
+
+    }
+
+    private static class ListenerMethod {
+
+        /**
+         * 订阅方法  带有{@link Subscribe}注解的方法
+         */
+        final Method method;
+
+        /**
+         * 订阅注解 {@link Subscribe}
+         */
+        final Subscribe[] annotations;
+
+        ListenerMethod(Method method, Subscribe[] annotations) {
+            this.method = method;
+            this.annotations = annotations;
+        }
+
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/core/AbstractEvenBusDataRichSourceFunction.java

@@ -0,0 +1,141 @@
+package cn.tr.plugin.eventbus.config.core;
+
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateTime;
+import cn.hutool.core.date.DateUnit;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.IdUtil;
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import cn.tr.plugin.eventbus.urils.EventBusStrategy;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.redisson.api.RBloomFilter;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+/**
+ * @ClassName : AbstractEvenBusDataRichSourceFunction
+ * @Description : 消息总线数据流处理
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+@Slf4j
+public abstract class AbstractEvenBusDataRichSourceFunction extends RichSourceFunction<TopicPayload> {
+    private static final Map<String, RBloomFilter<Long>> bloomFilterMap=new ConcurrentHashMap<>();
+
+    private static final  ExecutorService executorService=Executors.newSingleThreadExecutor();
+
+    private static final Map<String, Consumer<TopicPayload>> consumerMap=new HashMap<>();
+    @Getter
+    private Boolean close=false;
+
+    private final String id;
+
+    public AbstractEvenBusDataRichSourceFunction() {
+        id= IdUtil.randomUUID();
+    }
+
+    static {
+        executorService.execute(()->{
+            Set<String> keySet = bloomFilterMap.keySet();
+            Date now = new Date();
+            for (String date : keySet) {
+                DateTime dateTime = DateUtil.parseDate(date);
+                if (DateUtil.between(dateTime,now, DateUnit.HOUR)>8) {
+                    RBloomFilter<Long> bloomFilter = bloomFilterMap.remove(date);
+                    if(bloomFilter!=null){
+                        try {
+                            bloomFilter.delete();
+                        }finally {
+                        }
+                    }
+                }
+            }
+        });
+    }
+
+    public abstract Subscription.Feature getFeature();
+    @Override
+    public void run(SourceContext<TopicPayload> ctx) throws Exception {
+        while (!getClose()){
+            //非关闭状态
+            consumerMap.putIfAbsent(id,msg->{
+                if(!checkMsg(msg)){
+                    return;
+                }
+                if (repeatConsumer(msg)) {
+                    //去重
+                    if(log.isDebugEnabled()){
+                        log.debug("[Pub message] : [{}] repeated",msg);
+                    }
+                    return;
+                }
+                ctx.collect(msg);
+            });
+        }
+    }
+
+    public Consumer<TopicPayload> getCtxConsumer(){
+        return consumerMap.get(id);
+    }
+    /**
+     * 检查消息是否合法
+     * @param msg
+     * @return
+     */
+    public boolean checkMsg(TopicPayload msg){
+        return true;
+    }
+
+    /**
+     * 判断当前消息是否被重复消息
+     * @param msg
+     * @return
+     */
+    public boolean repeatConsumer(TopicPayload msg){
+        RBloomFilter<Long> bloomFilter = finBloomFilter(msg);
+        return !bloomFilter.add(msg.getId());
+    }
+
+
+    private RBloomFilter<Long> finBloomFilter(TopicPayload msg){
+        //按照小时划分
+        String date = DateUtil.format(DateUtil.beginOfHour(new Date(msg.getCreateTime())), DatePattern.NORM_DATETIME_FORMAT);
+        return bloomFilterMap.compute(date, (k, bloomFilter) -> {
+            if(bloomFilter==null){
+                bloomFilter =  EventBusStrategy.tr.getRedissionClient() .getBloomFilter(date);
+                bloomFilter.tryInit(1000000,0);
+                //bitmap 7天自动过期释放内存
+                bloomFilter.expire(Duration.of(7, ChronoUnit.HOURS));
+            }
+            return bloomFilter;
+        });
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        this.close=false;
+    }
+
+    @Override
+    public void close() throws Exception {
+        this.close=true;
+    }
+
+    @Override
+    public void cancel() {
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/core/EventBusDataSourceEmitter.java

@@ -0,0 +1,27 @@
+package cn.tr.plugin.eventbus.config.core;
+
+import cn.hutool.core.collection.CollectionUtil;
+import lombok.Getter;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+/**
+ * @ClassName : EventBusDataSourceEmitter
+ * @Description : 消息总线消息发射器
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+
+public abstract class EventBusDataSourceEmitter implements Consumer<List<AbstractEvenBusDataRichSourceFunction>> {
+    @Getter
+    private final List<AbstractEvenBusDataRichSourceFunction> richSourceFunctions;
+
+    public EventBusDataSourceEmitter(List<AbstractEvenBusDataRichSourceFunction> richSourceFunctions) {
+        this.richSourceFunctions = richSourceFunctions;
+    }
+
+    public void emitter(){
+        accept(richSourceFunctions);
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/core/LocalEventBusDataRichSourceFunction.java

@@ -0,0 +1,26 @@
+package cn.tr.plugin.eventbus.config.core;
+
+
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+
+/**
+ * @ClassName : LocalEventBusDataRichSourceFunction
+ * @Description : 本地消息总线 {@link cn.tr.plugin.eventbus.bean.Subscription.Feature#local}
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+
+public class LocalEventBusDataRichSourceFunction extends AbstractEvenBusDataRichSourceFunction {
+
+
+    @Override
+    public Subscription.Feature getFeature() {
+        return Subscription.Feature.local;
+    }
+
+    @Override
+    public boolean checkMsg(TopicPayload msg) {
+        return true;
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/core/ShareEventBusDataRichSourceFunction.java

@@ -0,0 +1,19 @@
+package cn.tr.plugin.eventbus.config.core;
+
+
+import cn.tr.plugin.eventbus.bean.Subscription;
+
+/**
+ * @ClassName : ShareEventBusDataRichSourceFunction
+ * @Description : 共享消息总线 {@link cn.tr.plugin.eventbus.bean.Subscription.Feature#shared}
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+
+public class ShareEventBusDataRichSourceFunction extends AbstractEvenBusDataRichSourceFunction {
+
+    @Override
+    public Subscription.Feature getFeature() {
+        return Subscription.Feature.shared;
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/redission/RedissionEventBus.java

@@ -0,0 +1,102 @@
+package cn.tr.plugin.eventbus.config.redission;
+
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.tr.plugin.eventbus.proxy.ProxyMessageListener;
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import cn.tr.plugin.eventbus.config.EventBus;
+import cn.tr.plugin.eventbus.config.core.AbstractEvenBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.constant.EventBusConstant;
+import cn.tr.plugin.eventbus.urils.TopicUtils;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.redisson.api.RTopic;
+import org.redisson.api.RedissonClient;
+import java.util.*;
+
+/**
+ * @ClassName : RedissonEventBus
+ * @Description : 订阅消息占用一个主线程进行通信，消费线程多个
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+@Slf4j
+public class RedissionEventBus implements EventBus {
+    private static final long serialVersionUID = 7277245154307718390L;
+
+    private final transient RTopic publishTopic;
+
+    private static final Map<Subscription.Feature,Map<String, List<ProxyMessageListener>>> subscriptionMap = new HashMap<>();
+
+    public RedissionEventBus(RedissonClient redissonClient, StreamExecutionEnvironment ec, List<AbstractEvenBusDataRichSourceFunction> dataSources) throws Exception {
+        this.publishTopic = redissonClient.getTopic(EventBusConstant.EVENT_BUS_TOPIC);
+        dataSources.forEach(dataSource->{
+            ec.addSource(dataSource)
+                    .filter(Objects::nonNull)
+                    .keyBy(TopicPayload::getTopic)
+                    .addSink(new RedissionEventBusSink(dataSource.getFeature()));
+        });
+        ec.executeAsync("tr-eventbus");
+    }
+
+    @Override
+    public <T> void subscribe(Set<Subscription> subscriptions, ProxyMessageListener messageListener) {
+        subscriptions.forEach(subscription -> {
+            String[] topics =subscription.getTopics();
+            Subscription.Feature feature = subscription.getFeature();
+            if(topics.length==0||feature==null){
+                return;
+            }
+            //保存订阅消息消费者
+            subscriptionMap.compute(feature,(k,topicMap)->{
+                if(topicMap==null){
+                    topicMap=new HashMap<>();
+                }
+                for (String topic : topics) {
+                    topicMap.compute(topic,(tp,listeners)->{
+                        if(listeners==null){
+                            listeners=new ArrayList<>();
+                        }
+                        listeners.add(messageListener);
+                        return listeners;
+                    });
+                }
+                return topicMap;
+            });
+        });
+    }
+
+    @Override
+    public <T> Long publish(TopicPayload msg) {
+        return publishTopic.publish(msg);
+    }
+
+
+    @AllArgsConstructor
+    static class RedissionEventBusSink implements SinkFunction<TopicPayload> {
+        private final Subscription.Feature feature;
+
+        @Override
+        public void invoke(TopicPayload payload, Context context) throws Exception {
+            String payloadTopic = payload.getTopic();
+            Map<String, List<ProxyMessageListener>> topicMap = subscriptionMap.get(feature);
+            if(CollectionUtil.isEmpty(topicMap)){
+                return;
+            }
+            topicMap.forEach((k,listeners)->{
+                if(TopicUtils.match(k,payloadTopic)){
+                    listeners.parallelStream().forEach(ln -> {
+                        try {
+                            ln.onMessage(payload);
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                        }
+                    });
+                }
+            });
+        }
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/redission/RedissionEventBusDataSourceEmitter.java

@@ -0,0 +1,45 @@
+package cn.tr.plugin.eventbus.config.redission;
+
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import cn.tr.plugin.eventbus.config.core.AbstractEvenBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.core.EventBusDataSourceEmitter;
+import cn.tr.plugin.eventbus.constant.EventBusConstant;
+import cn.tr.plugin.eventbus.urils.EventBusStrategy;
+import org.redisson.api.RTopic;
+import java.util.function.Consumer;
+import java.util.*;
+/**
+ * @ClassName : RedissionEventBusDataSourceEmitter
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+
+public class RedissionEventBusDataSourceEmitter extends EventBusDataSourceEmitter {
+    private static RTopic rTopic=null;
+    public RedissionEventBusDataSourceEmitter(List<AbstractEvenBusDataRichSourceFunction> richSourceFunctions) {
+        super(richSourceFunctions);
+    }
+
+    @Override
+    public void accept(List<AbstractEvenBusDataRichSourceFunction> richSourceFunction) {
+        if(rTopic==null){
+            synchronized (RedissionEventBusDataSourceEmitter.class){
+                if(rTopic==null){
+                    rTopic=EventBusStrategy.tr
+                            .getRedissionClient()
+                            .getTopic(EventBusConstant.EVENT_BUS_TOPIC);
+                    rTopic.addListener(TopicPayload.class,(channel,msg)->{
+                        for (AbstractEvenBusDataRichSourceFunction sourceFunction : richSourceFunction) {
+                            Consumer<TopicPayload> ctxConsumer = sourceFunction.getCtxConsumer();
+                            if(ctxConsumer!=null){
+                                ctxConsumer.accept(msg);
+                            }
+                        }
+                    });
+                }
+            }
+        }
+
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/config/register/SubscribeRegisterCenter.java

@@ -0,0 +1,22 @@
+package cn.tr.plugin.eventbus.config.register;
+
+import org.redisson.api.RedissonClient;
+
+/**
+ * @ClassName : SubscribeRegisterCenter
+ * @Description : 订阅者注册中心
+ * 保存各个节点的订阅者信息,当 {@link cn.tr.plugin.eventbus.bean.Subscription.Feature } = {@link cn.tr.plugin.eventbus.bean.Subscription.Feature#shared}时,
+ * 若有消息发送到来时, bitmap(id) 是否存在,
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+
+public class SubscribeRegisterCenter {
+    private RedissonClient redissonClient;
+
+
+
+    public void SubscribeRegisterCenter(){
+        boolean set = redissonClient.getBitSet("").set(123, true);
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/constant/EventBusConstant.java

@@ -0,0 +1,15 @@
+package cn.tr.plugin.eventbus.constant;
+
+/**
+ * @Interface : EventBusConstant
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+
+public interface EventBusConstant {
+    String EVENT_BUS_TOPIC ="tr-eventbus";
+
+
+    String SUBSCRIBE_REGISTER_CENTER="eventbus-subscribe";
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/proxy/ConsumerWrapper.java

@@ -0,0 +1,21 @@
+package cn.tr.plugin.eventbus.proxy;
+
+import lombok.AllArgsConstructor;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/**
+ * @ClassName : ConsumerWrapper
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月18日
+ */
+@AllArgsConstructor
+public class ConsumerWrapper implements Serializable {
+    private final Consumer<String> consumer;
+
+    public void accept(String s) {
+        consumer.accept(s);
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/proxy/ProxyMessageListener.java

@@ -0,0 +1,132 @@
+package cn.tr.plugin.eventbus.proxy;
+
+import cn.hutool.core.util.ModifierUtil;
+import cn.tr.core.utils.Proxy;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import cn.tr.plugin.eventbus.urils.EventBusStrategy;
+import com.fasterxml.jackson.databind.ObjectReader;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.util.SerializationUtils;
+
+import java.lang.reflect.Method;
+import java.util.*;
+import java.util.function.BiFunction;
+
+/**
+ * @ClassName : ProxyMessageListener
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+@Slf4j
+public class ProxyMessageListener {
+    private final Class<?> paramType;
+    private final Object target;
+    private final Method method;
+    private final BiFunction<Object, Object, Object> proxy;
+    private ObjectReader objectReader=null;
+
+    public ProxyMessageListener(Object target,Method method){
+        this.target = target;
+        this.method = method;
+        Class<?>[] parameterTypes = method.getParameterTypes();
+        if (parameterTypes.length > 1) {
+            throw new UnsupportedOperationException("unsupported method [" + method + "] parameter");
+        }
+        if (parameterTypes.length == 1) {
+            paramType = parameterTypes[0];
+            objectReader =  EventBusStrategy.tr.getObjectMapper().readerFor(paramType);
+        } else {
+            paramType = Void.class;
+        }
+        checkModifier();
+        Class<?> targetType = AopUtils.getTargetClass(target);
+
+        StringJoiner code = new StringJoiner("\n");
+
+        code.add("public Object apply(Object target,Object param){");
+        code.add(targetType.getName() + " _target = (" + targetType.getName() + ")target;");
+
+        String invokeCode;
+        if (paramType != Void.class) {
+            code.add(paramType.getName() + " _param = (" + paramType.getName() + ")param;");
+            invokeCode = " _target." + method.getName() + "(_param);";
+        } else {
+            invokeCode = " _target." + method.getName() + "();";
+        }
+        if (method.getReturnType() != Void.TYPE) {
+            code.add("return " + invokeCode);
+        } else {
+            code.add(invokeCode)
+                    .add("return null;");
+        }
+        code.add("}");
+
+        this.proxy = Proxy.create(BiFunction.class)
+                .addMethod(code.toString())
+                .newInstance();
+    }
+
+    public Object onMessage(TopicPayload payload) {
+        try {
+            String message = payload.getPayload();
+            boolean paramVoid = paramType == Void.class;
+            Object result = null;
+            if(paramType==String.class||paramType==Object.class){
+                result=proxy.apply(target,paramVoid?null:message);
+            } else if(paramType==TopicPayload.class){
+                result=proxy.apply(target,paramVoid?null:payload);
+            }
+            else {
+                result=proxy.apply(target,paramVoid?null:objectReader.readValue(message));
+            }
+            return paramVoid?null:result;
+        } catch (Exception e){
+            log.error("invoke event listener [{}] error", toString(), e);
+        }
+        return null;
+    }
+
+
+    private void checkModifier(){
+        //类修饰符必须为public
+        if(target.getClass().getName().contains("$Lambda")){
+            return;
+        }
+        if(!ModifierUtil.isPublic(target.getClass())){
+            throw new IllegalStateException(target.getClass().getName()+"'s modifer \n required : PUBLIC; actually :"+ getModifierType(target.getClass().getModifiers()));
+        }
+
+        //方法修饰符必须为public
+        if(!ModifierUtil.isPublic(method)){
+            throw new IllegalStateException(target.getClass().getName()+"."+method.getName()+"'s modifer \n required : PUBLIC ;actually :"+ getModifierType(method.getModifiers()));
+        }
+    }
+
+
+    private String getModifierType(int modifier){
+        ModifierUtil.ModifierType[] values = ModifierUtil.ModifierType.values();
+        for (ModifierUtil.ModifierType value : values) {
+            if(value.getValue()==modifier){
+                return value.name();
+            }
+        }
+        return "UNKOWN";
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ProxyMessageListener that = (ProxyMessageListener) o;
+        return Objects.equals(paramType, that.paramType) &&
+                Objects.equals(target, that.target) &&
+                Objects.equals(method, that.method);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(paramType, target, method);
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/redisson/RedissonEventBus.java

@@ -1,44 +0,0 @@
-package cn.tr.plugin.eventbus.redisson;
-
-import org.greenrobot.eventbus.EventBus;
-import org.redisson.api.RTopic;
-import org.redisson.api.RedissonClient;
-import org.redisson.api.listener.MessageListener;
-
-/**
- * @ClassName : RedissonEventBus
- * @Description : 通过redisson实现分布式消息总线
- * @Author : LF
- * @Date: 2023年03月16日
- */
-
-public class RedissonEventBus extends EventBus {
-    /**
-     * redisson消息总线主题
-     */
-    public static final String EVENT_TOPIC="redissonEventBus";
-
-    private final RTopic topic;
-
-    private final EventBus eventBus;
-    public RedissonEventBus(RedissonClient redissonClient) {
-        topic = redissonClient.getTopic(EVENT_TOPIC);
-        eventBus=EventBus
-                .builder()
-                .logNoSubscriberMessages(false)
-                .sendNoSubscriberEvent(false)
-                .build();
-
-        topic.addListener(Object.class, new MessageListener<Object>() {
-            @Override
-            public void onMessage(CharSequence channel, Object o) {
-                eventBus.post(o);
-            }
-        });
-    }
-
-    @Override
-    public void post(Object event) {
-        topic.publish(event);
-    }
-}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/urils/EventBusStrategy.java

@@ -0,0 +1,30 @@
+package cn.tr.plugin.eventbus.urils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.redisson.api.RedissonClient;
+
+import java.util.function.Supplier;
+
+/**
+ * @ClassName : EvenBusUtils
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+
+public class EventBusStrategy {
+    private EventBusStrategy() {
+    }
+
+    public Supplier<RedissonClient> redissonClientSupplier=()->null;
+    public Supplier<ObjectMapper> objectMapperSupplier=()->new ObjectMapper();
+    public static EventBusStrategy tr =new EventBusStrategy();
+
+    public RedissonClient getRedissionClient(){
+        return redissonClientSupplier.get();
+    }
+
+    public ObjectMapper getObjectMapper(){
+        return objectMapperSupplier.get();
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/java/cn/tr/plugin/eventbus/urils/TopicUtils.java

@@ -0,0 +1,258 @@
+package cn.tr.plugin.eventbus.urils;
+
+import org.springframework.util.AntPathMatcher;
+import org.springframework.util.PathMatcher;
+
+import java.util.*;
+
+public class TopicUtils {
+
+    private final static PathMatcher pathMatcher = new AntPathMatcher();
+
+    /**
+     * 匹配topic
+     *
+     * <pre>
+     *     match("/test/*","/test/1"); // true
+     *     match("/test/*","/test/1/2"); // false
+     *     match("/test/**","/test/1/2"); // true
+     * </pre>
+     *
+     * @param pattern 匹配模版
+     * @param topic   要匹配的topic
+     * @return 是否匹配
+     */
+    public static boolean match(String pattern, String topic) {
+
+        if (pattern.equals(topic)) {
+            return true;
+        }
+
+        if (!pattern.contains("*")
+                && !pattern.contains("#") && !pattern.contains("+")
+                && !pattern.contains("{")) {
+            return false;
+        }
+
+        return pathMatcher.match(pattern.replace("#", "**").replace("+", "*"), topic);
+
+    }
+
+    /**
+     * 根据模版从url上提取变量,如果提取出错则返回空Map
+     *
+     * <pre>
+     *   getPathVariables("/device/{productId}","/device/test123");
+     *   => {"productId","test1234"}
+     * </pre>
+     *
+     * @param template 模版
+     * @param topic    要提取的topic
+     * @return 提取结果
+     */
+    public static Map<String, String> getPathVariables(String template, String topic) {
+        try {
+            return pathMatcher.extractUriTemplateVariables(template, topic);
+        } catch (Exception e) {
+            return Collections.emptyMap();
+        }
+    }
+
+    /**
+     * 分隔topic
+     *
+     * @param topic topic
+     * @return 分隔结果
+     */
+    public static String[] split(String topic) {
+        return topic.split("/");
+    }
+
+    private static boolean matchStrings(String str, String pattern) {
+        return str.equals(pattern)
+                || "*".equals(pattern)
+                || "*".equals(str);
+    }
+
+    public static boolean match(String[] pattern, String[] topicParts) {
+        if (pattern.length == 0 && topicParts.length == 0) {
+            return true;
+        }
+        int pattIdxStart = 0;
+        int pattIdxEnd = pattern.length - 1;
+        int pathIdxStart = 0;
+        int pathIdxEnd = topicParts.length - 1;
+        while (pattIdxStart <= pattIdxEnd && pathIdxStart <= pathIdxEnd) {
+            String pattDir = pattern[pattIdxStart];
+            //匹配多层
+            if ("**".equals(pattDir)) {
+                break;
+            }
+            if (!matchStrings(pattDir, topicParts[pathIdxStart])) {
+                return false;
+            }
+            pattIdxStart++;
+            pathIdxStart++;
+        }
+        if (pathIdxStart > pathIdxEnd) {
+            if (pattIdxStart > pattIdxEnd) {
+                return (pattern[pattern.length - 1].equals("/") == topicParts[topicParts.length - 1].equals("/"));
+            }
+
+            if (pattIdxStart == pattIdxEnd && pattern[pattIdxStart].equals("*") && topicParts[topicParts.length - 1].equals("/")) {
+                return true;
+            }
+            for (int i = pattIdxStart; i <= pattIdxEnd; i++) {
+                if (!pattern[i].equals("**")) {
+                    return false;
+                }
+            }
+            return true;
+        } else if (pattIdxStart > pattIdxEnd) {
+            // String not exhausted, but pattern is. Failure.
+            return false;
+        } else if ("**".equals(topicParts[pattIdxStart])) {
+            // Path start definitely matches due to "**" part in pattern.
+            return true;
+        }
+        // up to last '**'
+        while (pattIdxStart <= pattIdxEnd && pathIdxStart <= pathIdxEnd) {
+            String pattDir = pattern[pattIdxEnd];
+            if (pattDir.equals("**")) {
+                break;
+            }
+            if (!matchStrings(pattDir, topicParts[pathIdxEnd])) {
+                return false;
+            }
+            pattIdxEnd--;
+            pathIdxEnd--;
+        }
+        if (pathIdxStart > pathIdxEnd) {
+            // String is exhausted
+            for (int i = pattIdxStart; i <= pattIdxEnd; i++) {
+                if (!pattern[i].equals("**")) {
+                    return false;
+                }
+            }
+            return true;
+        }
+
+        while (pattIdxStart != pattIdxEnd && pathIdxStart <= pathIdxEnd) {
+            int patIdxTmp = -1;
+            for (int i = pattIdxStart + 1; i <= pattIdxEnd; i++) {
+                if (pattern[i].equals("**")) {
+                    patIdxTmp = i;
+                    break;
+                }
+            }
+            if (patIdxTmp == pattIdxStart + 1) {
+                // '**/**' situation, so skip one
+                pattIdxStart++;
+                continue;
+            }
+            // Find the pattern between padIdxStart & padIdxTmp in str between
+            // strIdxStart & strIdxEnd
+            int patLength = (patIdxTmp - pattIdxStart - 1);
+            int strLength = (pathIdxEnd - pathIdxStart + 1);
+            int foundIdx = -1;
+
+            strLoop:
+            for (int i = 0; i <= strLength - patLength; i++) {
+                for (int j = 0; j < patLength; j++) {
+                    String subPat = pattern[pattIdxStart + j + 1];
+                    String subStr = topicParts[pathIdxStart + i + j];
+                    if (!matchStrings(subPat, subStr)) {
+                        continue strLoop;
+                    }
+                }
+                foundIdx = pathIdxStart + i;
+                break;
+            }
+
+            if (foundIdx == -1) {
+                return false;
+            }
+
+            pattIdxStart = patIdxTmp;
+            pathIdxStart = foundIdx + patLength;
+        }
+
+        for (int i = pattIdxStart; i <= pattIdxEnd; i++) {
+            if (!pattern[i].equals("**")) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * 展开topic
+     * <p>
+     * before:
+     * <pre>
+     *      /device/a,b,v/*
+     *  </pre>
+     * after:
+     * <pre>
+     *     /device/a/*
+     *     /device/b/*
+     *     /device/v/*
+     * </pre>
+     *
+     * before:
+     * <pre>
+     *     /device/{id}
+     * </pre>
+     * after:
+     * <pre>
+     *    /device/*
+     * </pre>
+     *
+     * @param topic topic
+     * @return 展开的topic集合
+     */
+    public static List<String> expand(String topic) {
+        if (!topic.contains(",")&&!topic.contains("{")) {
+            return Collections.singletonList(topic);
+        }
+        if (topic.startsWith("/")) {
+            topic = topic.substring(1);
+        }
+        String[] parts = topic.split("/", 2);
+
+        String first = parts[0];
+        List<String> expands = new ArrayList<>();
+
+        if (parts.length == 1) {
+            for (String split : first.split(",")) {
+                if (split.startsWith("{") && split.endsWith("}")) {
+                    split = "*";
+                }
+                expands.add("/" + split);
+            }
+            return expands;
+        }
+
+        List<String> nextTopics = expand(parts[1]);
+
+        for (String split : first.split(",")) {
+            if (split.startsWith("{") && split.endsWith("}")) {
+                split = "*";
+            }
+            for (String nextTopic : nextTopics) {
+                StringJoiner joiner = new StringJoiner("");
+                joiner.add("/");
+                joiner.add(split);
+                if (!nextTopic.startsWith("/")) {
+                    joiner.add("/");
+                }
+                joiner.add(nextTopic);
+                expands.add(joiner.toString());
+            }
+
+        }
+
+        return expands;
+    }
+
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/resources/application-unit-test → tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/main/resources/application-unit-test.yml

@@ -1,4 +1,15 @@
 spring:
+  rabbitmq:
+    host: 192.168.100.32
+    port: 5672
+    username: guest
+    password: guest
+    virtual-host: netpump
+    listener:
+      simple:
+        acknowledge-mode: none
+    #    publisher-confirms: true
+    publisher-returns: true
   redis:
     # 地址
     host: 192.168.100.32

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/test/java/cn/tr/plugin/eventbus/EventBusTest.java

@@ -1,41 +0,0 @@
-package cn.tr.plugin.eventbus;
-
-import cn.tr.plugin.test.ut.BaseMockitoUnitTest;
-import cn.tr.plugin.test.ut.BaseRedisUnitTest;
-import org.greenrobot.eventbus.EventBus;
-import org.greenrobot.eventbus.Subscribe;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.redisson.api.RedissonClient;
-
-/**
- * @ClassName : EventBusTest
- * @Description :
- * @Author : LF
- * @Date: 2023年03月15日
- */
-
-public class EventBusTest extends BaseRedisUnitTest {
-    private EventBus eventBus;
-
-    @BeforeEach
-    public void init(){
-        eventBus=EventBus.getDefault();
-        eventBus.register(new StringListener());
-    }
-
-    @Test
-    public void pub(){
-        for (int i = 0; i < 100; i++) {
-            eventBus.post(String.valueOf(i));
-        }
-    }
-
-
-    public static class StringListener{
-        @Subscribe
-        public void sub(String str){
-            System.out.println("订阅消息:"+str);
-        }
-    }
-}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/test/java/cn/tr/plugin/eventbus/RedissonEventBusTest.java

@@ -1,36 +0,0 @@
-package cn.tr.plugin.eventbus;
-
-import cn.tr.plugin.eventbus.redisson.RedissonEventBus;
-import cn.tr.plugin.test.ut.BaseRedisUnitTest;
-import org.greenrobot.eventbus.EventBus;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.redisson.api.RedissonClient;
-import org.springframework.beans.factory.annotation.Autowired;
-
-/**
- * @ClassName : RedissonEventBusTest
- * @Description :
- * @Author : LF
- * @Date: 2023年03月16日
- */
-
-public class RedissonEventBusTest extends BaseRedisUnitTest {
-    private EventBus eventBus;
-    @Autowired
-    private RedissonClient redissonClient;
-
-
-    @BeforeEach
-    public void init(){
-        eventBus=new RedissonEventBus(redissonClient);
-        eventBus.register(new EventBusTest.StringListener());
-    }
-
-    @Test
-    public void pub(){
-        for (int i = 0; i < 100; i++) {
-            eventBus.post(String.valueOf(i));
-        }
-    }
-}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/test/java/cn/tr/plugin/eventbus/RedissonTest.java

@@ -0,0 +1,73 @@
+package cn.tr.plugin.eventbus;
+
+import cn.tr.plugin.eventbus.bean.Subscription;
+import cn.tr.plugin.eventbus.config.core.AbstractEvenBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.core.LocalEventBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.core.ShareEventBusDataRichSourceFunction;
+import cn.tr.plugin.eventbus.config.redission.RedissionEventBus;
+import cn.tr.plugin.eventbus.config.redission.RedissionEventBusDataSourceEmitter;
+import cn.tr.plugin.eventbus.urils.EventBusStrategy;
+import cn.tr.plugin.test.ut.BaseRedisUnitTest;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @ClassName : RedissonTest
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月19日
+ */
+@Slf4j
+public class RedissonTest extends BaseRedisUnitTest {
+    @Autowired
+    private RedissonClient redissonClient;
+    private RedissionEventBus eventBus;
+    private StreamExecutionEnvironment ec=StreamExecutionEnvironment.getExecutionEnvironment();
+
+    @BeforeEach
+    public void setIp() throws Exception {
+        EventBusStrategy.tr.objectMapperSupplier=()->new ObjectMapper();
+        EventBusStrategy.tr.redissonClientSupplier=()->redissonClient;
+        List<AbstractEvenBusDataRichSourceFunction> sources = Arrays.asList(new LocalEventBusDataRichSourceFunction(), new ShareEventBusDataRichSourceFunction());
+        RedissionEventBusDataSourceEmitter emitter = new RedissionEventBusDataSourceEmitter(sources);
+        emitter.emitter();
+        eventBus=new RedissionEventBus(redissonClient,ec, sources);
+
+
+
+    }
+    @Test
+    public void pubAndSubJustOne() throws InterruptedException {
+        AtomicInteger atomicInteger=new AtomicInteger(0);
+
+        eventBus.subscribe(Subscription.builder().topics("topic*","tes123").local().build(), msg->{
+            log.error(">>>>>>>>>>>>>>第次接收到消息:"+msg+"<<<<<<<<<<<<<<<<<<<<");
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            atomicInteger.incrementAndGet();
+        });
+
+        Thread.sleep(1000);
+        for (int i = 0; i < 10; i++) {
+            for (int j = 0; j < 10; j++) {
+                eventBus.publish("topic"+i,"主题topic:"+i+",内容:"+j);
+            }
+        }
+        while (true){
+
+        }
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-eventbus/src/test/java/cn/tr/plugin/eventbus/SpringBootSubPubTest.java

@@ -0,0 +1,78 @@
+package cn.tr.plugin.eventbus;
+
+import cn.tr.plugin.eventbus.annotation.Subscribe;
+import cn.tr.plugin.eventbus.bean.TopicPayload;
+import cn.tr.plugin.eventbus.config.EventBus;
+import cn.tr.plugin.eventbus.config.SubscribeListenerAnnotationBeanPostProcessor;
+import cn.tr.plugin.flink.TrFlinkAutoConfiguration;
+import cn.tr.plugin.test.ut.BaseRedisUnitTest;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Import;
+
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @ClassName : SpringBootSubPubTest
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月19日
+ */
+@Import({TrEventBusAutoConfiguration.class,TrEventBusAutoConfiguration.RedissionEventBusAutoConfiguration.class, TrFlinkAutoConfiguration.class, SpringBootSubPubTest.Sub.class, ObjectMapper.class})
+public class SpringBootSubPubTest extends BaseRedisUnitTest {
+
+    @Autowired
+    private EventBus eventBus;
+
+    @Autowired
+    SubscribeListenerAnnotationBeanPostProcessor beanPostProcessor;
+    @BeforeEach
+    public void setup(){
+        beanPostProcessor.postProcessAfterInitialization(new Sub(),"123");
+    }
+
+
+    @Test
+    public void pub(){
+        for (int i = 0; i < 10; i++) {
+            eventBus.publish("test",User.of("123","测试"));
+        }
+        while (true){
+
+        }
+    }
+
+    @Slf4j
+    public static class Sub{
+        private static AtomicInteger atomicInteger=new AtomicInteger(0);
+        @Subscribe("test")
+        public void sub(TopicPayload user){
+            int i = atomicInteger.incrementAndGet();
+            log.info("第："+i+"次接到数据:"+user);
+        }
+    }
+
+
+    @AllArgsConstructor(staticName = "of")
+    @NoArgsConstructor
+    @Data
+    static class User implements Serializable {
+        private String id;
+        private String name;
+
+        @Override
+        public String toString() {
+            return "User{" +
+                    "id='" + id + '\'' +
+                    ", name='" + name + '\'' +
+                    '}';
+        }
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-file/pom.xml

@@ -25,6 +25,7 @@
         <dependency>
             <groupId>cn.tr</groupId>
             <artifactId>tr-spring-boot-starter-plugin-test</artifactId>
+            <scope>test</scope>
         </dependency>
 
         <!--ali OSS存储-->

tr-plugins/tr-spring-boot-starter-plugin-flink/pom.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>tr-plugins</artifactId>
+        <groupId>cn.tr</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>tr-spring-boot-starter-plugin-flink</artifactId>
+    <version>${revision}</version>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>cn.tr</groupId>
+            <artifactId>tr-framework</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+        </dependency>
+
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+        </dependency>
+
+
+        <dependency>
+            <groupId>cn.tr</groupId>
+            <artifactId>tr-spring-boot-starter-plugin-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

tr-plugins/tr-spring-boot-starter-plugin-flink/src/main/java/cn/tr/plugin/flink/TrFlinkAutoConfiguration.java

@@ -0,0 +1,19 @@
+package cn.tr.plugin.flink;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * @ClassName : TrFlinkAutoConfiguration
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+public class TrFlinkAutoConfiguration {
+    @Bean
+    public StreamExecutionEnvironment executionEnvironmen(){
+        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        return executionEnvironment;
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-flink/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+cn.tr.plugin.flink.TrFlinkAutoConfiguration

tr-plugins/tr-spring-boot-starter-plugin-flink/src/test/java/cn/tr/plugin/fink/FlinkTest.java

@@ -0,0 +1,46 @@
+package cn.tr.plugin.fink;
+
+import cn.tr.plugin.test.ut.BaseMockitoUnitTest;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+/**
+ * @ClassName : FlinkTest
+ * @Description :
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+
+public class FlinkTest extends BaseMockitoUnitTest {
+    @Test
+    public void test() throws Exception {
+        StreamExecutionEnvironment ec = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        DataStreamSource<Object> streamSource = ec.addSource(new SourceFunction<Object>() {
+            @Override
+            public void run(SourceContext<Object> ctx) throws Exception {
+                System.out.println("发送时间"+System.currentTimeMillis());
+                ctx.collect(Arrays.asList("1","2","3"));
+            }
+
+            @Override
+            public void cancel() {
+
+            }
+        });
+        streamSource.addSink(new SinkFunction<Object>() {
+            @Override
+            public void invoke(Object value, Context context) throws Exception {
+                System.out.println(value);
+                System.out.println("接收时间"+System.currentTimeMillis());
+            }
+        });
+
+        ec.execute();
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-mq/pom.xml

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>tr-plugins</artifactId>
+        <groupId>cn.tr</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>tr-spring-boot-starter-plugin-mq</artifactId>
+    <version>${revision}</version>
+    <description>消息队列</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+    </dependencies>
+</project>

tr-plugins/tr-spring-boot-starter-plugin-test/pom.xml

@@ -10,7 +10,6 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>tr-spring-boot-starter-plugin-test</artifactId>
-
     <version>${revision}</version>
     <packaging>jar</packaging>
 
@@ -42,6 +41,12 @@
             <artifactId>podam</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>

tr-plugins/tr-spring-boot-starter-plugin-test/src/main/java/cn/tr/plugin/test/ut/BaseRabbitMqUnitTest.java

@@ -0,0 +1,22 @@
+package cn.tr.plugin.test.ut;
+
+import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
+
+/**
+ * @ClassName : BaseRabbitMqUnitTest
+ * @Description  : rabbitmq测试
+ * @Author : LF
+ * @Date: 2023年03月17日
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = BaseRabbitMqUnitTest.Application.class)
+@ActiveProfiles("unit-test") // 设置使用 application-unit-test 配置文件
+public class BaseRabbitMqUnitTest {
+    @Import({
+            RabbitAutoConfiguration.class, // Spring Redis 自动配置类
+    })
+    public static class Application {
+    }
+}

tr-plugins/tr-spring-boot-starter-plugin-web/src/main/java/cn/tr/plugin/web/config/handler/GlobalExceptionHandler.java

@@ -29,8 +29,6 @@ import javax.validation.ConstraintViolationException;
 @Slf4j
 public class GlobalExceptionHandler {
 
-//    private final ApiErrorLogFrameworkService apiErrorLogFrameworkService;
-
     {
         //处理 SpringMVC 请求参数缺  例如说，接口上设置了 @RequestParam("xx") 参数，结果并未传递 xx 参数
         ExceptionStrategy.tr.registerThrowableHandler(MissingServletRequestParameterException.class,((request, ex) ->{
@@ -113,35 +111,5 @@ public class GlobalExceptionHandler {
 //            log.error("[createExceptionLog][url({}) log({}) 发生异常]", req.getRequestURI(),  JsonUtils.toJsonString(errorLog), th);
 //        }
 //    }
-//
-//    private void initExceptionLog(ApiErrorLog errorLog, HttpServletRequest request, Throwable e) {
-//        // 处理用户信息
-//        errorLog.setUserId(WebFrameworkUtils.getLoginUserId(request));
-//        errorLog.setUserType(WebFrameworkUtils.getLoginUserType(request));
-//        // 设置异常字段
-//        errorLog.setExceptionName(e.getClass().getName());
-//        errorLog.setExceptionMessage(ExceptionUtil.getMessage(e));
-//        errorLog.setExceptionRootCauseMessage(ExceptionUtil.getRootCauseMessage(e));
-//        errorLog.setExceptionStackTrace(ExceptionUtil.stacktraceToString(e));
-//        StackTraceElement[] stackTraceElements = e.getStackTrace();
-//        Assert.notEmpty(stackTraceElements, "异常 stackTraceElements 不能为空");
-//        StackTraceElement stackTraceElement = stackTraceElements[0];
-//        errorLog.setExceptionClassName(stackTraceElement.getClassName());
-//        errorLog.setExceptionFileName(stackTraceElement.getFileName());
-//        errorLog.setExceptionMethodName(stackTraceElement.getMethodName());
-//        errorLog.setExceptionLineNumber(stackTraceElement.getLineNumber());
-//        // 设置其它字段
-//        errorLog.setTraceId(TracerUtils.getTraceId());
-//        errorLog.setApplicationName(applicationName);
-//        errorLog.setRequestUrl(request.getRequestURI());
-//        Map<String, Object> requestParams = MapUtil.<String, Object>builder()
-//                .put("query", ServletUtil.getParamMap(request))
-//                .put("body", ServletUtil.getBody(request)).build();
-//        errorLog.setRequestParams(JsonUtils.toJsonString(requestParams));
-//        errorLog.setRequestMethod(request.getMethod());
-//        errorLog.setUserAgent(ServletUtils.getUserAgent(request));
-//        errorLog.setUserIp(ServletUtil.getClientIP(request));
-//        errorLog.setExceptionTime(LocalDateTime.now());
-//    }
 
 }

tr-plugins/tr-spring-boot-starter-plugin-web/src/main/java/cn/tr/plugin/web/config/jackson/mapper/serializer/EnumDeserializer.java

@@ -1,5 +1,6 @@
 package cn.tr.plugin.web.config.jackson.mapper.serializer;
 
+import cn.hutool.core.util.EnumUtil;
 import cn.tr.core.enums.IEnum;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.BeanProperty;
@@ -51,9 +52,11 @@ public class EnumDeserializer extends JsonDeserializer<Enum<?>> implements Conte
 
     public static Enum<?> defaultEnumTransform(Class<?> type, String indexString) {
         Enum<?>[] enumConstants = (Enum<?>[]) type.getEnumConstants();
+        if(enumConstants==null||enumConstants.length==0){
+            return null;
+        }
         try {
-            int index = Integer.parseInt(indexString);
-            return enumConstants[index];
+            return EnumUtil.likeValueOf(enumConstants[0].getClass(),indexString);
         } catch (NumberFormatException e) {
             return null;
         }

tr-plugins/tr-spring-boot-starter-plugin-websocket/pom.xml

@@ -24,6 +24,12 @@
         <dependency>
             <groupId>org.t-io</groupId>
             <artifactId>tio-websocket-spring-boot-starter</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.redisson</groupId>
+                    <artifactId>redisson</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
     </dependencies>
 </project>

tr-test/pom.xml

@@ -25,6 +25,12 @@
             <version>0.0.9</version>
         </dependency>
 
+        <dependency>
+            <groupId>cn.tr</groupId>
+            <artifactId>tr-spring-boot-starter-plugin-eventbus</artifactId>
+            <version>0.0.9</version>
+        </dependency>
+
         <dependency>
             <groupId>cn.tr</groupId>
             <artifactId>tr-spring-boot-starter-plugin-satoken</artifactId>
@@ -40,7 +46,17 @@
         <dependency>
             <groupId>cn.tr</groupId>
             <artifactId>tr-spring-boot-starter-plugin-dict</artifactId>
-            <version>0.0.9</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
     </dependencies>
 </project>

tr-test/src/main/java/cn/tr/test/TestController.java

@@ -1,10 +1,22 @@
 package cn.tr.test;
 
+import cn.tr.core.utils.JsonUtils;
+import cn.tr.plugin.dict.annotation.Dict;
+import cn.tr.plugin.dict.bo.DictBO;
+import cn.tr.plugin.dict.config.cache.DictManager;
+import cn.tr.plugin.eventbus.annotation.Subscribe;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.PostConstruct;
+import java.util.*;
 /**
  * @ClassName : TestController
  * @Description :
@@ -14,10 +26,39 @@ import org.springframework.web.bind.annotation.RestController;
 @RestController
 
 @RequestMapping("/test")
+@Slf4j
 public class TestController {
+    @Autowired
+    private DictManager dictManager;
+
+    @PostConstruct
+    public void init(){
+        dictManager.load("gender",Arrays.asList(new DictBO("1","男"),new DictBO("2","女")));
+    }
 
     @GetMapping("/123")
     public Integer test(@RequestParam("test") String test){
         return 2/1;
     }
+
+
+    @Subscribe({"test*"})
+    public void sub(User test){
+        log.info("User："+ JsonUtils.toJsonString(test));
+    }
+
+    @Subscribe({"test*"})
+    public void subString(String test){
+
+        log.info("String："+ JsonUtils.toJsonString(test));
+    }
+
+    @Data
+    @AllArgsConstructor(staticName = "of")
+    @NoArgsConstructor
+    public static class User{
+        private String id;
+        @Dict("gender")
+        private String name;
+    }
 }

tr-test/src/main/java/cn/tr/test/WebApplication.java

@@ -1,8 +1,16 @@
 package cn.tr.test;
 
+import cn.hutool.core.thread.ThreadUtil;
+import cn.tr.plugin.eventbus.config.EventBus;
+import cn.tr.plugin.eventbus.constant.EventBusConstant;
+import org.redisson.api.RTopic;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
+import javax.annotation.PostConstruct;
+
 /**
  * @ClassName : WebApplication
  * @Description :
@@ -11,6 +19,28 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
  */
 @SpringBootApplication
 public class WebApplication {
+    @Autowired
+    private RedissonClient redissonClient;
+
+    @Autowired
+    private EventBus eventBus;
+
+    @PostConstruct
+    public void init(){
+//        eventBus.subscribe("test1223", msg->{
+//            System.out.println("第------------"+ finalI +"  ----------------接收到信息:{ "+ JsonUtils.toJsonString(msg) +" }");
+//        });
+
+        ThreadUtil.execAsync(()->{
+            RTopic test = redissonClient.getTopic(EventBusConstant.EVENT_BUS_TOPIC);
+            while (true){
+                Thread.sleep(3000);
+                test.publish(TestController.User.of("test1223>>"+System.currentTimeMillis(),  "1"));
+            }
+        });
+    }
+
+
     public static void main(String[] args) {
         SpringApplication.run(WebApplication.class);
     }

tr-test/src/main/resources/application.yml

@@ -27,5 +27,5 @@ spring:
         max-active: 16
         # #连接池最大阻塞等待时间（使用负值表示没有限制）
         max-wait: -1ms
-  application:
-    name:
+  cache:
+    type: caffeine