解决sentinel结合nacos实现集群限流（嵌入式）

本文将为您介绍什么是集群流控，如何使用阿里的开源项目 Sentinel实现动态配置的集群流控的详细内容，以下是详情内容，希望我们能共同学习进步

Why？

为什么要用集群流控？
相对于单机流控而言，我们给每台机器设置单机限流阈值，在理想情况下整个集群的限流阈值为机器数量 * 单机阈值。不过实际情况下流量到每台机器可能会不均匀，会导致总量没有到的情况下某些机器就开始限流。因此仅靠单机维度去限制的话会无法精确地限制总体流量。而集群流控可以精确地控制整个集群的调用总量，结合单机限流兜底，可以更好地发挥流量控制的效果。

部署方式

通过sentinel官网可知，集群限流分为嵌入式与独立式。此文档主要介绍嵌入式。

话不多说，开整！！！

POM引入

        <!--      sentinel    start   -->
        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-spring-cloud-gateway-adapter</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-transport-simple-http</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-datasource-nacos</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
            <version>2.2.5.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-cluster-server-default</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-cluster-client-default</artifactId>
        </dependency>
        <!--      sentinel    end   -->
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

配置文件（根据自己实际地址对应修改即可）

# 服务启动名称
spring.application.name=gxf-localhost
# 服务启动端口
server.port=8077
###  nacos config
# nacos 配置中心连接地址
spring.cloud.nacos.config.server-addr=localhost:8848
# nacos 配置中心 namespace
spring.cloud.nacos.config.namespace=1111-2222-3333-444
spring.cloud.sentinel.transport.dashboard=localhost:9010
spring.cloud.sentinel.transport.port=8719
spring.cloud.sentinel.transport.clientIp=localhost
spring.cloud.sentinel.eager=true
1
2
3
4
5
6
7
8
9
10
11
12
13

新建常量类

/**
 * @author Gxf
 * @className: SentinelConstants
 * @description: 集群限流常量类
 * @date 11/4/22 4:39 PM
 * @version：1.0
 */
public class SentinelConstants {

    /** nacos 地址配置 */
    public static String SERVER_ADDR = "localhost:8848";

    /** nacos 限流默认分组 */
    public static final String SENTINEL_GREOUP_ID = "DEFAULT_GROUP";

    /** url 限流规则名称定义 */
    public static final String URL_FLOW_POSTFIX = "location-url-sentinel";

    /** 集群限流配置信息 名称定义*/
    public static final String CLUSTER_MAP_POSTFIX = "location-cluster-map";

    /** 集群限流配置 */
    public static final String CLIENT_CONFIG = "location-cluster-client-config";

    private SentinelConstants() {}

}
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

新建转换类

import java.io.Serializable;
import java.util.Set;

/**
 * @author Gxf
 * @className: ClusterGroupEntity
 * @description: 集群限流实体类（nacos读取配置信息转sentinel规则）
 * @date 11/4/22 4:36 PM
 * @version：1.0
 */
public class ClusterGroupEntity implements Serializable {

    private String machineId;
    private String ip;
    private Integer port;

    private Set<String> clientSet;

    public String getMachineId() {
        return machineId;
    }

    public void setMachineId(String machineId) {
        this.machineId = machineId;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public Set<String> getClientSet() {
        return clientSet;
    }

    public void setClientSet(Set<String> clientSet) {
        this.clientSet = clientSet;
    }


    @Override
    public String toString() {
        return "ClusterGroupEntity{" +
                "machineId='" + machineId + '\'' +
                ", ip='" + ip + '\'' +
                ", port=" + port +
                ", clientSet=" + clientSet +
                '}';
    }
}
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

新建初始化sentinel配置信息类

因我这里是网关项目，所以热点参数舍弃不使用

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;

import com.alibaba.csp.sentinel.cluster.ClusterStateManager;
import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientAssignConfig;
import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientConfig;
import com.alibaba.csp.sentinel.cluster.client.config.ClusterClientConfigManager;
import com.alibaba.csp.sentinel.cluster.flow.rule.ClusterFlowRuleManager;
import com.alibaba.csp.sentinel.cluster.server.config.ClusterServerConfigManager;
import com.alibaba.csp.sentinel.cluster.server.config.ServerTransportConfig;
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
import com.alibaba.csp.sentinel.datasource.nacos.NacosDataSource;
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.config.TransportConfig;
import com.alibaba.csp.sentinel.util.HostNameUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.constant.SentinelConstants;
import com.entity.ClusterGroupEntity;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * sentinel 动态配置 初始化类
 * @author Gxf
 */
public class ClusterInitFunc implements InitFunc {


