Sentinel之集群限流源码分析（一）

集群限流源码分析（一）

• sentinel-cluster-common-default: 公共模块，包含公共接口和实体
• sentinel-cluster-client-default: 默认集群流控 client 模块，使用 Netty 进行通信，提供接口方便序列化协议扩展
• sentinel-cluster-server-default: 默认集群流控 server 模块，使用 Netty 进行通信，提供接口方便序列化协议扩展；
  同时提供扩展接口对接规则判断的具体实现（TokenService），默认实现是复用 sentinel-core 的相关逻辑

sentinel-cluster-common-default

源码内容如下：

image.png

可以看到该模块下主要定义公共的注解、接口公共方法。

codec包

数据流接口

该包下定义EntityWriter和EntityDecoder两个顶层接口。

• EntityWriter:用于对目标数据的序列化

public interface EntityWriter<E, T> {

    /**
     * Write the provided entity to target stream.
     *
     * @param entity entity to publish
     * @param target the target stream
     */
    void writeTo(E entity, T target);
}

• EntityDecoder：对响应数据的反序列接口

public interface EntityDecoder<S, T> {

    /**
     * Decode target object from source stream.
     *
     * @param source source stream
     * @return decoded target object
     */
    T decode(S source);
}

• RequestEntityWriter:客户端请求的数据序列化处理接口，继承自EntityWriter;

public interface RequestEntityWriter<E extends Request, T> extends EntityWriter<E, T> {

}

请求实体对象继承自Request

• RequestEntityDecoder:客户端对服务返回的数据反序列化接口，继续自EntityDecoder：

public interface RequestEntityDecoder<S, T extends Request> extends EntityDecoder<S, T> {

}

• ResponseEntityDecoder:服务端对请求客户端的数据反序列化接口，继续自EntityDecoder：

public interface ResponseEntityDecoder<S, T extends Response> extends EntityDecoder<S, T> {

}

响应实体对象继承自Response

• ResponseEntityWriter:服务端对响应数据的序列化接口，继承自EntityWriter。

public interface ResponseEntityWriter<E extends Response, T> extends EntityWriter<E, T> {

}

request包

请求对象接口Request

public interface Request {

    /**
     * Get request type.
     *
     * @return request type
     */
    int getType();

    /**
     * Get request ID.
     *
     * @return unique request ID
     */
    int getId();
}

• 该接口定义两个方法：getType()和getId()；用于获取请求的类型和Id;实际上就是ClusterRequest实体对象的get方法。

ClusterRequest

public class ClusterRequest<T> implements Request {

    private int id;
    private int type;

    private T data;

    public ClusterRequest() {}

    public ClusterRequest(int id, int type, T data) {
        this.id = id;
        this.type = type;
        this.data = data;
    }
    //省略部分代码
}

• ClusterRequest：集群客户端请求对象，对象实体是泛型T表示

FlowRequestData、ParamFlowRequestData
这两个对象表示正常限流请求对象，以及热点参数限流对象；源码如下：

public class FlowRequestData {
    //限流id，规则id
    private long flowId;
    private int count;
    private boolean priority;
}

public class ParamFlowRequestData {

    private long flowId;
    private int count;
    private Collection<Object> params;
}

• 两个主体具体用法后面分析；

response包

响应对象接口Response

public interface Response {
    /**
     * Get response ID.
     *
     * @return response ID
     */
    int getId();

    /**
     * Get response type.
     *
     * @return response type
     */
    int getType();

    /**
     * Get response status.
     *
     * @return response status
     */
    int getStatus();
}

• 和Request接口一样，定义了几个获取Id，type,status的方法。
  ClusterResponse：集群响应对象

public class ClusterResponse<T> implements Response {

    private int id;
    private int type;
    private int status;

    //响应的具体数据
    private T data;

    public ClusterResponse() {}
    
    //省略部分代码
}

FlowTokenResponseData:服务端响应的data

public class FlowTokenResponseData {

    //剩余次数
    private int remainingCount;
    //等待时间
    private int waitInMs;
}

ClusterTransportClient接口

public interface ClusterTransportClient {

    /**
     * Start the client.
     *
     * @throws Exception some error occurred (e.g. initialization failed)
     */
     //启动客户端
    void start() throws Exception;

    /**
     * Stop the client.
     *
     * @throws Exception some error occurred (e.g. shutdown failed)
     */
     //停止客户端
    void stop() throws Exception;

    /**
     * Send request to remote server and get response.
     *
     * @param request Sentinel cluster request
     * @return response from remote server
     * @throws Exception some error occurs
     */
     //发送请求
    ClusterResponse sendRequest(ClusterRequest request) throws Exception;

    /**
     * Check whether the client has been started and ready for sending requests.
     *
     * @return true if the client is ready to send requests, otherwise false
     */
     //判断客户端是否准备好
    boolean isReady();
}

• ClusterTransportClient客户端传送接口，定义4个方法如上；默认实现类NettyTransportClient在client模块中，后面讲解。

ConfigSupplierRegistry配置注册

public final class ConfigSupplierRegistry {

    /**
     * The default namespace supplier provides appName as namespace.
     */
    private static final Supplier<String> DEFAULT_APP_NAME_SUPPLIER = new Supplier<String>() {
        @Override
        public String get() {
            return AppNameUtil.getAppName();
        }
    };
    /**
     * Registered namespace supplier.
     */
    private static Supplier<String> namespaceSupplier = DEFAULT_APP_NAME_SUPPLIER;

    /**
     * Get the registered namespace supplier.
     *
     * @return the registered namespace supplier
     */
    public static Supplier<String> getNamespaceSupplier() {
        return namespaceSupplier;
    }

    public static void setNamespaceSupplier(Supplier<String> namespaceSupplier) {
        AssertUtil.notNull(namespaceSupplier, "namespaceSupplier cannot be null");
        ConfigSupplierRegistry.namespaceSupplier = namespaceSupplier;
        RecordLog.info("[ConfigSupplierRegistry] New namespace supplier provided, current supplied: "
            + namespaceSupplier.get());
    }

    private ConfigSupplierRegistry() {}

• 集群限流服务端作用设置或获取，默认是服务的name

其他

• ClusterConstants：常量定义
• ClusterErrorMessages：器群错误消息定义
• SentinelClusterException：定义了SentinelCluster异常
• RequestType：定义了一个请求类型注解