规则刷新通知，接收不到通知

ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/config/GatewayConfig.java

@@ -5,9 +5,14 @@ import com.ruoyi.gateway.config.handler.GlobalErrorWebExceptionHandler;
 import com.ruoyi.gateway.config.provider.RedisRouteDefinitionRepository;
 import com.ruoyi.gateway.filter.CustomerGlobalFilter;
 import com.ruoyi.gateway.ratelimit.CustomerRedisRateLimiter;
+import com.ruoyi.gateway.utils.beans.TopicMsg;
+import com.ruoyi.gateway.utils.caching.CachingRule;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RTopic;
 import org.redisson.api.RedissonClient;
+import org.redisson.client.codec.Codec;
+import org.redisson.codec.SerializationCodec;
+import org.redisson.codec.TypedJsonJacksonCodec;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.web.reactive.error.ErrorWebExceptionHandler;
 import org.springframework.cloud.gateway.event.RefreshRoutesEvent;
@@ -22,8 +27,8 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
 import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.lang.NonNull;
 
-import javax.annotation.Resource;
 import java.util.List;
 import java.util.function.Consumer;
 
@@ -35,38 +40,55 @@ import java.util.function.Consumer;
  */
 @Slf4j
 @Configuration
-public class GatewayConfig implements ApplicationEventPublisherAware
+public class GatewayConfig implements ApplicationEventPublisherAware {
-{
+    public static final Codec TOPIC_MSG_CODES_INSTANCE = new TypedJsonJacksonCodec(String.class, TopicMsg.class);
-    private RedissonClient client = SpringUtil.getBean(RedissonClient.class);
-    public ApplicationEventPublisher publisher;
     public static final String TOPIC_GATEWAY_REFRESH_ROUTE = "TOPIC_GATEWAY_REFRESH_ROUTE";
+    public static final String TOPIC_GATEWAY_RULE = "TOPIC_GATEWAY_RULE";
+    public ApplicationEventPublisher publisher;
+    private final RedissonClient client = SpringUtil.getBean(RedissonClient.class);
+
     {
-        subscribe(TOPIC_GATEWAY_REFRESH_ROUTE, String.class, msg -> {
+        subscribe(TOPIC_GATEWAY_REFRESH_ROUTE, TopicMsg.class, msg -> {
             publisher.publishEvent(new RefreshRoutesEvent(this));
             log.info("网关接收通知[{}]刷新本地路由!", msg);
         });
+        subscribe(TOPIC_GATEWAY_RULE, TopicMsg.class, msg -> {
+            switch (msg.getType()) {
+                case ADD:
+                case UPDATE:
+                case DELETE:
+                    CachingRule.evict(msg.getId());
+                    log.info("网关接收通知id:[{}], type:[{}], msg:[{}]清除本地规则!", msg.getId(), msg.getType(), msg.getMsg());
+                    break;
+                case REFRESH:
+                    CachingRule.refresh();
+                    log.info("网关接收通知id:[{}], type:[{}], msg:[{}]本地所有规则!", msg.getId(), msg.getType(), msg.getMsg());
+                    break;
+                default:
+                    break;
+            }
+        });
     }
+
     /**
      * Redis 路由仓库
      *
      * @return redis路由仓库
      */
     @Bean
-    public RouteDefinitionRepository redisRouteDefinitionRepository()
+    public RouteDefinitionRepository redisRouteDefinitionRepository() {
-    {
         return new RedisRouteDefinitionRepository(client);
     }
 
     @Bean
     public CustomerRedisRateLimiter customerRedisRateLimiter(ReactiveStringRedisTemplate redisTemplate,
-             @Qualifier(RedisRateLimiter.REDIS_SCRIPT_NAME) RedisScript<List<Long>> redisScript,
+                                                             @Qualifier(RedisRateLimiter.REDIS_SCRIPT_NAME) RedisScript<List<Long>> redisScript,
-             ConfigurationService configurationService, DefaultRedisScript<Long> timeRedisScript) {
+                                                             ConfigurationService configurationService, DefaultRedisScript<Long> timeRedisScript) {
         return new CustomerRedisRateLimiter(redisTemplate, redisScript, configurationService, timeRedisScript);
     }
 
     @Bean
-    public GlobalFilter customerGlobalFilter(CustomerRedisRateLimiter customerRedisRateLimiter)
+    public GlobalFilter customerGlobalFilter(CustomerRedisRateLimiter customerRedisRateLimiter) {
-    {
         return new CustomerGlobalFilter(customerRedisRateLimiter);
     }
 
@@ -83,13 +105,12 @@ public class GatewayConfig implements ApplicationEventPublisherAware
      * @param consumer   自定义处理
      */
     public <T> void subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) {
-        RTopic topic = client.getTopic(channelKey);
+        RTopic topic = client.getTopic(channelKey, new SerializationCodec());
         topic.addListener(clazz, (channel, msg) -> consumer.accept(msg));
     }
 
     @Override
-    public void setApplicationEventPublisher(ApplicationEventPublisher publisher)
+    public void setApplicationEventPublisher(@NonNull ApplicationEventPublisher publisher) {
-    {
         this.publisher = publisher;
     }
 }

ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/GatewayUtils.java

@@ -4,6 +4,8 @@ import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.lang.Assert;
 import cn.hutool.extra.spring.SpringUtil;
 import com.ruoyi.gateway.utils.beans.IscRule;
+import com.ruoyi.gateway.utils.caching.CachingRule;
+import org.redisson.api.DeletedObjectListener;
 import org.redisson.api.RMap;
 import org.redisson.api.RedissonClient;
 import org.redisson.client.codec.Codec;
@@ -164,8 +166,18 @@ public class GatewayUtils {
      * @return
      */
     public static IscRule getRule(String ak, String routeId) {
+        String key = ak + ':' + routeId;
+        return CachingRule.getRule(key, GatewayUtils::getRedisRule);
+    }
+
+    /**
+     * 获取Redis Rule
+     * @param key
+     * @return
+     */
+    private static IscRule getRedisRule(String key) {
         final RMap<String, IscRule> map = client.getMap(KEY_RULES, RULE_CODES_INSTANCE);
-        return map.get(ak + ':' + routeId);
+        return map.get(key);
     }
 
     /**

ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/beans/TopicMsg.java

@@ -0,0 +1,37 @@
+package com.ruoyi.gateway.utils.beans;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+/**
+ * @author Wenchao Gong
+ * @date 2021-10-26
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class TopicMsg implements Serializable {
+    /** 消息ID */
+    private String id;
+
+    /** 消息内容 */
+    private String msg;
+
+    /** 消息类型 */
+    private Type type;
+
+    /** 消息类型枚举 */
+    public static enum Type {
+        /** 新增单条记录 */
+        ADD,
+        /** 更新单条记录 */
+        UPDATE,
+        /** 删除单条记录 */
+        DELETE,
+        /** 刷新所有 */
+        REFRESH
+    }
+}

ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/caching/CachingRule.java

@@ -0,0 +1,53 @@
+package com.ruoyi.gateway.utils.caching;
+
+import com.ruoyi.gateway.utils.beans.IscRule;
+import org.springframework.cache.Cache;
+import org.springframework.cache.concurrent.ConcurrentMapCache;
+
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * @author Wenchao Gong
+ * @date 2021-10-22
+ */
+public class CachingRule {
+    public static final String RULE_CACHE_NAME = "RULE_CACHE";
+    public static final Cache RULE_CACHE = new ConcurrentMapCache(RULE_CACHE_NAME);
+
+    /**
+     * 通过 Key 获取缓存中的 Rule
+     *
+     * @param key    缓存key
+     * @param mapper 如果缓存中没有 key 从mapper 中获取
+     * @return
+     */
+    public static IscRule getRule(String key, Function<String, IscRule> mapper) {
+        Cache.ValueWrapper value = RULE_CACHE.get(key);
+        if (Objects.nonNull(value)) {
+            return (IscRule) value.get();
+        }
+        IscRule rule = mapper.apply(key);
+        RULE_CACHE.put(key, rule);
+        return rule;
+    }
+
+    /**
+     * 清除 缓存 如果key存在
+     *
+     * @param key 缓存key
+     * @return 是否清除
+     */
+    public static boolean evict(String key) {
+        return RULE_CACHE.evictIfPresent(key);
+    }
+
+    /**
+     * 清除所有缓存 一刷新整个缓存
+     * @return
+     */
+    public static boolean refresh() {
+        RULE_CACHE.clear();
+        return true;
+    }
+}

ruoyi-isc/src/main/java/com/ruoyi/isc/utils/RouteUtils.java

@@ -4,16 +4,14 @@ import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
 import com.ruoyi.common.constant.IscConstants;
 import com.ruoyi.common.exception.ServiceException;
-import com.ruoyi.common.utils.RedisUtils;
 import com.ruoyi.common.utils.spring.SpringUtils;
 import com.ruoyi.isc.domain.IscAppService;
 import com.ruoyi.isc.domain.IscService;
-import com.ruoyi.isc.utils.beans.IscFilterDefinition;
+import com.ruoyi.isc.utils.beans.*;
-import com.ruoyi.isc.utils.beans.IscPredicateDefinition;
+import com.ruoyi.isc.utils.beans.TopicMsg.Type;
-import com.ruoyi.isc.utils.beans.IscRouteDefinition;
-import com.ruoyi.isc.utils.beans.IscRule;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RMap;
+import org.redisson.api.RTopic;
 import org.redisson.api.RedissonClient;
 import org.redisson.client.codec.Codec;
 import org.redisson.codec.TypedJsonJacksonCodec;
@@ -21,7 +19,6 @@ import org.redisson.codec.TypedJsonJacksonCodec;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.*;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -35,13 +32,15 @@ public class RouteUtils {
 
     public static final Codec ROUTE_CODES_INSTANCE = new TypedJsonJacksonCodec(String.class, IscRouteDefinition.class);
     public static final Codec RULE_CODES_INSTANCE = new TypedJsonJacksonCodec(String.class, IscRule.class);
+    public static final Codec TOPIC_MSG_CODES_INSTANCE = new TypedJsonJacksonCodec(String.class, TopicMsg.class);
     public static final String TOPIC_GATEWAY_REFRESH_ROUTE = "TOPIC_GATEWAY_REFRESH_ROUTE";
+    public static final String TOPIC_GATEWAY_RULE = "TOPIC_GATEWAY_RULE";
     /**
      * Gateway 虚拟路径前缀
      */
     public static final String PATH_PREFIX = "/proxy";
     public static final String ACCESS_KEY_NAME = "ak";
-    private static RedissonClient client = SpringUtils.getBean(RedissonClient.class);
+    private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class);
 
     /**
      * 获取虚拟URL Path 部分
@@ -82,12 +81,32 @@ public class RouteUtils {
     }
 
     /**
-     * 发布路由刷新通知
+     * 发布通道消息
-     * @param consumer
+     *
-     * @return
+     * @param channelKey 通道key
+     * @param msg        发送数据
+     * @param consumer   自定义处理
      */
-    public static void sendRefreshRouteToGateway(Consumer<String> consumer) {
+    public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) {
-        RedisUtils.publish(TOPIC_GATEWAY_REFRESH_ROUTE, DateUtil.now(), consumer);
+        RTopic topic = CLIENT.getTopic(channelKey, TOPIC_MSG_CODES_INSTANCE);
+        topic.publish(msg);
+        consumer.accept(msg);
+    }
+
+    /**
+     * 发布路由刷新通知
+     * @param consumer 发送消息后执行
+     */
+    public static void sendRefreshRouteToGateway(String id, Type type, Consumer<TopicMsg> consumer) {
+        publish(TOPIC_GATEWAY_REFRESH_ROUTE, new TopicMsg(id, DateUtil.now(), type), consumer);
+    }
+
+    /**
+     * 发布规则通知
+     * @param consumer 发送消息后执行
+     */
+    public static void sendRuleNoticeToGateway(String id, Type type, Consumer<TopicMsg> consumer) {
+        publish(TOPIC_GATEWAY_RULE, new TopicMsg(id, DateUtil.now(), type), consumer);
     }
 
     /**
@@ -100,10 +119,11 @@ public class RouteUtils {
     {
         Map<String, IscRouteDefinition> routeMap = routes.stream().collect(Collectors.toMap(IscRouteDefinition::getId,
                 Function.identity(), (o1, o2) -> o2));
-        final RMap<String, IscRouteDefinition> map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
+        final RMap<String, IscRouteDefinition> map = CLIENT.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
         map.clear();
         map.putAll(routeMap);
-        sendRefreshRouteToGateway((time) -> log.info("路由刷新完成，通知网关刷新路由![{}]", time));
+        sendRefreshRouteToGateway(null, Type.REFRESH,
+                (msg) -> log.info("路由刷新完成，通知网关刷新路由![{}]", msg.getMsg()));
         return true;
     }
 
@@ -124,18 +144,21 @@ public class RouteUtils {
      * @param consumer 回调操作
      * @return 是否成功
      */
-    public static boolean saveRoute(Collection<IscRouteDefinition> routes, Consumer<String> consumer)
+    public static boolean saveRoute(Collection<IscRouteDefinition> routes, Consumer<TopicMsg> consumer)
     {
         if(CollectionUtil.isEmpty(routes)) {
             return true;
         }
-        final RMap<String, IscRouteDefinition> map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
+        final RMap<String, IscRouteDefinition> map = CLIENT.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
         Map<String, IscRouteDefinition> collect = routes.stream().collect(Collectors.toMap(IscRouteDefinition::getId, Function.identity()));
         map.putAll(collect);
+        final String id = String.join(",", collect.keySet());
+        Type type = Type.UPDATE;
         if(Objects.isNull(consumer)) {
-            consumer = (time) -> log.info("保存路由[{}]完成，通知网关刷新路由![{}]", map.keySet(), time);
+            type = Type.ADD;
+            consumer = (msg) -> log.info("保存[{}]路由[{}]完成，通知网关刷新路由![{}]", msg.getType(), msg.getId(), msg.getMsg());
         }
-        sendRefreshRouteToGateway(consumer);
+        sendRefreshRouteToGateway(id, type, consumer);
         return true;
     }
 
@@ -147,9 +170,12 @@ public class RouteUtils {
      */
     public static boolean deleteRoute(Collection<String> routeIds)
     {
-        final RMap<String, IscRouteDefinition> map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
+        final RMap<String, IscRouteDefinition> map = CLIENT.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE);
-        map.fastRemove(routeIds.toArray(new String[0]));
+        final String[] ids = routeIds.toArray(new String[0]);
-        sendRefreshRouteToGateway((time) -> log.info("删除路由[{}]完成，通知网关刷新路由![{}]", routeIds, time));
+        map.fastRemove(ids);
+        final String id = String.join(",", routeIds);
+        sendRefreshRouteToGateway(id, Type.DELETE,
+                (msg) -> log.info("删除[{}]路由[{}]完成，通知网关刷新路由![{}]", msg.getType(), msg.getId(), msg.getMsg()));
         return true;
     }
 
@@ -161,9 +187,8 @@ public class RouteUtils {
      */
     public static boolean updateRoute(IscRouteDefinition route)
     {
-        boolean result = saveRoute(Collections.singletonList(route), (time) ->
+        return saveRoute(Collections.singletonList(route),
-                log.info("更新路由[{}]完成，通知网关刷新路由![{}]", route.getId(), time));
+                (msg) -> log.info("保存[{}]路由[{}]完成，通知网关刷新路由![{}]", msg.getType(), msg.getId(), msg.getMsg()));
-        return result;
     }
 
     /**
@@ -212,9 +237,9 @@ public class RouteUtils {
     /**
      * 生成 服务对应AK调用规则
      *
-     * @param appService
+     * @param appService 服务申请信息
-     * @param accessKey
+     * @param accessKey AccessKey
-     * @return
+     * @return 规则信息
      */
     public static IscRule generateRule(IscAppService appService, String accessKey)
     {
@@ -237,22 +262,42 @@ public class RouteUtils {
     public static boolean refreshRules(List<IscRule> rules)
     {
         Map<String, IscRule> ruleMap = rules.stream().collect(Collectors.toMap(IscRule::getId, Function.identity(), (o1, o2) -> o2));
-        final RMap<String, IscRule> map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
+        final RMap<String, IscRule> map = CLIENT.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
         map.clear();
         map.putAll(ruleMap);
+        final String id = String.join(",", ruleMap.keySet());
+        sendRuleNoticeToGateway(id, Type.REFRESH,
+                (msg) ->  log.info("刷新[{}]所有规则[{}]完成，通知网关刷新规则![{}]", msg.getType(), msg.getId(), msg.getMsg()));
         return true;
     }
 
     /**
      * 保存规则信息
      *
-     * @param rule 规则信息
+     * @param rule     规则信息
      * @return 是否成功
      */
-    public static boolean saveRule(IscRule rule)
+    public static boolean saveRule(IscRule rule) {
+        return saveRule(rule, null);
+    }
+
+    /**
+     * 保存规则信息
+     *
+     * @param rule     规则信息
+     * @param consumer 回调
+     * @return 是否成功
+     */
+    public static boolean saveRule(IscRule rule, Consumer<TopicMsg> consumer)
     {
-        final RMap<String, IscRule> map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
+        final RMap<String, IscRule> map = CLIENT.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
         map.put(rule.getId(), rule);
+        Type type = Type.UPDATE;
+        if(Objects.isNull(consumer)) {
+            type = Type.ADD;
+            consumer = (msg) ->  log.info("新增[{}]规则[{}]完成，通知网关刷新规则![{}]", msg.getType(), msg.getId(), msg.getMsg());
+        }
+        sendRuleNoticeToGateway(rule.getId(), type, consumer);
         return true;
     }
 
@@ -264,8 +309,10 @@ public class RouteUtils {
      */
     public static boolean deleteRule(String ruleId)
     {
-        final RMap<String, IscRule> map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
+        final RMap<String, IscRule> map = CLIENT.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE);
         map.remove(ruleId);
+        sendRuleNoticeToGateway(ruleId, Type.DELETE,
+                (msg) ->  log.info("删除[{}]规则[{}]完成，通知网关刷新规则![{}]", msg.getType(), msg.getId(), msg.getMsg()));
         return true;
     }
 
@@ -277,6 +324,7 @@ public class RouteUtils {
      */
     public static boolean updateRule(IscRule rule)
     {
-        return saveRule(rule);
+        return saveRule(rule,
+                (msg) ->  log.info("更新[{}]规则[{}]完成，通知网关刷新规则![{}]", msg.getType(), msg.getId(), msg.getMsg()));
     }
 }

ruoyi-isc/src/main/java/com/ruoyi/isc/utils/beans/TopicMsg.java

@@ -0,0 +1,37 @@
+package com.ruoyi.isc.utils.beans;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+/**
+ * @author Wenchao Gong
+ * @date 2021-10-26
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class TopicMsg implements Serializable {
+    /** 消息ID */
+    private String id;
+
+    /** 消息内容 */
+    private String msg;
+
+    /** 消息类型 */
+    private Type type;
+
+    /** 消息类型枚举 */
+    public static enum Type {
+        /** 新增单条记录 */
+        ADD,
+        /** 更新单条记录 */
+        UPDATE,
+        /** 删除单条记录 */
+        DELETE,
+        /** 刷新所有 */
+        REFRESH
+    }
+}