    /**
     * 日志
     */
    private static final Logger log = LogManager.getLogger(ClusterInitFunc.class);

    private static final String SEPARATOR = "@";

    private static Properties properties = new Properties();
    static {
        //todo 写成配置，对应修改各项参数
        properties.put("serverAddr", SentinelConstants.SERVER_ADDR);
        properties.put("namespace", "1111");
        properties.put("username","nacos");
        properties.put("password","nacos");
    }


    @Override
    public void init() {

        log.info("start init sentinel --------");

        // Register client dynamic rule data source.
        //动态数据源的方式配置sentinel的流量控制和热点参数限流的规则。
        initDynamicRuleProperty();

        // Register token client related data source.
        // Token client common config
        // 集群限流客户端的配置属性
        initClientConfigProperty();
        // Token client assign config (e.g. target token server) retrieved from assign map:
        //初始化Token客户端
        initClientServerAssignProperty();

        // Register token server related data source.
        // Register dynamic rule data source supplier for token server:
        //集群的流控规则，比如限制整个集群的流控阀值,启动的时候需要添加-Dproject.name=项目名
        registerClusterRuleSupplier();
        // Token server transport config extracted from assign map:
        //初始化server的端口配置
        initServerTransportConfigProperty();

        // Init cluster state property for extracting mode from cluster map data source.
        //初始化集群中服务是客户端还是服务端
        initStateProperty();
    }

    private void initDynamicRuleProperty() {

        log.info("初始化url限流规则,DataId={}",SentinelConstants.URL_FLOW_POSTFIX);
        //url限流规则的DataId是 sdmk-gateway-periplaneta-url-sentinel
        ReadableDataSource<String, List<FlowRule>> ruleSource = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                SentinelConstants.URL_FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));
        FlowRuleManager.register2Property(ruleSource.getProperty());

        //热点参数限流规则的DataId是 sdmk-gateway-periplaneta-parameter-sentinel;
        //TODO 热点参数网关不需要
//        log.info("初始化热点限流规则,DataId={}",SentinelConstants.PARAM_FLOW_POSTFIX);
//        ReadableDataSource> paramRuleSource = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
//                SentinelConstants.PARAM_FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference>() {}));
//        ParamFlowRuleManager.register2Property(paramRuleSource.getProperty());
    }

    private void initClientConfigProperty() {
        log.info("配置客户端");
        ReadableDataSource<String, ClusterClientConfig> clientConfigDs = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                SentinelConstants.CLIENT_CONFIG, source -> JSON.parseObject(source, new TypeReference<ClusterClientConfig>() {}));
        ClusterClientConfigManager.registerClientConfigProperty(clientConfigDs.getProperty());
    }

    private void initServerTransportConfigProperty() {
        log.info("初始化server端口配置");
        ReadableDataSource<String, ServerTransportConfig> serverTransportDs = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                SentinelConstants.CLUSTER_MAP_POSTFIX, 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() {
        // Register cluster flow rule property supplier which creates data source by namespace.
        // Flow rule dataId format: ${namespace}-flow-rules
        ClusterFlowRuleManager.setPropertySupplier(namespace -> {
            log.info("配置集群的url流控规则");
            ReadableDataSource<String, List<FlowRule>> ds = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                    SentinelConstants.URL_FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));
            return ds.getProperty();
        });

        // Register cluster parameter flow rule property supplier which creates data source by namespace.
        // TODO 热点参数网关不需要
