自定义同步处理器代替eventbus

ruoyi-admin/src/main/java/com/ruoyi/web/controller/agentcenter/AgentCenterController.java

@@ -62,7 +62,7 @@ public class AgentCenterController extends BaseController
         {
             return NSSDKServer.Fail(e.getErrno(), e.getMessage());
         }
-        catch(IllegalArgumentException e)
+        catch(Exception e)
         {
             return NSSDKServer.Fail(NSErrno.ERRNO_ERROR, e.getMessage());
         }

ruoyi-agentcenter/src/main/java/com/ruoyi/agentcenter/listener/ContractionListener.java

@@ -3,6 +3,7 @@ package com.ruoyi.agentcenter.listener;
 import cn.hutool.core.lang.Assert;
 import com.alibaba.fastjson2.JSONObject;
 import com.google.common.eventbus.Subscribe;
+import com.nsgk.agentcentersdk.core.NSProtocol;
 import com.nsgk.agentcentersdk.entity.NSContractionEntity;
 import com.ruoyi.agentcenter.domain.TAgentContraction;
 import com.ruoyi.agentcenter.domain.TAgentTask;
@@ -13,6 +14,8 @@ import com.ruoyi.agentcenter.service.ITAgentContractionService;
 import com.ruoyi.agentcenter.service.ITAgentTaskService;
 import com.ruoyi.common.core.domain.entity.SysDept;
 import com.ruoyi.common.utils.EventBusEngine;
+import com.ruoyi.common.utils.handler.CallbackHandler;
+import com.ruoyi.common.utils.handler.HandlerEngine;
 import com.ruoyi.common.utils.sql.SqlUtil;
 import com.ruoyi.system.service.ISysDeptService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -34,11 +37,10 @@ public class ContractionListener
     @PostConstruct
     public void init()
     {
-        EventBusEngine.Register(this);
+        HandlerEngine.Register(this);
     }
 
-    @Subscribe
-    //@AllowConcurrentEvents
+    @CallbackHandler(protocol = NSProtocol.NS_PROTOCOL_CONTRACTION)
     public void handle(ContractionSession session)
     {
         TAgentContraction contraction = conv(session.message);

ruoyi-agentcenter/src/main/java/com/ruoyi/agentcenter/service/impl/AgentCenterImpl.java

@@ -16,6 +16,7 @@ import com.ruoyi.agentcenter.object.Session;
 import com.ruoyi.agentcenter.service.IAgentCenter;
 import com.ruoyi.common.config.RuoYiConfig;
 import com.ruoyi.common.utils.EventBusEngine;
+import com.ruoyi.common.utils.handler.HandlerEngine;
 import org.springframework.stereotype.Service;
 
 import javax.servlet.http.HttpServletRequest;
@@ -80,7 +81,7 @@ public class AgentCenterImpl implements IAgentCenter
     {
         Message<?> message = getMessage(request);
         Session<?, ?> session = createSession(message);
-        EventBusEngine.Post(session);
+        HandlerEngine.Post(session);
         return session.result;
     }
 }

ruoyi-common/src/main/java/com/ruoyi/common/utils/handler/CallbackHandler.java

@@ -0,0 +1,23 @@
+package com.ruoyi.common.utils.handler;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * 上报处理器
+ * 
+ * @author zhao
+ */
+@Target( ElementType.METHOD )
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface CallbackHandler
+{
+    /**
+     * 协议
+     */
+    public int protocol();
+}

ruoyi-common/src/main/java/com/ruoyi/common/utils/handler/Handler.java

