Spring Cloud Alibaba——Sentinel持久化和集群流控

本文已参与「新人创作礼」活动，一起开启掘金创作之路。

1. 持久化
   之前在Sentinel后台管理界面中配置了一些流量控制规则、降级规则，但只要Sentinel服务重启就全部消失，因此需要将一系列规则持久化存储，实现重启Sentinel应用配置依然存在。

Sentinel提供了这几种持久化方案

储存到文件
使用Redis储存
使用Nacos存储
使用Zookeeper存储
使用Apollo存储
下面将使用同为Spring Cloud Alibaba开发套件中的Naocs作为持久化方案，因为在前期已经将Nacos持久化到MySQL数据库

1.1 导入依赖

<!-- sentinel持久化 访问nacos数据源的依赖 -->
<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-datasource-nacos</artifactId>
</dependency>
1
2
3
4
5

1.2 修改配置

bootstrap.yml

server:
  port: 8081 #程序端口号
spring:
  application:
    name: consumer # 应用名称
  cloud:
    sentinel:
      datasource:
        ds1: # ds1是自己取的名字
          nacos: #表示使用nacos
            server-addr: 127.0.0.1:8848
            dataId: sentinel-consumer-ds1 #nacos dataId
            groupId: DEFAULT_GROUP # 分组 默认分组
            data-type: json # 数据类型json
            rule-type: flow # 表示流控规则
      transport:
        port: 8719 # 启动HTTP Server，并且该服务与Sentinel仪表进行交互，是Sentinel仪表盘可以控制应用，如被占用，则从8719依次+1扫描
        dashboard: 127.0.0.1:8080 # 指定仪表盘地址
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848 # nacos服务注册、发现地址
      config:
        server-addr: 127.0.0.1:8848 # nacos配置中心地址
        file-extension: yml # 指定配置内容的数据格式
management:
  endpoints:
    web:
      exposure:
        include: '*' # 公开所有端点
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

1.3 Nacos配置
（1）测试配置流控规则

在Nacos控制面板新建sentinel-consumer-ds1的dataId的配置

[
    {
        "resource":"/sentinelTestB",
        "limitApp":"default",
        "grade":1,
        "count":1,
        "strategy":0,
        "controlBehavior":0,
        "clusterMode":false
    }
]
1
2
3
4
5
6
7
8
9
10
11
12
13

各字段含义：

resource：表示资源名称
limitApp：表示要限制来自哪些来源的调用，default是全部都限制
grade：表示阈值类型，取值参考RuleConstant类（0-线程数限流、1-QPS限流）
count：表示限流阈值
strategy：表示流控模式（直接、关联、链路）
controlBehavior：表示流控效果（快速失败、Warm Up、排队等待）
clusterMode：是否为集群模式
json字段的来源：

配置流控规则除了可以在Sentinel控制面板配置，还可以在Java代码中配置，比如这里的流控规则对应的类是FlowRule，json数据字段就是它的属性字段

如上配置代表配置流控规则为：来源应用【default】、流控模式【直接】、阈值类型【QPS】、阈值【1】、单机模式、流控效果【快速失败】

在1.2中的配置中，spring.cloud.sentinel.datasource.ds1.nacos.dataId、groupId与Nacos控制面板的DataId、GroupId是一一对应的，并且配置内容中[ ]数组符号也可以看出，这是可以配置多个规则的

使用JMeter使用10个线程请求/sentinelTestB接口，测试是否达到流控效果

测试结果：QPS阈值为1，在发起第二个请求就被阻断了，到Sentinel控制台发现自动增加了一条流控规则，也说明配置生效了

（2）测试配置熔断规则

bootstrap.yml增加配置

ds2: # ds2是自己取的名字
  nacos: #表示使用nacos
    server-addr: 127.0.0.1:8848
    dataId: sentinel-consumer-ds2 #nacos dataId
    groupId: DEFAULT_GROUP # 分组 默认分组
    data-type: json # 数据类型json
    rule-type: degrade # 表示熔断规则
nacos控制面板新增DataId sentinel-consumer-ds2的配置

