sentinel源码分析: dashboard与微服务的交互、pull模式持久化

server/2024/9/24 4:20:40/

文章目录

    • 原始方式
      • 微服务端规则如何保存
      • 规则如何加载进内存
      • 微服务端接收控制台请求
      • 控制台推送规则
      • 总结
    • pull拉模式
      • 官方demo
      • 如何整合Spring Cloud
      • 整合Spring Cloud



前置知识 @SentinelResource的实现原理、SphU.entry()方法中ProcessorSlotChain链、entry.exit()

建议先会使用sentinel，并对底层实现有一些理解再来看sentinel规则持久化。当然你直接看也能看懂。



Sentinel规则的推送有下面三种模式:

推送模式说明优点缺点
原始模式API 将规则推送至客户端并直接更新到内存中，扩展写数据源 WritableDataSource简单，无任何依赖不保证一致性；规则保存 在内存中，重启即消失。 严重不建议用于生产环境
Pull模式客户端主 动向某个规则管理中心定期轮询拉取规 则，这个规则中心可以是 DB、文件等。扩展写数据源 WritableDataSource简单，无任何依赖； 规则持久化不保证一致性；实时性不 保证，拉取过于频繁也可 能会有性能问题。
Push模式规则中心 统一推送，客户端通过注册监听器的方 式时刻监听变化，比如使用 Nacos、 Zookeeper 等配置中心。这种方式有 更好的实时性和一致性保证。生产环境 下一般采用 push 模式的数据源。扩展读数据源 ReadableDataSource规则持久化；一致 性；快速引入第三方依赖



原始方式

微服务与控制台通信流程 在线流程图

在这里插入图片描述



如果不做任何修改，Dashboard 的推送规则方式是通过 API 将规则推送至客户端 并直接更新到内存中：

在这里插入图片描述



微服务端规则如何保存

我们先来看看我们微服务这边规则是如何保存的。就以流控为例

在FlowSlot类的entry()方法中，它会先取出当前资源所有流控规则

