From fa54edaf6eec1365084d41190fac77ad25a4ed16 Mon Sep 17 00:00:00 2001 From: Wenchao Gong Date: Sat, 30 Oct 2021 21:48:41 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=84=E5=88=99=E5=88=B7=E6=96=B0=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=EF=BC=8C=E6=8E=A5=E6=94=B6=E4=B8=8D=E5=88=B0=E9=80=9A?= =?UTF-8?q?=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ruoyi/gateway/config/GatewayConfig.java | 51 +++++--- .../com/ruoyi/gateway/utils/GatewayUtils.java | 14 ++- .../ruoyi/gateway/utils/beans/TopicMsg.java | 37 ++++++ .../gateway/utils/caching/CachingRule.java | 53 ++++++++ .../java/com/ruoyi/isc/utils/RouteUtils.java | 114 +++++++++++++----- .../com/ruoyi/isc/utils/beans/TopicMsg.java | 37 ++++++ 6 files changed, 257 insertions(+), 49 deletions(-) create mode 100644 ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/beans/TopicMsg.java create mode 100644 ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/caching/CachingRule.java create mode 100644 ruoyi-isc/src/main/java/com/ruoyi/isc/utils/beans/TopicMsg.java diff --git a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/config/GatewayConfig.java b/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/config/GatewayConfig.java index f56ea21e5..e8ba55e62 100644 --- a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/config/GatewayConfig.java +++ b/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 -{ - private RedissonClient client = SpringUtil.getBean(RedissonClient.class); - public ApplicationEventPublisher publisher; +public class GatewayConfig implements ApplicationEventPublisherAware { + 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"; + 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> redisScript, - ConfigurationService configurationService, DefaultRedisScript timeRedisScript) { + @Qualifier(RedisRateLimiter.REDIS_SCRIPT_NAME) RedisScript> redisScript, + ConfigurationService configurationService, DefaultRedisScript 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 void subscribe(String channelKey, Class clazz, Consumer 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; } } diff --git a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/GatewayUtils.java b/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/GatewayUtils.java index 58ebc3806..da5847b6f 100644 --- a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/GatewayUtils.java +++ b/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 map = client.getMap(KEY_RULES, RULE_CODES_INSTANCE); - return map.get(ak + ':' + routeId); + return map.get(key); } /** diff --git a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/beans/TopicMsg.java b/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/beans/TopicMsg.java new file mode 100644 index 000000000..7a82c2ee2 --- /dev/null +++ b/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 + } +} diff --git a/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/caching/CachingRule.java b/ruoyi-extend/ruoyi-isc-gateway/src/main/java/com/ruoyi/gateway/utils/caching/CachingRule.java new file mode 100644 index 000000000..594787655 --- /dev/null +++ b/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 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; + } +} diff --git a/ruoyi-isc/src/main/java/com/ruoyi/isc/utils/RouteUtils.java b/ruoyi-isc/src/main/java/com/ruoyi/isc/utils/RouteUtils.java index 83d45e519..a1bfa766c 100644 --- a/ruoyi-isc/src/main/java/com/ruoyi/isc/utils/RouteUtils.java +++ b/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.IscPredicateDefinition; -import com.ruoyi.isc.utils.beans.IscRouteDefinition; -import com.ruoyi.isc.utils.beans.IscRule; +import com.ruoyi.isc.utils.beans.*; +import com.ruoyi.isc.utils.beans.TopicMsg.Type; 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 consumer) { - RedisUtils.publish(TOPIC_GATEWAY_REFRESH_ROUTE, DateUtil.now(), consumer); + public static void publish(String channelKey, T msg, Consumer 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 consumer) { + publish(TOPIC_GATEWAY_REFRESH_ROUTE, new TopicMsg(id, DateUtil.now(), type), consumer); + } + + /** + * 发布规则通知 + * @param consumer 发送消息后执行 + */ + public static void sendRuleNoticeToGateway(String id, Type type, Consumer consumer) { + publish(TOPIC_GATEWAY_RULE, new TopicMsg(id, DateUtil.now(), type), consumer); } /** @@ -100,10 +119,11 @@ public class RouteUtils { { Map routeMap = routes.stream().collect(Collectors.toMap(IscRouteDefinition::getId, Function.identity(), (o1, o2) -> o2)); - final RMap map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE); + final RMap 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 routes, Consumer consumer) + public static boolean saveRoute(Collection routes, Consumer consumer) { if(CollectionUtil.isEmpty(routes)) { return true; } - final RMap map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE); + final RMap map = CLIENT.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE); Map 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 routeIds) { - final RMap map = client.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE); - map.fastRemove(routeIds.toArray(new String[0])); - sendRefreshRouteToGateway((time) -> log.info("删除路由[{}]完成，通知网关刷新路由![{}]", routeIds, time)); + final RMap map = CLIENT.getMap(IscConstants.KEY_ROUTES, ROUTE_CODES_INSTANCE); + final String[] ids = routeIds.toArray(new String[0]); + 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) -> - log.info("更新路由[{}]完成，通知网关刷新路由![{}]", route.getId(), time)); - return result; + return saveRoute(Collections.singletonList(route), + (msg) -> log.info("保存[{}]路由[{}]完成，通知网关刷新路由![{}]", msg.getType(), msg.getId(), msg.getMsg())); } /** @@ -212,9 +237,9 @@ public class RouteUtils { /** * 生成 服务对应AK调用规则 * - * @param appService - * @param accessKey - * @return + * @param appService 服务申请信息 + * @param accessKey AccessKey + * @return 规则信息 */ public static IscRule generateRule(IscAppService appService, String accessKey) { @@ -237,22 +262,42 @@ public class RouteUtils { public static boolean refreshRules(List rules) { Map ruleMap = rules.stream().collect(Collectors.toMap(IscRule::getId, Function.identity(), (o1, o2) -> o2)); - final RMap map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE); + final RMap 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 consumer) { - final RMap map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE); + final RMap 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 map = client.getMap(IscConstants.KEY_RULES, RULE_CODES_INSTANCE); + final RMap 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())); } } diff --git a/ruoyi-isc/src/main/java/com/ruoyi/isc/utils/beans/TopicMsg.java b/ruoyi-isc/src/main/java/com/ruoyi/isc/utils/beans/TopicMsg.java new file mode 100644 index 000000000..c5dfb78ce --- /dev/null +++ b/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 + } +}