[
    {
        "resource":"/sentinelTest",
        "count":2,
        "grade":2,
        "timeWindow":5
    }
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

各字段含义：

resource：表示资源名称
count：表示限流阈值
grade：表示阈值类型，取值参考RuleConstant类（0-慢比例调用、1-异常比例、2-异常数）
timeWindow：表示时间窗口
使用JMeter调用/sentinelTest接口（这个接口是一个一定会报错的接口），第一、二次请求抛出算数异常，异常数到达两个之后，也就是从第三个请求开始进入熔断状态，要过了熔断时长5秒后才能恢复。 并且，sentinel控制台已经有了熔断规则

（3）测试配置系统规则

bootstrap.yml增加配置

ds3: # ds3是自己取的名字
  nacos: #表示使用nacos
    server-addr: 127.0.0.1:8848
    dataId: sentinel-consumer-ds3 #nacos dataId
    groupId: DEFAULT_GROUP # 分组 默认分组
    data-type: json # 数据类型json
    rule-type: system # 表示系统规则
在nacos控制面板新增DataId sentinel-consumer-ds3的配置

[
    {
        "qps":"1"
    }
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

使用JMeter使用10个线程一直循环测试/sentinelTestC接口，然后使用curl调用/sentinelTestB接口，被阻断了，其实不止是B接口，其它请求也发生了大量被阻断。 同时，sentinel控制台也同时有了系统规则

有两点需要注意：

在Nacos控制台修改规则，Sentinel规则会即使生效，重启服务后依然有效
在Sentinel控制台修改规则，不会修改到Nacos里的数据，自然数据不是持久的，重启后又会恢复原来的值
2. 集群流控
集群流控是为了解决在服务集群下流量不均匀导致总体限流效果不佳的问题，之前做的都是单机的流控，但随着网站访问的增加，做集群是势在必行的，为了提升载量，会再增加服务提供者。

例如，服务提供者集群有2台机器，设置单机阈值为10QPS，理想状态下整个集群的限流阈值为20QPS，不过实际的流量到具体服务分配不均，导致总量没有到的情况下某些机器就开始限流了，还有可能会超过阈值，可能实际限流QPS=阈值×节点数

这时候就需要集群流控了，集群流控的基本原理就是用server端来专门统计总量，其他client实例都与server端通信来判断是否可以调用，并结合单机限流兜底，发挥更好的流量控制效果。

Token Client（集群流控客户端）与Token Server（集群流控服务端）连接失败或通信失败时，如果勾选了失败退化，会退化到本地的限流模式。

集群流控有两种角色：

Token Server：集群流控服务端，处理来自Token Client的请求，根据配置的集群规则判断运行是否通过
Token Client：集群流控客户端，向Token Server发起请求，根据返回结果判断是否需要限流
实现集群流控服务端（Token Server）有两种方式：

独立模式（alone）：服务独立运行，独立部署，隔离性好
嵌入模式（embedded）：可以时集群流控服务端，也可以是集群流控客户端，无需单独部署，灵活性较好，不过为了不影响服务本身，需要限制QPS。
例如，使用嵌入模式

（1）向消费服务者增加依赖
<!-- 集群流控客户端依赖 -->
<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-cluster-client-default</artifactId>
</dependency>
<!-- 集群流控服务端依赖 -->
<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-cluster-server-default</artifactId>
</dependency>
<!-- gson依赖 -->
<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.9</version>
</dependency>
（2）通过SPI完成配置源注册
定义com.alibaba.csp.sentinel.init.IntiFunc的实现

SPI（Serveice Provider Interface）是一种服务发现机制， 它会查找META-INF/services下的文件，加载文件里所定义的类。

其实配置源注册就是自定义InitFunc的实现类

package com.springcloudalibaba.sample.init;

// ...导包省略

public class ApplicationInitializer implements InitFunc {
    //配置的dataId
    private final String flowDataId = "provider" + Constants.FLOW_POSTFIX;
    private final String paramDataId = "provider" + Constants.PARAM_FLOW_POSTFIX;
    private final String clusterMapDataId = "provider" + Constants.CLUSTER_MAP_POSTFIX;
    
    private static final String SEPARATOR = "@";

    @Override
    public void init() {
        //动态数据源的方式配置sentinel的流量控制和热点参数限流的规则。
        initDynamicRuleProperty();
        //初始化Token客户端
        initClientServerAssignProperty();
        //注册动态规则数据源
        registerClusterRuleSupplier();
        //初始化server的端口配置
        initServerTransportConfigProperty();
        //初始化集群中服务是客户端还是服务端
        initStateProperty();
    }

    private void initDynamicRuleProperty() {
        ReadableDataSource<String, List<FlowRule>> ruleSource = new NacosDataSource<>(remoteAddress, groupId,
                flowDataId, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {
        }));
        FlowRuleManager.register2Property(ruleSource.getProperty());
    }

    private void initServerTransportConfigProperty() {
        ReadableDataSource<String, ServerTransportConfig> serverTransportDs = new NacosDataSource<>(remoteAddress, groupId,
                clusterMapDataId, source -> {
            List<ClusterGroupEntity> groupList = new Gson().fromJson(source, new TypeToken<List<ClusterGroupEntity>>() {
            }.getType());
            return Optional.ofNullable(groupList)
                    .flatMap(this::extractServerTransportConfig)
                    .orElse(null);
        });
        ClusterServerConfigManager.registerServerTransportProperty(serverTransportDs.getProperty());
    }

    private void registerClusterRuleSupplier() {
        ClusterFlowRuleManager.setPropertySupplier(namespace -> {
            ReadableDataSource<String, List<FlowRule>> ds = new NacosDataSource<>(remoteAddress, groupId,
                    namespace + Constants.FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {
            }));
            return ds.getProperty();
        });
        ClusterParamFlowRuleManager.setPropertySupplier(namespace -> {
            ReadableDataSource<String, List<ParamFlowRule>> ds = new NacosDataSource<>(remoteAddress, groupId,
                    namespace + Constants.PARAM_FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference<List<ParamFlowRule>>() {
            }));
            return ds.getProperty();
        });
    }

    private void initClientServerAssignProperty() {
        ReadableDataSource<String, ClusterClientAssignConfig> clientAssignDs = new NacosDataSource<>(remoteAddress, groupId,
                clusterMapDataId, source -> {
            List<ClusterGroupEntity> groupList = new Gson().fromJson(source, new TypeToken<List<ClusterGroupEntity>>() {
            }.getType());
            return Optional.ofNullable(groupList)
                    .flatMap(this::extractClientAssignment)
                    .orElse(null);
        });
        ClusterClientConfigManager.registerServerAssignProperty(clientAssignDs.getProperty());
    }

    private void initStateProperty() {
        ReadableDataSource<String, Integer> clusterModeDs = new NacosDataSource<>(remoteAddress, groupId,
                clusterMapDataId, source -> {
            List<ClusterGroupEntity> groupList = new Gson().fromJson(source, new TypeToken<List<ClusterGroupEntity>>() {
            }.getType());
            return Optional.ofNullable(groupList)
                    .map(this::extractMode)
                    .orElse(ClusterStateManager.CLUSTER_NOT_STARTED);
        });
        ClusterStateManager.registerProperty(clusterModeDs.getProperty());
    }

    private int extractMode(List<ClusterGroupEntity> groupList) {
        if (groupList.stream().anyMatch(this::machineEqual)) {
            return ClusterStateManager.CLUSTER_SERVER;
        }
        boolean canBeClient = groupList.stream()
                .flatMap(e -> e.getClientSet().stream())
                .filter(Objects::nonNull)
                .anyMatch(e -> e.equals(getCurrentMachineId()));
        return canBeClient ? ClusterStateManager.CLUSTER_CLIENT : ClusterStateManager.CLUSTER_NOT_STARTED;
    }

    private Optional<ServerTransportConfig> extractServerTransportConfig(List<ClusterGroupEntity> groupList) {
        return groupList.stream()
                .filter(this::machineEqual)
                .findAny()
                .map(e -> new ServerTransportConfig().setPort(e.getPort()).setIdleSeconds(600));
    }

    private Optional<ClusterClientAssignConfig> extractClientAssignment(List<ClusterGroupEntity> groupList) {
        if (groupList.stream().anyMatch(this::machineEqual)) {
            return Optional.empty();
        }
        for (ClusterGroupEntity group : groupList) {
            if (group.getClientSet().contains(getCurrentMachineId())) {
                String ip = group.getIp();
                Integer port = group.getPort();
                return Optional.of(new ClusterClientAssignConfig(ip, port));
            }
        }
        return Optional.empty();
    }

    private boolean machineEqual(/*@Valid*/ ClusterGroupEntity group) {
        return getCurrentMachineId().equals(group.getMachineId());
    }

    private String getCurrentMachineId() {
        return HostNameUtil.getIp() + SEPARATOR + TransportConfig.getRuntimePort();
    }
}
相关类

