关于springboot:从99打造Sentinel高可用集群限流中间件

35次阅读

共计 9204 个字符，预计需要花费 24 分钟才能阅读完成。

接上篇 Sentinel 集群限流摸索，上次简略提到了集群限流的原理，而后用官网给的 demo 简略批改了一下，能够失常运行失效。

这一次须要更进一步，基于 Sentinel 实现内嵌式集群限流的高可用计划，并且包装成一个中间件 starter 提供给三方应用。

对于高可用，咱们次要须要解决两个问题，这无论是应用内嵌或者独立模式都须要解决的问题，相比而言，内嵌式模式更简略一点。

  1. 集群 server 主动选举
  2. 主动故障转移
  3. Sentinel-Dashboard 长久化到 Apollo

集群限流

首先，思考到大部分的服务可能都不须要集群限流这个性能，因而实现一个注解用于手动开启集群限流模式，只有开启注解的状况下，才去实例化集群限流的 Bean 和限流数据。

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Import({EnableClusterImportSelector.class})
@Documented
public @interface SentinelCluster {
}

public class EnableClusterImportSelector implements DeferredImportSelector {
    @Override
    public String[] selectImports(AnnotationMetadata annotationMetadata) {return new String[]{ClusterConfiguration.class.getName()};
    }
}

这样写好之后，当扫描到有咱们的 SentinelCluster 注解的时候，就会去实例化 ClusterConfiguration。

@Slf4j
public class ClusterConfiguration implements BeanDefinitionRegistryPostProcessor, EnvironmentAware {
    private Environment environment;

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ClusterManager.class);
        beanDefinitionBuilder.addConstructorArgValue(this.environment);
        registry.registerBeanDefinition("clusterManager", beanDefinitionBuilder.getBeanDefinition());
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { }

    @Override
    public void setEnvironment(Environment environment) {this.environment = environment;}
}

在配置中去实例化用于治理集群限流的 ClusterManager，这段逻辑和咱们之前文章中应用到的一般无二，注册到ApolloDataSource 之后主动监听 Apollo 的变动达到动静失效的成果。

@Slf4j
public class ClusterManager {
    private Environment environment;
    private String namespace;
    private static final String CLUSTER_SERVER_KEY = "sentinel.cluster.server"; // 服务集群配置
    private static final String DEFAULT_RULE_VALUE = "[]"; // 集群默认规定
    private static final String FLOW_RULE_KEY = "sentinel.flow.rules"; // 限流规定
    private static final String DEGRADE_RULE_KEY = "sentinel.degrade.rules"; // 降级规定
    private static final String PARAM_FLOW_RULE_KEY = "sentinel.param.rules"; // 热点限流规定
    private static final String CLUSTER_CLIENT_CONFIG_KEY = "sentinel.client.config"; // 客户端配置

    public ClusterManager(Environment environment) {
        this.environment = environment;
        this.namespace = "YourNamespace";
        init();}

    private void init() {initClientConfig();
        initClientServerAssign();
        registerRuleSupplier();
        initServerTransportConfig();
        initState();}

    private void initClientConfig() {
        ReadableDataSource<String, ClusterClientConfig> clientConfigDs = new ApolloDataSource<>(
                namespace,
                CLUSTER_CLIENT_CONFIG_KEY,
                DEFAULT_SERVER_VALUE,
                source -> JacksonUtil.from(source, ClusterClientConfig.class)
        );
        ClusterClientConfigManager.registerClientConfigProperty(clientConfigDs.getProperty());
    }

    private void initClientServerAssign() {
        ReadableDataSource<String, ClusterClientAssignConfig> clientAssignDs = new ApolloDataSource<>(
                namespace,
                CLUSTER_SERVER_KEY,
                DEFAULT_SERVER_VALUE,
                new ServerAssignConverter(environment)
        );
        ClusterClientConfigManager.registerServerAssignProperty(clientAssignDs.getProperty());
    }

    private void registerRuleSupplier() {
        ClusterFlowRuleManager.setPropertySupplier(ns -> {
            ReadableDataSource<String, List<FlowRule>> ds = new ApolloDataSource<>(
                    namespace,
                    FLOW_RULE_KEY,
                    DEFAULT_RULE_VALUE,
                    source -> JacksonUtil.fromList(source, FlowRule.class));
            return ds.getProperty();});
        ClusterParamFlowRuleManager.setPropertySupplier(ns -> {
            ReadableDataSource<String, List<ParamFlowRule>> ds = new ApolloDataSource<>(
                    namespace,
                    PARAM_FLOW_RULE_KEY,
                    DEFAULT_RULE_VALUE,
                    source -> JacksonUtil.fromList(source, ParamFlowRule.class)
            );
            return ds.getProperty();});
    }