// 取当前资源所有流控规则
Collection<FlowRule> rules = ruleProvider.apply(resource.getName());if (rules != null) {for (FlowRule rule : rules) {// 校验流控if (!canPassCheck(rule, context, node, count, prioritized)) {throw new FlowException(rule.getLimitApp(), rule);}}
}// 继续更近ruleProvider.apply(resource.getName())方法，进入到FlowSlot.apply()
public Collection<FlowRule> apply(String resource) {// 缓存规则Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();return flowRules.get(resource);
}// 我们知道了是通过FlowRuleManager.getFlowRuleMap();这个流控规则管理器得到了流控规则集合。继续更近该方法
public class FlowRuleManager {// 找到了真正保存内存中规则对象的mapprivate static volatile Map<String, List<FlowRule>> flowRules = new HashMap<>();。。。
}



我们现在就知道了下面这些数据：

  • 微服务端 规则实体对象：FlowRule

  • 微服务端 缓存流控规则管理器：FlowRuleManager.getFlowRuleMap()

  • 微服务端 保存规则的集合定义Map<String, List<FlowRule>> flowRules = new HashMap<>()

  • 微服务端 加载规则进内存FlowRuleManager.loadRules(rules);



规则如何加载进内存

sentinel的入门代码如下

public static void main(String[] args) {// 配置规则.initFlowRules();while (true) {// 1.5.0 版本开始可以直接利用 try-with-resources 特性try (Entry entry = SphU.entry("HelloWorld")) {// 被保护的逻辑System.out.println("hello world");} catch (BlockException ex) {// 处理被流控的逻辑System.out.println("blocked!");}}
}private static void initFlowRules(){List<FlowRule> rules = new ArrayList<>();FlowRule rule = new FlowRule();rule.setResource("HelloWorld");rule.setGrade(RuleConstant.FLOW_GRADE_QPS);// Set limit QPS to 20.rule.setCount(20);rules.add(rule);// 核心代码，加载规则进内存// 可以发现 这里也是使用的流控规则管理器FlowRuleManagerFlowRuleManager.loadRules(rules);
}



真正将规则加载进内存中FlowRuleManager.loadRules(rules);：

  • 遍历PropertyListener监听器集合，找对应的监听器去处理
  • 将传过来的list转换为我们对应的Map集合 Map<String, List<FlowRule>> rules
  • 直接复制给flowRules这个成员属性，这里就和上面微服务端规则如何保存串联起来了
public static void loadRules(List<FlowRule> rules) {// 加载资源进内存 Map<String, List<FlowRule>> flowRules 集合中currentProperty.updateValue(rules);
}// 调用进DynamicSentinelProperty#updateValue方法中
public boolean updateValue(T newValue) {if (isEqual(value, newValue)) {return false;}RecordLog.info("[DynamicSentinelProperty] Config will be updated to: {}", newValue);value = newValue;for (PropertyListener<T> listener : listeners) {// 对应的监听器去处理listener.configUpdate(newValue);}return true;
}// 这里会调用进FlowRuleManager的内部类FlowPropertyListener.configUpdate()
public synchronized void configUpdate(List<FlowRule> value) {// 将传过来的list转换为我们对应的Map集合Map<String, List<FlowRule>> rules = FlowRuleUtil.buildFlowRuleMap(value);if (rules != null) {// 直接复制给flowRules这个成员属性//  这里就和上面微服务端规则如何保存串联起来了flowRules = rules;}RecordLog.info("[FlowRuleManager] Flow rules received: {}", rules);
}



微服务端接收控制台请求

  • 我们知道我们使用sentinel和springcloudAlibaba整合会调用spring-cloud-starter-alibaba-sentinel依赖，而在这个依赖中的spring.factories文件中会定义一个SentinelWebAutoConfiguration自动配置类，它实现了WebMvcConfigurer接口，并重写addInterceptors()方法它会添加拦截器，拦截器中会对我们的请求进行拦截并添加定义资源的代码 Entry entry = SphU.entry(...)

  • 我们直接使用@SentinelResource注解方式，spring.factories文件中会定义一个SentinelAutoConfiguration自动配置类，并添加了一个SentinelResourceAspectBean对象，它通过AOP对要执行的目标方法也加了定义资源的代码 Entry entry = SphU.entry(...)



而在Env类的静态代码块中，会使用SPI加载InitFunc接口，加载出来的其中一个核心类，客户端启动的接口服务，提供给dashboard查询数据以及接收各种规则使用：com.alibaba.csp.sentinel.transport.init.CommandCenterInitFunc ：

  • 使用spi技术，加载各个CommandHandler，其中有ModifyRulesCommandHandler是处理修改规则的handler
  • 开启ServerSocket通信
public class CommandCenterInitFunc implements InitFunc {@Overridepublic void init() throws Exception {CommandCenter commandCenter = CommandCenterProvider.getCommandCenter();if (commandCenter == null) {RecordLog.warn("[CommandCenterInitFunc] Cannot resolve CommandCenter");return;}// 使用spi技术，加载各个CommandHandler，其中有ModifyRulesCommandHandler是处理修改规则的handlercommandCenter.beforeStart();// 开启ServerSocket通信commandCenter.start();RecordLog.info(...);}
}



进入ModifyRulesCommandHandler类的handler()方法：

  • 把控制台传递过来的请求参数转换为 List<FlowRule>
  • 加载规则，直接调用FlowRuleManager流控规则管理器的loadRules()方法
  • 写数据源的操作，默认情况下是没有WritableDataSource，我们可以在这里进行扩展进行持久化操作
  • 响应给sentinel控制台
// 注意name = "setRules"，这就是控制台请求服务端的url路径
@CommandMapping(name = "setRules", desc = "modify the rules, accept param: type={ruleType}&data={ruleJson}")
public class ModifyRulesCommandHandler implements CommandHandler<String> {public CommandResponse<String> handle(CommandRequest request) {//......// 处理流控规则if (FLOW_RULE_TYPE.equalsIgnoreCase(type)) {// 把控制台传递过来的请求参数转换为 List<FlowRule>List<FlowRule> flowRules = JSONArray.parseArray(data, FlowRule.class);// 加载规则，直接调用FlowRuleManager流控规则管理器的loadRules()方法// 这里就和上方 微服务端规则是如何加载进内存的串联起来了FlowRuleManager.loadRules(flowRules);// 关键一步，这里会有一个写数据源的操作。默认情况下是没有WritableDataSource，我们可以在这里进行扩展if (!writeToDataSource(getFlowDataSource(), flowRules)) {result = WRITE_DS_FAILURE_MSG;}// 响应给sentinel控制台return CommandResponse.ofSuccess(result);// 处理权限规则} else if (AUTHORITY_RULE_TYPE.equalsIgnoreCase(type)) {List<AuthorityRule> rules = JSONArray.parseArray(data, AuthorityRule.class);AuthorityRuleManager.loadRules(rules);if (!writeToDataSource(getAuthorityDataSource(), rules)) {result = WRITE_DS_FAILURE_MSG;}return CommandResponse.ofSuccess(result);// 处理熔断规则} else if (DEGRADE_RULE_TYPE.equalsIgnoreCase(type)) {List<DegradeRule> rules = JSONArray.parseArray(data, DegradeRule.class);DegradeRuleManager.loadRules(rules);if (!writeToDataSource(getDegradeDataSource(), rules)) {result = WRITE_DS_FAILURE_MSG;}return CommandResponse.ofSuccess(result);// 处理系统规则} else if (SYSTEM_RULE_TYPE.equalsIgnoreCase(type)) {List<SystemRule> rules = JSONArray.parseArray(data, SystemRule.class);SystemRuleManager.loadRules(rules);if (!writeToDataSource(getSystemSource(), rules)) {result = WRITE_DS_FAILURE_MSG;}return CommandResponse.ofSuccess(result);}return CommandResponse.ofFailure(new IllegalArgumentException("invalid type"));}
}



控制台推送规则

我们在sentinel控制台进行了资源规则的更新，控制台是如何通知微服务进行相应的更新操作的嘞？

dashboard控制台处理web界面请求的位置是FlowControllerV1，请求路径为/v1/flow/rule，新增规则是POST请求，修改规则是PUT规则

  • 在dashboard这边也会将规则在内存中保存一份
  • 调用微服务端，更新规则
// 不管是新增规则还是更新规则，处理细枝末节的代码，关键代码就是下面这两行
public Result<FlowRuleEntity> apiAddFlowRule(@RequestBody FlowRuleEntity entity) {......try {// 在dashboard这边也会将规则在内存中进行保存entity = repository.save(entity);// 调用微服务端，更新规则publishRules(entity.getApp(), entity.getIp(), entity.getPort()).get(5000, TimeUnit.MILLISECONDS);return Result.ofSuccess(entity);} catch (Throwable t) {......}
}



首先看dashboard这边是如何将规则在内存中保存的，这里会调用到InMemoryRuleRepositoryAdapter#save，从这个类名就能看出来是操作内存的

这个类实现了RuleRepository接口，那么这里也是一个规则持久化的扩展点，我们可以自定义类，实现该接口，然后写相应的处理逻辑，在改一下dashboard控制台这边调用的源代码。（这种方式可以先考虑，但不是必须要这么做）

  • 将规则数据分别保存在allRules machineRules appRules这三个集合中
public T save(T entity) {if (entity.getId() == null) {entity.setId(nextId());}T processedEntity = preProcess(entity);if (processedEntity != null) {// 所有规则集合中保存一份allRules.put(processedEntity.getId(), processedEntity);// 对应微服务ip+port机器的集合中保存一份machineRules.computeIfAbsent(MachineInfo.of(processedEntity.getApp(), processedEntity.getIp(),processedEntity.getPort()), e -> new ConcurrentHashMap<>(32)).put(processedEntity.getId(), processedEntity);// appRules集合中也保存一份appRules.computeIfAbsent(processedEntity.getApp(), v -> new ConcurrentHashMap<>(32)).put(processedEntity.getId(), processedEntity);}return processedEntity;
}



在看看publishRules()是如何调用微服务端进行规则更新的

  • 从machineRules 对应微服务ip+port机器的集合 中把规则取出来List<FlowRuleEntity> rules
  • 发送请求给微服务端 http://ip:8719/setRules
private CompletableFuture<Void> publishRules(String app, String ip, Integer port) {// 通过machineRules  对应微服务ip+port机器的集合 中把规则取出来// 控制台的规则实体对象是FlowRuleEntity，而微服务端的规则实体对象是FlowRuleList<FlowRuleEntity> rules = repository.findAllByMachine(MachineInfo.of(app, ip, port));// 发送请求给微服务端return sentinelApiClient.setFlowRuleOfMachineAsync(app, ip, port, rules);
}// 方法调用
public CompletableFuture<Void> setFlowRuleOfMachineAsync(String app, String ip, int port, List<FlowRuleEntity> rules) {return setRulesAsync(app, ip, port, FLOW_RULE_TYPE, rules);}private CompletableFuture<Void> setRulesAsync(String app, String ip, int port, String type, List<? extends RuleEntity> entities) {try {AssertUtil.notNull(entities, "rules cannot be null");AssertUtil.notEmpty(app, "Bad app name");AssertUtil.notEmpty(ip, "Bad machine IP");AssertUtil.isTrue(port > 0, "Bad machine port");// 请求参数data// toRule()方法中会把控制台的规则实体对象转换为微服务端的规则实体对象// 以流控为例 控制台的规则实体对象是FlowRuleEntity，而微服务端的规则实体对象是FlowRuleString data = JSON.toJSONString(entities.stream().map(r -> r.toRule()).collect(Collectors.toList()));Map<String, String> params = new HashMap<>(2);params.put("type", type);params.put("data", data);// 想微服务端发送请求  SET_RULES_PATH = "setRules"   // 所以调用微服务的url是 http://ip:8719/setRules  ，微服务端启动时8719端口如果被占用则会往后自增，继续找端口// 这里就和上面的服务端接收控制台请求串联起来了return executeCommand(app, ip, port, SET_RULES_PATH, params, true).thenCompose(r -> {if ("success".equalsIgnoreCase(r.trim())) {return CompletableFuture.completedFuture(null);}return AsyncUtils.newFailedFuture(new CommandFailedException(r));});} catch (Exception e) {logger.error("setRulesAsync API failed, type={}", type, e);return AsyncUtils.newFailedFuture(e);}}



总结

微服务端

  • 微服务端 规则实体对象：FlowRule

  • 微服务端 缓存流控规则管理器：FlowRuleManager.getFlowRuleMap()

  • 微服务端 保存规则的集合定义Map<String, List<FlowRule>> flowRules = new HashMap<>()

  • 微服务端 加载规则进内存FlowRuleManager.loadRules(rules);



控制台端

处理请求 FlowControllerV1 /v1/flow/rule

规则实体： FlowRuleEntity

缓存规则： 基于内存模式 allRules machineRules appRules

接口： RuleRepository 规则持久化的扩展点（考虑）

发布缓存： 推送到微服务端内存中

  • 请求参数转换： FlowRuleEntity ——》FlowRule
  • 发起请求： http://ip:8719/setRules



微服务端接收请求：

ModifyRulesCommandHandler#handle

加载规则: FlowRuleManager.loadRules(flowRules)

接口： WritableDataSource 规则持久化的扩展点



pull拉模式

pull 模式的数据源（如本地文件、RDBMS 等）一般是可写入的。使用时需要在客户端注册数据源：

  • 将对应的读数据源注册至对应的 RuleManager
  • 将写数据源注册至 transport 的 WritableDataSourceRegistry 中。

在这里插入图片描述



首先 Sentinel 控制台通过 API 将规则推送至客户端并更新到内存中，接着注册的写数据源会将新的规则保存到本地的文件中。

当我们直接修改本地文件，读数据源的线程会定期3s读取文件，将变更的配置更新到内存中

使用 pull 模式的数据源时一般不需 要对 Sentinel 控制台进行改造。这种实现方法好处是简单，坏处是无法保证监控数据的一致性。



官方demo

在sentinel的源码中，它其实提供了一些写文件的demo，位置在下图的位置

在这里插入图片描述



具体实现的核心的代码就是下面这个类

import java.io.File;
import java.util.List;import com.alibaba.csp.sentinel.datasource.FileRefreshableDataSource;
import com.alibaba.csp.sentinel.datasource.FileWritableDataSource;
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
import com.alibaba.csp.sentinel.datasource.WritableDataSource;
import com.alibaba.csp.sentinel.init.InitFunc;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
import com.alibaba.csp.sentinel.transport.util.WritableDataSourceRegistry;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;public class FileDataSourceInit implements InitFunc {@Overridepublic void init() throws Exception {//定义文件路径 String flowRuleDir = System.getProperty("user.home") + File.separator + "sentinel" + File.separator + "rules";String flowRuleFile = "flowRule.json";String flowRulePath = flowRuleDir + File.separator + flowRuleFile;// 将对应的读数据源注册至对应的FlowRuleManagerReadableDataSource<String, List<FlowRule>> ds = new FileRefreshableDataSource<>(flowRulePath, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));FlowRuleManager.register2Property(ds.getProperty());// 将写数据源注册至 transport 的WritableDataSourceRegistry中WritableDataSource<List<FlowRule>> wds = new FileWritableDataSource<>(flowRulePath, this::encodeJson);WritableDataSourceRegistry.registerFlowDataSource(wds);}private <T> String encodeJson(T t) {return JSON.toJSONString(t);}
}



我们先看看将对应的读数据源注册至对应的FlowRuleManager这一块的逻辑，读数据源FileRefreshableDataSource它的结构如下

在这里插入图片描述


关键代码

// FileRefreshableDataSource类方法     定义文件输入流   校验文件是否更新
public String readSource() throws Exception {// ......// 创建文件输入流，文件的最后修改时间进行判断，文件修改后父类就会调用到这里来FileInputStream inputStream = null;try {inputStream = new FileInputStream(file);FileChannel channel = inputStream.getChannel();if (channel.size() > buf.length) {throw new IllegalStateException(file.getAbsolutePath() + " file size=" + channel.size()+ ", is bigger than bufSize=" + buf.length + ". Can't read");}int len = inputStream.read(buf);return new String(buf, 0, len, charset);} finally {// ......}
}// 根据文件最后的修改时间进行判断
protected boolean isModified() {long curLastModified = file.lastModified();if (curLastModified != this.lastModified) {this.lastModified = curLastModified;return true;}return false;
}//---------------------------------------------------------------------------------------------
// AutoRefreshDataSource 中间父类方法 。 开启线程任务 调用父类方法 加载配置 +  更新配置
private void startTimerService() {// 创建只有一个线程的线程池service = Executors.newScheduledThreadPool(1,new NamedThreadFactory("sentinel-datasource-auto-refresh-task", true));// 每隔3s执行一次定时任务service.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 调用子类方法  获取文件的最后修改时间进行判断if (!isModified()) {return;}// 调用父类方法 加载配置T newValue = loadConfig();// 更新配置getProperty().updateValue(newValue);} catch (Throwable e) {RecordLog.info("loadConfig exception", e);}}}, recommendRefreshMs, recommendRefreshMs, TimeUnit.MILLISECONDS);
}//---------------------------------------------------------------------------------------------
// AbstractDataSource定级父类方法。   调用子类的readSource()方法，并解析配置
public T loadConfig() throws Exception {return loadConfig(readSource());
}public T loadConfig(S conf) throws Exception {// 解析配置T value = parser.convert(conf);return value;
}



我们在看看将写数据源注册至 transport 的WritableDataSourceRegistry中。点进去就会发现这里会和我们微服务端接收控制台请求时处理请求串联起来

我们这边保存写数据源，而微服务端每次接收到控制台的规则更改请求后，都会获取写数据源进行相应的写入操作。

在这里插入图片描述



所以最核心的代码就是下面这一块

public void init() throws Exception {//定义文件路径 String flowRuleDir = System.getProperty("user.home") + File.separator + "sentinel" + File.separator + "rules";String flowRuleFile = "flowRule.json";String flowRulePath = flowRuleDir + File.separator + flowRuleFile;// 将对应的读数据源注册至对应的FlowRuleManagerReadableDataSource<String, List<FlowRule>> ds = new FileRefreshableDataSource<>(flowRulePath, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));FlowRuleManager.register2Property(ds.getProperty());// 将写数据源注册至 transport 的WritableDataSourceRegistry中WritableDataSource<List<FlowRule>> wds = new FileWritableDataSource<>(flowRulePath, this::encodeJson);WritableDataSourceRegistry.registerFlowDataSource(wds);
}

这一块的代码如果要和我们SpringCloud微服务进行整合，那应该如何做嘞？



如何整合Spring Cloud

扩展点：

  • spi

spring:

  • beanPostProcessor beanFactoryPostProcessor

  • SmartInitializingSingleton

  • ApplicationListener

  • FactoryBean getObject

springboot

  • ApplicationRunner



整合Spring Cloud

我这里使用SPI的方式进行。因为Sentinel本来就有SPI的机制

我们直接在sentinel源码下创建一个子工程，并在META-INF/services目录下创建InitFunc接口对应的文件

请添加图片描述



引入的maven依赖如下

<?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>sentinel-extension</artifactId><groupId>com.alibaba.csp</groupId><version>1.8.4</version></parent><modelVersion>4.0.0</modelVersion><artifactId>sentinel-datasource-extension-file-pull</artifactId><packaging>jar</packaging><dependencies><dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-core</artifactId></dependency><dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-datasource-extension</artifactId></dependency><dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-transport-simple-http</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency><dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-parameter-flow-control</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-source-plugin</artifactId><version>3.0.1</version><executions><execution><id>attach-sources</id><goals><goal>jar</goal></goals></execution></executions></plugin></plugins></build></project>



然后编写代码如下，其实就是官方demo的那几行代码的具体实现。

/** Copyright 1999-2018 Alibaba Group Holding Ltd.** Licensed under the Apache License, Version 2.0 (the "License");* you may not use this file except in compliance with the License.* You may obtain a copy of the License at**      http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/
package com.tuling.sentinel.extension.filepull;import com.alibaba.csp.sentinel.command.handler.ModifyParamFlowRulesCommandHandler;
import com.alibaba.csp.sentinel.datasource.FileRefreshableDataSource;
import com.alibaba.csp.sentinel.datasource.FileWritableDataSource;
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
import com.alibaba.csp.sentinel.datasource.WritableDataSource;
import com.alibaba.csp.sentinel.init.InitFunc;
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRuleManager;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRuleManager;
import com.alibaba.csp.sentinel.slots.system.SystemRule;
import com.alibaba.csp.sentinel.slots.system.SystemRuleManager;
import com.alibaba.csp.sentinel.transport.util.WritableDataSourceRegistry;import java.io.FileNotFoundException;
import java.util.List;/*** InitFunc实现类，处理dataSource初始化逻辑**/
public class FileDataSourceInit implements InitFunc {@Overridepublic void init() throws Exception {//创建文件存储目录RuleFileUtils.mkdirIfNotExits(PersistenceRuleConstant.storePath);//创建规则文件RuleFileUtils.createFileIfNotExits(PersistenceRuleConstant.rulesMap);//处理流控规则逻辑  配置读写数据源dealFlowRules();// 处理降级规则dealDegradeRules();// 处理系统规则dealSystemRules();// 处理热点参数规则dealParamFlowRules();// 处理授权规则dealAuthRules();}private void dealFlowRules() throws FileNotFoundException {String ruleFilePath = PersistenceRuleConstant.rulesMap.get(PersistenceRuleConstant.FLOW_RULE_PATH).toString();//创建流控规则的可读数据源ReadableDataSource<String, List<FlowRule>> flowRuleRDS = new FileRefreshableDataSource(ruleFilePath, RuleListConverterUtils.flowRuleListParser);// 将可读数据源注册至FlowRuleManager 这样当规则文件发生变化时，就会更新规则到内存FlowRuleManager.register2Property(flowRuleRDS.getProperty());WritableDataSource<List<FlowRule>> flowRuleWDS = new FileWritableDataSource<List<FlowRule>>(ruleFilePath, RuleListConverterUtils.flowFuleEnCoding);// 将可写数据源注册至 transport 模块的 WritableDataSourceRegistry 中.// 这样收到控制台推送的规则时，Sentinel 会先更新到内存，然后将规则写入到文件中.WritableDataSourceRegistry.registerFlowDataSource(flowRuleWDS);}private void dealDegradeRules() throws FileNotFoundException {//获取规则文件路径String degradeRuleFilePath = PersistenceRuleConstant.rulesMap.get(PersistenceRuleConstant.DEGRAGE_RULE_PATH).toString();//创建流控规则的可读数据源ReadableDataSource<String, List<DegradeRule>> degradeRuleRDS = new FileRefreshableDataSource(degradeRuleFilePath, RuleListConverterUtils.degradeRuleListParse);// 将可读数据源注册至FlowRuleManager 这样当规则文件发生变化时，就会更新规则到内存DegradeRuleManager.register2Property(degradeRuleRDS.getProperty());WritableDataSource<List<DegradeRule>> degradeRuleWDS = new FileWritableDataSource<>(degradeRuleFilePath, RuleListConverterUtils.degradeRuleEnCoding);// 将可写数据源注册至 transport 模块的 WritableDataSourceRegistry 中.// 这样收到控制台推送的规则时，Sentinel 会先更新到内存，然后将规则写入到文件中.WritableDataSourceRegistry.registerDegradeDataSource(degradeRuleWDS);}private void dealSystemRules() throws FileNotFoundException {//获取规则文件路径String systemRuleFilePath = PersistenceRuleConstant.rulesMap.get(PersistenceRuleConstant.SYSTEM_RULE_PATH).toString();//创建流控规则的可读数据源ReadableDataSource<String, List<SystemRule>> systemRuleRDS = new FileRefreshableDataSource(systemRuleFilePath, RuleListConverterUtils.sysRuleListParse);// 将可读数据源注册至FlowRuleManager 这样当规则文件发生变化时，就会更新规则到内存SystemRuleManager.register2Property(systemRuleRDS.getProperty());WritableDataSource<List<SystemRule>> systemRuleWDS = new FileWritableDataSource<>(systemRuleFilePath, RuleListConverterUtils.sysRuleEnCoding);// 将可写数据源注册至 transport 模块的 WritableDataSourceRegistry 中.// 这样收到控制台推送的规则时，Sentinel 会先更新到内存，然后将规则写入到文件中.WritableDataSourceRegistry.registerSystemDataSource(systemRuleWDS);}private void dealParamFlowRules() throws FileNotFoundException {//获取规则文件路径String paramFlowRuleFilePath = PersistenceRuleConstant.rulesMap.get(PersistenceRuleConstant.HOT_PARAM_RULE).toString();//创建流控规则的可读数据源ReadableDataSource<String, List<ParamFlowRule>> paramFlowRuleRDS = new FileRefreshableDataSource(paramFlowRuleFilePath, RuleListConverterUtils.paramFlowRuleListParse);// 将可读数据源注册至FlowRuleManager 这样当规则文件发生变化时，就会更新规则到内存ParamFlowRuleManager.register2Property(paramFlowRuleRDS.getProperty());WritableDataSource<List<ParamFlowRule>> paramFlowRuleWDS = new FileWritableDataSource<>(paramFlowRuleFilePath, RuleListConverterUtils.paramRuleEnCoding);// 将可写数据源注册至 transport 模块的 WritableDataSourceRegistry 中.// 这样收到控制台推送的规则时，Sentinel 会先更新到内存，然后将规则写入到文件中.ModifyParamFlowRulesCommandHandler.setWritableDataSource(paramFlowRuleWDS);}private void dealAuthRules() throws FileNotFoundException {//获取规则文件路径String authFilePath = PersistenceRuleConstant.rulesMap.get(PersistenceRuleConstant.AUTH_RULE_PATH).toString();//创建流控规则的可读数据源ReadableDataSource<String, List<AuthorityRule>> authRuleRDS = new FileRefreshableDataSource(authFilePath, RuleListConverterUtils.authorityRuleParse);// 将可读数据源注册至FlowRuleManager 这样当规则文件发生变化时，就会更新规则到内存AuthorityRuleManager.register2Property(authRuleRDS.getProperty());//创建流控规则的写数据源WritableDataSource<List<AuthorityRule>> authRuleWDS = new FileWritableDataSource<>(authFilePath, RuleListConverterUtils.authorityEncoding);// 将可写数据源注册至 transport 模块的 WritableDataSourceRegistry 中.// 这样收到控制台推送的规则时，Sentinel 会先更新到内存，然后将规则写入到文件中.WritableDataSourceRegistry.registerAuthorityDataSource(authRuleWDS);}}
package com.tuling.sentinel.extension.filepull;import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;/*** 创建规则持久化目录和文件的工具类**/
public class RuleFileUtils {public static void mkdirIfNotExits(String filePath) throws IOException {File file = new File(filePath);if(!file.exists()) {file.mkdirs();}}public static void createFileIfNotExits(Map<String,String> ruleFileMap) throws IOException {Set<String> ruleFilePathSet = ruleFileMap.keySet();Iterator<String> ruleFilePathIter = ruleFilePathSet.iterator();while (ruleFilePathIter.hasNext()) {String ruleFilePathKey = ruleFilePathIter.next();String ruleFilePath  = PersistenceRuleConstant.rulesMap.get(ruleFilePathKey).toString();File ruleFile = new File(ruleFilePath);if(!ruleFile.exists()) {ruleFile.createNewFile();}}}}

package com.tuling.sentinel.extension.filepull;import java.io.File;
import java.util.HashMap;
import java.util.Map;/*** Sentinel 规则持久化 常量配置类**/
public class PersistenceRuleConstant {/*** 存储文件路径*/public static final String storePath = System.getProperty("user.home") + File.separator + "sentinel" + File.separator + "rules";/*** 各种存储sentinel规则映射map*/public static final Map rulesMap = new HashMap<String,String>();//流控规则文件public static final String FLOW_RULE_PATH = "flowRulePath";//降级规则文件public static final String DEGRAGE_RULE_PATH = "degradeRulePath";//授权规则文件public static final String AUTH_RULE_PATH = "authRulePath";//系统规则文件public static final String SYSTEM_RULE_PATH = "systemRulePath";//热点参数文件public static final String HOT_PARAM_RULE = "hotParamRulePath";static {rulesMap.put(FLOW_RULE_PATH,storePath+ File.separator +"flowRule.json");rulesMap.put(DEGRAGE_RULE_PATH,storePath+File.separator +"degradeRule.json");rulesMap.put(SYSTEM_RULE_PATH,storePath+File.separator +"systemRule.json");rulesMap.put(AUTH_RULE_PATH,storePath+File.separator +"authRule.json");rulesMap.put(HOT_PARAM_RULE,storePath+File.separator +"hotParamRule.json");}
}

package com.tuling.sentinel.extension.filepull;import com.alibaba.csp.sentinel.datasource.Converter;
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRule;
import com.alibaba.csp.sentinel.slots.system.SystemRule;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;import java.util.List;/*** 规则列表解析工具类**/
public class RuleListConverterUtils {public static final Converter<String, List<FlowRule>> flowRuleListParser = new Converter<String, List<FlowRule>>() {@Overridepublic List<FlowRule> convert(String source) {return JSON.parseObject(source, new TypeReference<List<FlowRule>>() {});}};public static final Converter<String,List<DegradeRule>> degradeRuleListParse = new Converter<String, List<DegradeRule>>() {@Overridepublic List<DegradeRule> convert(String source) {return JSON.parseObject(source,new TypeReference<List<DegradeRule>>(){});}};public static final Converter<String,List<SystemRule>> sysRuleListParse = new Converter<String, List<SystemRule>>() {@Overridepublic List<SystemRule> convert(String source) {return JSON.parseObject(source,new TypeReference<List<SystemRule>>(){});}};public static final Converter<String,List<ParamFlowRule>> paramFlowRuleListParse = new Converter<String, List<ParamFlowRule>>() {@Overridepublic List<ParamFlowRule> convert(String source) {return JSON.parseObject(source,new TypeReference<List<ParamFlowRule>>(){});}};public static final Converter<String,List<AuthorityRule>> authorityRuleParse = new Converter<String, List<AuthorityRule>>() {@Overridepublic List<AuthorityRule> convert(String source) {return JSON.parseObject(source,new TypeReference<List<AuthorityRule>>(){});}};public static final Converter<List<FlowRule>,String> flowFuleEnCoding= new Converter<List<FlowRule>,String>() {@Overridepublic String convert(List<FlowRule> source) {return JSON.toJSONString(source);}};public static final Converter<List<SystemRule>,String> sysRuleEnCoding= new Converter<List<SystemRule>,String>() {@Overridepublic String convert(List<SystemRule> source) {return JSON.toJSONString(source);}};public static final Converter<List<DegradeRule>,String> degradeRuleEnCoding= new Converter<List<DegradeRule>,String>() {@Overridepublic String convert(List<DegradeRule> source) {return JSON.toJSONString(source);}};public static final Converter<List<ParamFlowRule>,String> paramRuleEnCoding= new Converter<List<ParamFlowRule>,String>() {@Overridepublic String convert(List<ParamFlowRule> source) {return JSON.toJSONString(source);}};public static final Converter<List<AuthorityRule>,String> authorityEncoding= new Converter<List<AuthorityRule>,String>() {@Overridepublic String convert(List<AuthorityRule> source) {return JSON.toJSONString(source);}};
}



打一个jar之后上传到公司仓库。

在微服务中引入下面的依赖即可

<dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-datasource-extension-file-pull</artifactId><version>1.8.4</version>
</dependency>

http://www.ppmy.cn/server/62269.html

相关文章

HDFS 块重构和RedundancyMonitor详解

HDFS 块重构和RedundancyMonitor详解

文章目录 1. 前言2 故障块的重构(Reconstruct)2.1 故障块的状态定义和各个状态的统计信息2.2 故障文件块的查找收集2.2.1 Mis Replica的检测2.2.2 延迟队列(postponedMisreplicatedBlocks)的构造和实现向postponedMisreplicatedBlocks中添加Block从postponedMisreplicatedBlock…
阅读更多...
P2p网络性能测度及监测系统模型

P2p网络性能测度及监测系统模型

P2p网络性能测度及监测系统模型 网络IP性能参数 IP包传输时延时延变化误差率丢失率虚假率吞吐量可用性连接性测度单向延迟测度单向分组丢失测度往返延迟测度 OSI中的位置-> 网络层 用途 面相业务的网络分布式计算网络游戏IP软件电话流媒体分发多媒体通信 业务质量 通过…
阅读更多...
玩转HarmonyOS NEXT之IM应用首页布局

玩转HarmonyOS NEXT之IM应用首页布局

本文从目前流行的垂类市场中&#xff0c;选择即时通讯应用作为典型案例详细介绍HarmonyOS NEXT的各类布局在实际开发中的综合应用。即时通讯应用的核心功能为用户交互&#xff0c;主要包含对话聊天、通讯录&#xff0c;社交圈等交互功能。 应用首页 创建一个包含一列的栅格布…
阅读更多...
【Oracle】实验三 Oracle数据库的创建和管理

【Oracle】实验三 Oracle数据库的创建和管理

【实验目的】 掌握Oracle数据库的创建方法使用DBCA创建数据库在数据库中装入SCOTT用户及其表 【实验内容】 使用DBCA创建数据库&#xff0c;名为MYDB&#xff0c;找到其初始化文件(文本型和服务器型文件都要找到)&#xff0c;查看各类默认位置并记录下来(包括物理文件所在目…
阅读更多...
Perl高手秘籍：自定义操作符的炼金术

Perl高手秘籍：自定义操作符的炼金术

&#x1f31f; Perl高手秘籍&#xff1a;自定义操作符的炼金术 Perl是一种极其灵活的编程语言&#xff0c;它不仅支持内置的操作符&#xff0c;还允许开发者定义自己的操作符。自定义操作符可以极大地增强Perl代码的表达力和功能性。本文将深入探讨如何在Perl中定义自定义操作…
阅读更多...
超市管理系统 需求分析与设计 UML 方向

超市管理系统 需求分析与设计 UML 方向

一、项目介绍 1.1项目背景 随着经济一体化和电子商务的迅速发展&#xff0c;网络传播信息的速度打破了传统信息传递的模式&#xff0c;互联网的高速发展和计算机应用在各个高校进展迅速&#xff0c;更多信息化产品的突飞猛进&#xff0c;让现代的管理模式也发生了巨大的变化&…
阅读更多...
自然语言处理基本概念

自然语言处理基本概念

自然语言处理基本概念 所有学习循环神经网络的人都是看这一篇博客长大的&#xff1a; https://colah.github.io/posts/2015-08-Understanding-LSTMs/ import jieba import torch from torch import nns1 "我吃饭了&#xff01;" s2 "今天天气很好&#xff01…
阅读更多...
【EI征稿】第四届机器人、自动化与智能控制国际会议

【EI征稿】第四届机器人、自动化与智能控制国际会议

【快速通道】 参会方式&#xff1a;担任会议committee成员、组建workshop 、参会报告、参会交流、审稿专家、投稿参会。 会议地点&#xff1a; 湖南 长沙 会议时间&#xff1a;12月6日-9日 会议检索&#xff1a;EI检索 会议官网&#xff1a;https://www.icraic.org/ 投稿链接&a…
阅读更多...
最新文章

Copyright @ 2022~2023