@@ -0,0 +1,134 @@
+package com.ruoyi.common.utils.handler;
+
+import cn.hutool.core.lang.Assert;
+import cn.hutool.core.util.StrUtil;
+import com.ruoyi.common.utils.spring.SpringUtils;
+
+import java.lang.reflect.Method;
+
+final class Handler
+{
+    public final int protocol;
+    public final Object handler;
+    public final Method method;
+
+    public Handler(int protocol, Object handler, Method method)
+    {
+        this.protocol = protocol;
+        this.handler = handler;
+        this.method = method;
+    }
+
+    public Handler(int protocol, Object handler, String methodName)
+    {
+        this.protocol = protocol;
+        this.handler = handler;
+        try
+        {
+            method = handler.getClass().getDeclaredMethod(methodName);
+            Class<?>[] parameterTypes = method.getParameterTypes();
+            if(parameterTypes.length < 1)
+                throw new RuntimeException("处理器函数参数至少有一个参数");
+        }
+        catch(Exception e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public Handler(int protocol, String beanName, String methodName)
+    {
+        this.protocol = protocol;
+        this.handler = SpringUtils.getBean(beanName);
+        try
+        {
+            method = handler.getClass().getDeclaredMethod(methodName);
+            Class<?>[] parameterTypes = method.getParameterTypes();
+            if(parameterTypes.length < 1)
+                throw new RuntimeException("处理器函数参数至少有一个参数");
+        }
+        catch(Exception e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public Handler(int protocol, Class<?> beanClazz, String methodName)
+    {
+        this.protocol = protocol;
+        this.handler = SpringUtils.getBean(beanClazz);
+        try
+        {
+            method = handler.getClass().getDeclaredMethod(methodName);
+        }
+        catch(Exception e)
+        {
+            e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+    }
+
+    public Object Invoke(Object...args)
+    {
+        Check(args);
+        if(!method.isAccessible())
+            method.setAccessible(true);
+        try
+        {
+            return method.invoke(handler, args);
+        }
+        catch(Exception e)
+        {
+            e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void Check(Object...args)
+    {
+        String err = Match(args);
+        if(null != err)
+            throw new RuntimeException(err);
+    }
+
+    public String Match(Object...args)
+    {
+        int argsLength = null != args ? args.length : 0;
+        Class<?>[] parameterTypes = method.getParameterTypes();
+        if(argsLength != parameterTypes.length)
+            return StrUtil.format("调用处理器函数参数数量不一致: 需要({})|实际({})", parameterTypes.length, argsLength);
+        for(int i = 0; i < argsLength; i++)
+        {
+            Class<?> parameterType = parameterTypes[i];
+            if(null == args[i])
+            {
+                if(parameterType.equals(boolean.class)
+                        || parameterType.equals(byte.class)
+                        || parameterType.equals(short.class)
+                        || parameterType.equals(int.class)
+                        || parameterType.equals(long.class)
+                        || parameterType.equals(float.class)
+                        || parameterType.equals(double.class)
+                        || parameterType.equals(char.class)
+                )
+                    return StrUtil.format("调用处理器函数第{}个参数传参为null, 但实参是基本数据类型", i + 1);
+            }
+            else
+            {
+                if(!parameterType.isAssignableFrom(args[i].getClass()))
+                    return StrUtil.format("调用处理器函数第{}个参数传参与实参类型不一致: 需要({})|实际({})", i + 1, parameterType.getName(), args[i].getClass().getName());
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "Handler{" +
+                "protocol=" + protocol +
+                ", handler=" + handler +
+                ", method=" + method +
+                '}';
+    }
+}

ruoyi-common/src/main/java/com/ruoyi/common/utils/handler/HandlerEngine.java

@@ -0,0 +1,139 @@
+package com.ruoyi.common.utils.handler;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.lang.Assert;
+import com.google.common.eventbus.EventBus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PreDestroy;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+@Component
+public final class HandlerEngine
+{
+    private static final Logger       logger        = LoggerFactory.getLogger("HandlerEngine" );
+    private static final List<Object> _listenerList = new ArrayList<>();
+    private static final Map<Integer, Handler> _handlerMap = new HashMap<>();
+
+    public static Object Post(Object...message)
+    {
+        Handler handler = _handlerMap.values().stream().filter((x) -> null == x.Match(message)).findFirst().orElse(null);
+        Assert.notNull(handler, "没有对应处理器被注册");
+        return handler.Invoke(message);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> T PostT(Object...message)
+    {
+        return (T)Post(message);
+    }
+
+    public static Object Run(int protocol, Object...message)
+    {
+        Handler handler = _handlerMap.get(protocol);
+        Assert.notNull(handler, "协议对应的处理器不存在: {}", protocol);
+        return handler.Invoke(message);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> T RunT(int protocol, Object...message)
+    {
+        return (T)Run(protocol, message);
+    }
+
+    public static void Register(Object listener)
+    {
+        if(null == listener)
+            return;
+        synchronized(_listenerList) {
+            if(_listenerList.contains(listener))
+                return;
+            List<Handler> handlers = parseObject(listener);
+            if(CollectionUtil.isEmpty(handlers))
+                return;
+            if(handlers.stream().anyMatch((x) -> _handlerMap.containsKey(x.protocol)))
+                return;
+            for(Handler handler : handlers)
+            {
+                _handlerMap.put(handler.protocol, handler);
+                logger.info("Register -> " + handler);
+            }
+            _listenerList.add(listener);
+            logger.info("HandlerEngine::Register -> " + listener);
+        }
+    }
+
+    public static void Unregister(Object listener)
+    {
+        if(null == listener)
+            return;
+        synchronized(_listenerList) {
+            if(!_listenerList.contains(listener))
+                return;
+            List<Integer> rm = new ArrayList<>();
+            _handlerMap.forEach((k, v) -> {
+                if(v.handler == listener)
+                {
+                    rm.add(k);
+                    logger.info("Unregister -> " + v);
+                }
+            });
+            rm.forEach(_handlerMap::remove);
+            _listenerList.remove(listener);
+            logger.info("HandlerEngine::Unregister -> " + listener);
+        }
+    }
+
+    @PreDestroy
+    public static void UnregisterAll()
+    {
+        synchronized(_listenerList) {
+            logger.info("HandlerEngine::UnregisterAll -> " + _listenerList.size());
+            _handlerMap.clear();
+            _listenerList.clear();
+        }
+    }
+
+    private static List<Handler> parseObject(Object obj)
+    {
+        if(null == obj)
+            return null;
+        Class<?> clazz = obj.getClass();
+        return parseObject(obj, clazz);
+    }
+
+    private static List<Handler> parseObject(Object obj, Class<?> clazz)
+    {
+        if(null == clazz)
+            return null;
+        List<Handler> list = new ArrayList<>();
+        Method[] declaredMethods = clazz.getDeclaredMethods();
+        for(Method method : declaredMethods)
+        {
+            CallbackHandler annotation = method.getAnnotation(CallbackHandler.class);
+            if(null == annotation)
+                continue;
+            int protocol = annotation.protocol();
+            Handler handler = new Handler(protocol, obj, method);
+            list.add(handler);
+        }
+        Class<?> superclass = clazz.getSuperclass();
+        if(null != superclass && !Object.class.equals(superclass))
+        {
+            List<Handler> handlers = parseObject(obj, superclass);
+            if(CollectionUtil.isNotEmpty(handlers))
+                list.addAll(handlers);
+        }
+        return list;
+    }
+}