文章目录

• | 工作原理
• || push模式工作原理
• • 控制台配置规则
  • 应用程序监听规则
• ||| 源码分析
• • A 控制台
  • • 控制台配置
    • 控制台推送
    • • apiAddFlowRule
      • publishRules
  • B 客户端程序
  • • 客户端改造
    • 客户端监听与更新
    • • 更新实现
• 总结

| 工作原理

• 应用程序启动时需配置控制台地址,使得应用程序与控制台进行交互
• 应用程序负责向控制台注册
• 控制台负责配置规则信息
• 应用程序通过监听等方式拉取rule到本地,等待流量到达依据配置进行限流

|| push模式工作原理

控制台配置规则

应用程序监听规则

||| 源码分析

A 控制台

• 控制台模块是一个轻量级springboot模块
• 用户通过FlowControllerV2配置规则并且推送zk等配置中心
  

控制台配置

• 按照类型修改DynamicRuleProvider和DynamicRulePublisher为zk,nacos等

@RestController
@RequestMapping(value = "/v2/flow")
public class FlowControllerV2 {

    private final Logger logger = LoggerFactory.getLogger(FlowControllerV2.class);

    @Autowired
    private InMemoryRuleRepositoryAdapter<FlowRuleEntity> repository;

    @Autowired
    @Qualifier("flowRuleDefaultProvider")
    private DynamicRuleProvider<List<FlowRuleEntity>> ruleProvider;
    @Autowired
    @Qualifier("flowRuleDefaultPublisher")
    private DynamicRulePublisher<List<FlowRuleEntity>> rulePublisher;
}

控制台推送

apiAddFlowRule

• 持久化于内存
• 推送zk

 @PostMapping("/rule")
 @AuthAction(value = AuthService.PrivilegeType.WRITE_RULE)
 public Result<FlowRuleEntity> apiAddFlowRule(@RequestBody FlowRuleEntity entity) {

     Result<FlowRuleEntity> checkResult = checkEntityInternal(entity);
     if (checkResult != null) {
         return checkResult;
     }
     entity.setId(null);
     Date date = new Date();
     entity.setGmtCreate(date);
     entity.setGmtModified(date);
     entity.setLimitApp(entity.getLimitApp().trim());
     entity.setResource(entity.getResource().trim());
     try {
         持久化于内存
         entity = repository.save(entity);
         推送zk
         publishRules(entity.getApp());
     } catch (Throwable throwable) {
         logger.error("Failed to add flow rule", throwable);
         return Result.ofThrowable(-1, throwable);
     }
     return Result.ofSuccess(entity);
 }

publishRules

• 获取所有的规则并且推送zk

private void publishRules(/*@NonNull*/ String app) throws Exception {
     List<FlowRuleEntity> rules = repository.findAllByApp(app);
     // FlowRuleApiPublisher
     // FlowRuleApolloPublisher
     // FlowRuleNacosPublisher
     // FlowRuleZookeeperPublisher
     rulePublisher.publish(app, rules);
 }

B 客户端程序

客户端改造

• Sentinel 针对 ZooKeeper 作了相应适配，底层可以采用 ZooKeeper 作为规则配置数据源。使用时只需添加以下依赖：

<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-datasource-zookeeper</artifactId>
    <version>x.y.z</version>
</dependency>

• 然后创建 ZookeeperDataSource 并将其注册至对应的 RuleManager 上即可

// remoteAddress 代表 ZooKeeper 服务端的地址
// path 对应 ZK 中的数据路径
ReadableDataSource<String, List<FlowRule>> flowRuleDataSource = new ZookeeperDataSource<>(remoteAddress, path, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));
FlowRuleManager.register2Property(flowRuleDataSource.getProperty());

客户端监听与更新

• 通过zklistener 监听规则变更
• 全量覆盖更新规则

public class ZookeeperDataSource<T> extends AbstractDataSource<String, T> {
    ...... 删除大量代码
    private void initZookeeperListener(final String serverAddr, final List<AuthInfo> authInfos) {
        ...... 删除大量代码
       this.listener = new NodeCacheListener() {
            @Override
            public void nodeChanged() {
                zk变更
                try {
                    newValue也就是配置的rule信息 比如限流 熔断 等
                    T newValue = loadConfig();
                    RecordLog.info(String.format("[ZookeeperDataSource] New property value received for (%s, %s): %s",
                            serverAddr, path, newValue));
                    
                    更新时同时更新ruleManager的相关规则信息
                    Property对应也存在与ruleManager中
                    getProperty().updateValue(newValue);
                } catch (Exception ex) {
                    RecordLog.warn("[ZookeeperDataSource] loadConfig exception", ex);
                }
            }
        };
    }
}

更新实现

• Property注册到FlowRuleManager

FlowRuleManager.register2Property(flowRuleDataSource.getProperty());

• 修改currentProperty规则信息为zk监听器的property

 public class FlowRuleManager {
    限流规则信息
    private static SentinelProperty<List<FlowRule>> currentProperty = new DynamicSentinelProperty<List<FlowRule>>();

    public static void register2Property(SentinelProperty<List<FlowRule>> property) {
    AssertUtil.notNull(property, "property cannot be null");
    synchronized (LISTENER) {
        currentProperty.removeListener(LISTENER);
        property.addListener(LISTENER);
        修改currentProperty规则信息为zk监听器的property
        currentProperty = property;
    }
}   

总结

• 生成环境一般采用注册中心保障规则持久化,不会因为宕机丢失
• 本文概述push模式的实现,sentinel还有pull,内存直接推送模式