@Data
public class ClusterGroupEntity implements Serializable {
    private String machineId; // 机器id
    private String ip; //ip 地址
    private Integer port; // 端口
    private Set<String> clientSet;
}
public class Constants {
    public static final String FLOW_POSTFIX = "-flow-rules";
    public static final String PARAM_FLOW_POSTFIX = "-param-rules";
    public static final String CLUSTER_MAP_POSTFIX = "-cluster-map";
}
resources下新增META-INF/services/com.alibaba.csp.sentinel.init.InitFunc文件，指定ApplicationInitializer为实现类

com.springcloudalibaba.sample.init.ApplicationInitializer
（3）服务提供者再增加一个节点
复制一份服务提供者并修改端口，模拟2个节点的集群

（4）Nacos配置
创建一条新的DataId数据

provider-flow-rules对应的就是ApplicationInitializer类中flowDataId变量的值

[
    {
        "resource" : "/test",     // 限流的资源名称
        "grade" : 1,                         // 限流模式为：qps，线程数限流0，qps限流1
        "count" : 100,                        // 阈值为：100
        "clusterMode" :  true,               // 是否是集群模式，集群模式为：true
        "clusterConfig" : {
            "flowId" : 111,                  // 全局唯一id
            "thresholdType" : 1,             // 阈值模式为：全局阈值，0是单机均摊，1是全局阀值
            "fallbackToLocalWhenFail" : true // 在 client 连接失败或通信失败时，是否退化到本地的限流模式
        }
    }
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186

（5）启动2个provider和1个consumer
请求一次他们各自的接口，然后刷新sentinel控制面板，就会出现可以针对接口操作的选项

（6）测试
如果nacos持久化规则配置成功后，会出现/test资源的流控规则

使用JMeter 2个线程一直循环调用服务消费者的test接口，服务消费者再调用服务提供者的test接口，再sentinel控制台实时监控的结果如图，整个集群的QPS为100，跟上述配置是一样的