    private void initServerTransportConfig() {
        ReadableDataSource<String, ServerTransportConfig> serverTransportDs = new ApolloDataSource<>(
                namespace,
                CLUSTER_SERVER_KEY,
                DEFAULT_SERVER_VALUE,
                new ServerTransportConverter(environment)
        );

        ClusterServerConfigManager.registerServerTransportProperty(serverTransportDs.getProperty());
    }

    private void initState() {
        ReadableDataSource<String, Integer> clusterModeDs = new ApolloDataSource<>(
                namespace,
                CLUSTER_SERVER_KEY,
                DEFAULT_SERVER_VALUE,
                new ServerStateConverter(environment)
        );

        ClusterStateManager.registerProperty(clusterModeDs.getProperty());
    }
}

这样的话，一个集群限流的基本功能曾经差不多是 OK 了，上述步骤都比较简单，依照官网文档根本都能跑起来，接下来要实现文章结尾提及到的外围的几个性能了。

主动选举 & 故障转移

主动选举怎么实现？简略点，不必思考那么多，每台机器启动胜利之后间接写入到 Apollo 当中，第一个写入胜利的就是 Server 节点。

这个过程为了保障并发带来的问题，咱们须要加锁确保只有一台机器胜利写入本人的本机信息。

因为我应用 Eureka 作为注册核心，Eureka 又有 CacheRefreshedEvent 本地缓存刷新的事件，基于此每当本地缓存刷新，咱们就去检测以后 Server 节点是否存在，而后依据理论状况去实现选举。

首先在 spring.factories 中增加咱们的监听器。

org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.test.config.SentinelEurekaEventListener

监听器只有当开启了集群限流注解 SentinelCluster 之后才会失效。

@Configuration
@Slf4j
@ConditionalOnBean(annotation = SentinelCluster.class)
public class SentinelEurekaEventListener implements ApplicationListener<CacheRefreshedEvent> {
    @Resource
    private DiscoveryClient discoveryClient;
    @Resource
    private Environment environment;
    @Resource
    private ApolloManager apolloManager;

    @Override
    public void onApplicationEvent(EurekaClientLocalCacheRefreshedEvent event) {if (!leaderAlive(loadEureka(), loadApollo())) {
            boolean tryLockResult = redis.lock; //redis 或者其余加分布式锁
            if (tryLockResult) {
                try {flush();
                } catch (Exception e) { } finally {unlock();
                }
            }
        }
    }
  
    private boolean leaderAlive(List<ClusterGroup> eurekaList, ClusterGroup server) {if (Objects.isNull(server)) {return false;}
        for (ClusterGroup clusterGroup : eurekaList) {if (clusterGroup.getMachineId().equals(server.getMachineId())) {return true;}
        }
        return false;
    }
}

OK，其实看到代码曾经晓得咱们把故障转移的逻辑也实现了，其实情理是一样的。

第一次启动的时候 Apollo 中的 server 信息是空的，所以第一台加锁写入的机器就是 server 节点，后续如果 server 宕机下线，本地注册表缓存刷新，比照 Eureka 的实例信息和 Apollo 中的 server，如果 server 不存在，那么就从新执行选举的逻辑。

须要留神的是，本地缓存刷新的工夫极其状况下可能会达到几分钟级别，那么也就是说在服务下线的可能几分钟内没有从新选举出新的 server 节点整个集群限流是不可用的状态，对于业务要求十分严格的状况这个计划就不太实用了。

对于 Eureka 缓存工夫同步的问题，能够参考之前的文章 Eureka 服务下线太慢，电话被告警打爆了。

Dashboard 长久化革新

到这儿为止，咱们曾经把高可用计划实现好了，接下来最初一步，只有通过 Sentinel 自带的控制台可能把配置写入到 Apollo 中，那么利用就天然会监听到配置的变动，达到动静失效的成果。

依据官网的形容，官网曾经实现了 FlowControllerV2 用于集群限流，同时在测试目录下有简略的案例帮忙咱们疾速实现控制台的长久化的逻辑。

咱们只有实现 DynamicRuleProvider，同时注入到Controller 中应用即可，这里咱们实现 flowRuleApolloProvider 用于提供从 Apollo 查问数据，flowRuleApolloPublisher用于写入限流配置到 Apollo。

@RestController
@RequestMapping(value = "/v2/flow")
public class FlowControllerV2 {private final Logger logger = LoggerFactory.getLogger(FlowControllerV2.class);

    @Autowired
    private InMemoryRuleRepositoryAdapter<FlowRuleEntity> repository;

    @Autowired
    @Qualifier("flowRuleApolloProvider")
    private DynamicRuleProvider<List<FlowRuleEntity>> ruleProvider;
    @Autowired
    @Qualifier("flowRuleApolloPublisher")
    private DynamicRulePublisher<List<FlowRuleEntity>> rulePublisher;


}