//        ClusterParamFlowRuleManager.setPropertySupplier(namespace -> {
//            log.info("配置集群的热点参数流控规则");
//            ReadableDataSource> ds = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
//                    SentinelConstants.PARAM_FLOW_POSTFIX, source -> JSON.parseObject(source, new TypeReference>() {}));
//            return ds.getProperty();
//        });
    }

    private void initClientServerAssignProperty() {
        // Cluster map format:
        // [{"clientSet":["112.12.88.66@8729","112.12.88.67@8727"],"ip":"112.12.88.68","serverId":"112.12.88.68@8728","port":11111}]
        // serverId: , commandPort for port exposed to Sentinel dashboard (transport module)
        log.info("初始化Token客户端");
        ReadableDataSource<String, ClusterClientAssignConfig> clientAssignDs = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                SentinelConstants.CLUSTER_MAP_POSTFIX, 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() {
        // Cluster map format:
        // [{"clientSet":["112.12.88.66@8729","112.12.88.67@8727"],"ip":"112.12.88.68","serverId":"112.12.88.68@8728","port":11111}]
        // serverId: , commandPort for port exposed to Sentinel dashboard (transport module)
        log.info("初始化判断为客户端还是服务端");
        ReadableDataSource<String, Integer> clusterModeDs = new NacosDataSource<>(properties, SentinelConstants.SENTINEL_GREOUP_ID,
                SentinelConstants.CLUSTER_MAP_POSTFIX, 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 any server group serverId matches current, then it's token server.
        if (groupList.stream().anyMatch(this::machineEqual)) {
            return ClusterStateManager.CLUSTER_SERVER;
        }
        // If current machine belongs to any of the token server group, then it's token client.
        // Otherwise it's unassigned, should be set to NOT_STARTED.
        boolean canBeClient = groupList.stream()
                .flatMap(e -> e.getClientSet().stream())
                .filter(Objects::nonNull)
                .anyMatch(e -> e.equals(getCurrentMachineId()));
        log.info("为客户端还是服务端={}",canBeClient ? ClusterStateManager.CLUSTER_CLIENT : ClusterStateManager.CLUSTER_NOT_STARTED);
        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();
        }
        // Build client assign config from the client set of target server group.
        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() {
        // Note: this may not work well for container-based env.
        log.info("sentinel-ip:port={}",HostNameUtil.getIp() + SEPARATOR + TransportConfig.getRuntimePort());
        return HostNameUtil.getIp() + SEPARATOR + TransportConfig.getRuntimePort();
    }
}
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214

在项目启动类中添加初始化代码

需要放在SpringApplication.run(GatewayApplication.class, args);下面，不然会注册不到sentinel上

    public static void main(String[] args) {
        SpringApplication.run(GatewayApplication.class, args);
        try {
            // 这里我们要主动加载ClusterInitFunc.init()方法
            // 也可以通过SPI的方式加载
            new ClusterInitFunc().init();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
1
2
3
4
5
6
7
8
9
10

手动在nacos添加以下配置信息

在Nacos中添加动态规则配置，以及token server与token client的配置（需注意要与sentinel常量类中的值相同即可）

DataId：location-url-sentinel
Group：DEFAULT_GROUP
配置内容（json格式）
添加配置，请去除注释内容。

[
    {
        "resource" : "/test/{id}",     // 限流的资源名称
        "grade" : 1,                         // 限流模式为：qps，线程数限流0，qps限流1
        "count" : 10,                        // 阈值为：10
        "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

DataId：location-cluster-map
Group：DEFAULT_GROUP
配置内容（json格式）
添加配置，请去除注释内容。

[{
	"clientSet": ["127.0.0.1@8721", "127.0.0.1@8722"],
	"ip": "127.0.0.1",
	"serverId": "127.0.0.1@8720",
	"port": 18730   //这个端口是token server通信的端口
}]
1
2
3
4
5
6

DataId：location-cluster-client-config
Group：DEFAULT_GROUP
配置内容（json格式）

{
    "requestTimeout": 20
}
1
2
3

再本地分别启动三台机器，可以在sentinel控制台看到集群信息与流控管理；


在通过jmeter进行三个端口压测，观测实时监控，发现通过QPS与设置一致。

我完事了，你呢？