Sentinel 流量控制（flow control）

https://github.com/alibaba/Sentinel/wiki/%E6%B5%81%E9%87%8F%E6%8E%A7%E5%88%B6

流量控制（flow control），其原理是監控應用流量的 QPS 或並發線程數等指標，當達到指定的閾值時對流量進行控制，以避免被瞬時的流量高峰沖垮，從而保障應用的高可用性。

應用場景：

應對洪峰流量：秒殺、大促、下單、訂單回流處理

消息型場景：削峰填谷，冷熱啟動

付費系統：根據使用流量付費

API Gateway：精准控制API流量

任何應用：探測應用中運行的慢程序塊，時行限制

Provider端控制脈沖流量

針對不同調用來源進行流控

Web接口流控

 

配置規則：

梳理核心接口

通過事前壓測評估核心接口的容量

配置QPS閾值

 

限流閾值類型

QPS(Query Per Second)：每秒請求數。就是說服務器在一秒的時間內處理了多少個請求。

 

QPS

進入簇點鏈路選擇具體的訪問的API，然后點擊流控按鈕

 

 

 

 

 

 

 

package com.wsm.order.controller;

import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/order")
public class OrderController {

    @RequestMapping("/add")
    public String add(){
        System.out.println("=========add===");
        return "hello world";
    }

    @RequestMapping("/flow")
    @SentinelResource(value = "flow",blockHandler = "flowBlockHandler")
    public String flow(){
        System.out.println("========flow====");
        return "正常訪問";
    }

    public String flowBlockHandler(BlockException e){
        return "流控了";
    }
}

 

 

 

 

 

 

 

 

 

 

 

並發線程數控制

並發數控制用於保護業務線程池不被慢調用耗盡。例如，當應用所依賴的下游應用由於某種原因導致服務不穩定、響應延遲增加，對於調用者來說，意味着吞吐量下降和更多的線程數占用，極端情況下甚至導致線程池耗盡。為應對太多線程占用的情況，業內有使用隔離的方案，比如通過不同業務邏輯使用不同線程池來隔離業務自身之間的資源爭搶（線程池隔離）。這種隔離方案雖然隔離性比較好，但是代價就是線程數目太多，線程上下文切換的 overhead 比較大，特別是對低延時的調用有比較大的影響。Sentinel 並發控制不負責創建和管理線程池，而是簡單統計當前請求上下文的線程數目（正在執行的調用數目），如果超出閾值，新的請求會被立即拒絕，效果類似於信號量隔離。並發數控制通常在調用端進行配置。

 

package com.wsm.order.controller;

import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/order")
public class OrderController {

    @RequestMapping("/add")
    public String add(){
        System.out.println("=========add===");
        return "hello world";
    }

    @RequestMapping("/flow")
    @SentinelResource(value = "flow",blockHandler = "flowBlockHandler")
    public String flow(){
        System.out.println("========flow====");
        return "正常訪問";
    }

    public String flowBlockHandler(BlockException e){
        return "流控了";
    }

    @RequestMapping("/flowThread")
    @SentinelResource(value = "flowThread",blockHandler = "flowBlockHandler")
    public String flowThread() throws InterruptedException {
        TimeUnit.SECONDS.sleep(5);
        System.out.println("========flowThread====");
        return "正常訪問";
    }
}

 

 

 

 

 

 

 

 

 

 

 

server端統一限流降級返回值

原理

這里給出server端限流原理的源碼查看流程,可以看出spring-cloud-starter-alibaba-sentinel中自動裝配了攔截器來攔截所有http請求，最終的異常處理類是BlockExceptionHandler。

SentinelWebAutoConfiguration -SentinelWebInterceptor-AbstractSentinelInterceptor-BaseWebMvcConfig-BlockExceptionHandler

sentinel給了一個默認實現類，這也就是我們看到的"Blocked by Sentinel (flow limiting)"。

public class DefaultBlockExceptionHandler implements BlockExceptionHandler {
    public DefaultBlockExceptionHandler() {
    }

    public void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception {
        response.setStatus(429);
        StringBuffer url = request.getRequestURL();
        if ("GET".equals(request.getMethod())  StringUtil.isNotBlank(request.getQueryString())) {
            url.append("").append(request.getQueryString());
        }

        PrintWriter out = response.getWriter();
        out.print("Blocked by Sentinel (flow limiting)");
        out.flush();
        out.close();
    }
}

改造

定義自己的異常處理類並加入的spring容器中

 

package com.wsm.order.domain;

public class Result<T> {
    private Integer code;
    private String msg;
    private T data;

    public Result(Integer code, String msg, T data) {
        this.code = code;
        this.msg = msg;
        this.data = data;
    }

    public Result(Integer code, String msg) {
        this.code = code;
        this.msg = msg;
    }

    public Integer getCode() {
        return code;
    }

    public void setCode(Integer code) {
        this.code = code;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }

    public static Result error(Integer code, String msg){
        return new Result(code,msg);
    }
}

 

package com.wsm.order.exception;

import com.alibaba.csp.sentinel.adapter.spring.webmvc.callback.BlockExceptionHandler;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityException;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeException;
import com.alibaba.csp.sentinel.slots.block.flow.FlowException;
import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowException;
import com.alibaba.csp.sentinel.slots.system.SystemBlockException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wsm.order.domain.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

@Component
public class MyBlockExceptionHandler implements BlockExceptionHandler {

    Logger log = LoggerFactory.getLogger(this.getClass());

    @Override
    public void handle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, BlockException e) throws Exception {
//        e.getRule() 資源 規則的詳細信息
        log.info("==BlockExceptionHandler ======"+ e.getRule());

        Result result = null;

        if(e instanceof FlowException){
            result = Result.error(100,"接口被限流了");
        }else if(e instanceof DegradeException){
            result = Result.error(101,"服務降級了");
        }else if(e instanceof ParamFlowException){
            result = Result.error(102,"熱點參數限流了");
        }else if(e instanceof SystemBlockException){
            result = Result.error(103,"觸發系統保護規則了");
        }else if(e instanceof AuthorityException){
            result = Result.error(104,"授權規則不通過");
        }

        //返回JSON數據
        httpServletResponse.setStatus(500);
        httpServletResponse.setCharacterEncoding("utf-8");
        httpServletResponse.setContentType(MediaType.APPLICATION_JSON_VALUE);
        new ObjectMapper().writeValue(httpServletResponse.getWriter(),result);

    }

//        private BlockExceptionUtil blockExceptionUtil;
//
//    public MyBlockExceptionHandler(BlockExceptionUtil blockExceptionUtil) {
//            this.blockExceptionUtil = blockExceptionUtil;
//        }
//    public MyBlockExceptionHandler() {
//        }
//
//        @Override
//        public void handle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, BlockException e) throws Exception {
//            BaseDtoResponse baseDtoResponse = blockExceptionUtil.getResponseDto(e, null);
//            httpServletResponse.setStatus(HttpStatus.OK.value());
//            httpServletResponse.setCharacterEncoding("UTF-8");
//            httpServletResponse.setContentType("application/json;charset=utf-8");
//            httpServletResponse.setHeader("Content-Type","application/json;charset=utf-8");
//            new ObjectMapper().writeValue(httpServletResponse.getWriter(),baseDtoResponse);
//        }
}