实现形式很简略，provider 通过 Apollo 的 open-api 从 namespace 中读取配置，publisher 则是通过 open-api 写入规定。

@Component("flowRuleApolloProvider")
public class FlowRuleApolloProvider implements DynamicRuleProvider<List<FlowRuleEntity>> {

    @Autowired
    private ApolloManager apolloManager;
    @Autowired
    private Converter<String, List<FlowRuleEntity>> converter;

    @Override
    public List<FlowRuleEntity> getRules(String appName) {String rules = apolloManager.loadNamespaceRuleList(appName, ApolloManager.FLOW_RULES_KEY);

        if (StringUtil.isEmpty(rules)) {return new ArrayList<>();
        }
        return converter.convert(rules);
    }
}

@Component("flowRuleApolloPublisher")
public class FlowRuleApolloPublisher implements DynamicRulePublisher<List<FlowRuleEntity>> {

    @Autowired
    private ApolloManager apolloManager;
    @Autowired
    private Converter<List<FlowRuleEntity>, String> converter;

    @Override
    public void publish(String app, List<FlowRuleEntity> rules) {AssertUtil.notEmpty(app, "app name cannot be empty");
        if (rules == null) {return;}
        apolloManager.writeAndPublish(app, ApolloManager.FLOW_RULES_KEY, converter.convert(rules));
    }
}

ApolloManager实现了通过 open-api 查问和写入配置的能力，应用须要自行配置 Apollo Portal 地址和 token，这里不赘述，能够自行查看 Apollo 的官网文档。

@Component
public class ApolloManager {
    private static final String APOLLO_USERNAME = "apollo";
    public static final String FLOW_RULES_KEY = "sentinel.flow.rules";
    public static final String DEGRADE_RULES_KEY = "sentinel.degrade.rules";
    public static final String PARAM_FLOW_RULES_KEY = "sentinel.param.rules";
    public static final String APP_NAME = "YourAppName";

    @Value("${apollo.portal.url}")
    private String portalUrl;
    @Value("${apollo.portal.token}")
    private String portalToken;
    private String apolloEnv;
    private String apolloCluster = "default";
    private ApolloOpenApiClient client;

    @PostConstruct
    public void init() {this.client = ApolloOpenApiClient.newBuilder()
                .withPortalUrl(portalUrl)
                .withToken(portalToken)
                .build();
        this.apolloEnv = "default";
    }

    public String loadNamespaceRuleList(String appName, String ruleKey) {OpenNamespaceDTO openNamespaceDTO = client.getNamespace(APP_NAME, apolloEnv, apolloCluster, "default");
        return openNamespaceDTO
                .getItems()
                .stream()
                .filter(p -> p.getKey().equals(ruleKey))
                .map(OpenItemDTO::getValue)
                .findFirst()
                .orElse("");
    }

    public void writeAndPublish(String appName, String ruleKey, String value) {OpenItemDTO openItemDTO = new OpenItemDTO();
        openItemDTO.setKey(ruleKey);
        openItemDTO.setValue(value);
        openItemDTO.setComment("Add Sentinel Config");
        openItemDTO.setDataChangeCreatedBy(APOLLO_USERNAME);
        openItemDTO.setDataChangeLastModifiedBy(APOLLO_USERNAME);
        client.createOrUpdateItem(APP_NAME, apolloEnv, apolloCluster, "default", openItemDTO);

        NamespaceReleaseDTO namespaceReleaseDTO = new NamespaceReleaseDTO();
        namespaceReleaseDTO.setEmergencyPublish(true);
        namespaceReleaseDTO.setReleasedBy(APOLLO_USERNAME);
        namespaceReleaseDTO.setReleaseTitle("Add Sentinel Config Release");
        client.publishNamespace(APP_NAME, apolloEnv, apolloCluster, "default", namespaceReleaseDTO);
    }

}

对于其余规定，比方降级、热点限流都能够参考此形式去批改，当然控制台要做的批改必定不是这一点点，比方集群的 flowId 默认应用的单机自增，这个必定须要批改，还有页面的传参、查问路由的批改等等，比拟繁琐，就不在此赘述了，总归也就是工作量的问题。

好了，本期内容就这些，我是艾小仙，咱们下期见。

正文完
springboot
发表至： springboot
2022-08-04
 0
关于springboot:SpringBoot2-整合OAuth2组件模拟第三方授权访问
关于springboot:Spring-Cache缓存框架
关于springboot:基于springbootvue前后端分离的activiti工作流后台管理系统
关于springboot:初步对SSM框架中Dao层Mapper层service等层的理解