分布式接口幂等性、分布式限流（Guava 、nginx和lua限流）

一、接口幂等性

接口幂等性就是使用者對于同一操作發起的一次請求或者多次請求的結果是一緻的，不會因為多次點選而産生了副作用。舉個最簡單的例子，那就是支付，使用者購買商品後支付，支付扣款成功，但是傳回結果的時候網絡異常，此時錢已經扣了，使用者再次點選按鈕，此時會進行第二次扣款，傳回結果成功，使用者查詢餘額返發現多扣錢了，流水記錄也變成了兩條,這就沒有保證接口的幂等性。

幂等性的核心思想：通過唯一的業務單号保障幂等性，非并發的情況下，查詢業務單号有沒有操作過，沒有則執行操作，并發情況下，這個操作過程需要加鎖。

1、Update操作的幂等性

1）根據唯一業務号去更新資料

通過版本号的方式，來控制update的操作的幂等性，使用者查詢出要修改的資料，系統将資料傳回給頁面，将資料版本号放入隐藏域，使用者修改資料，點選送出，将版本号一同送出給背景，背景使用版本号作為更新條件

update set version = version +1 ,xxx=${xxx} where id =xxx and version = ${version};
           

2、使用Token機制，保證update、insert操作的幂等性

1）沒有唯一業務号的update與insert操作

進入到注冊頁時，背景統一生成Token， 傳回前台隐藏域中，使用者在頁面點選送出時，将Token一同傳入背景，使用Token擷取分布式鎖，完成Insert操作，執行成功後，不釋放鎖，等待過期自動釋放。

二、分布式限流

1、分布式限流的幾種次元

時間 限流基于某段時間範圍或者某個時間點，也就是我們常說的“時間視窗”，比如對每分鐘、每秒鐘的時間視窗做限定 資源 基于可用資源的限制，比如設定最大通路次數，或最高可用連接配接數

上面兩個次元結合起來看，限流就是在某個時間視窗對資源通路做限制，比如設定每秒最多100個通路請求。但在真正的場景裡，我們不止設定一種限流規則，而是會設定多個限流規則共同作用，主要的幾種限流規則如下：

1）QPS和連接配接數控制

針對上圖中的連接配接數和QPS(query per second)限流來說，我們可以設定IP次元的限流，也可以設定基于單個伺服器的限流。在真實環境中通常會設定多個次元的限流規則，比如設定同一個IP每秒通路頻率小于10，連接配接數小于5，再設定每台機器QPS最高1000，連接配接數最大保持200。

更進一步，我們可以把某個伺服器組或整個機房的伺服器當做一個整體，設定更high-level的限流規則，這些所有限流規則都會共同作用于流量控制。

2）傳輸速率

對于“傳輸速率”大家都不會陌生，比如資源的下載下傳速度。有的網站在這方面的限流邏輯做的更細緻，比如普通注冊使用者下載下傳速度為100k/s，購買會員後是10M/s，這背後就是基于使用者組或者使用者标簽的限流邏輯。

3）黑白名單

黑白名單是各個大型企業應用裡很常見的限流和放行手段，而且黑白名單往往是動态變化的。舉個例子，如果某個IP在一段時間的通路次數過于頻繁，被系統識别為機器人使用者或流量攻擊，那麼這個IP就會被加入到黑名單，進而限制其對系統資源的通路，這就是我們俗稱的“封IP”。

我們平時見到的爬蟲程式，比如說爬知乎上的美女圖檔，或者爬券商系統的股票分時資訊，這類爬蟲程式都必須實作更換IP的功能，以防被加入黑名單。有時我們還會發現公司的網絡無法通路12306這類大型公共網站，這也是因為某些公司的出網IP是同一個位址，是以在通路量過高的情況下，這個IP位址就被對方系統識别，進而被添加到了黑名單。使用家庭寬帶的同學們應該知道，大部分網絡營運商都會将使用者配置設定到不同出網IP段，或者時不時動态更換使用者的IP位址。

白名單就更好了解了，相當于禦賜金牌在身，可以自由穿梭在各種限流規則裡，暢行無阻。比如某些電商公司會将超大賣家的賬号加入白名單，因為這類賣家往往有自己的一套運維系統，需要對接公司的IT系統做大量的商品釋出、補貨等等操作。

4）分布式環境

所謂的分布式限流，其實道理很簡單，一句話就可以解釋清楚。分布式差別于單機限流的場景，它把整個分布式環境中所有伺服器當做一個整體來考量。比如說針對IP的限流，我們限制了1個IP每秒最多10個通路，不管來自這個IP的請求落在了哪台機器上，隻要是通路了叢集中的服務節點，那麼都會受到限流規則的制約。

從上面的例子不難看出，我們必須将限流資訊儲存在一個“中心化”的元件上，這樣它就可以擷取到叢集中所有機器的通路狀态，目前有兩個比較主流的限流方案：

網關層限流

• 将限流規則應用在所有流量的入口處

中間件限流

• 将限流資訊存儲在分布式環境中某個中間件裡（比如Redis緩存），每個元件都可以從這裡擷取到目前時刻的流量統計，進而決定是拒絕服務還是放行流量

2、限流方案常用算法講解

1）令牌桶算法

Token Bucket令牌桶算法是目前應用最為廣泛的限流算法，顧名思義，它有以下兩個關鍵角色：

• 令牌 擷取到令牌的Request才會被處理，其他Requests要麼排隊要麼被直接丢棄
• 桶 用來裝令牌的地方，所有Request都從這個桶裡面擷取令牌

  

令牌生成

這個流程涉及到令牌生成器和令牌桶，前面我們提到過令牌桶是一個裝令牌的地方，既然是個桶那麼必然有一個容量，也就是說令牌桶所能容納的令牌數量是一個固定的數值。

對于令牌生成器來說，它會根據一個預定的速率向桶中添加令牌，比如我們可以配置讓它以每秒100個請求的速率發放令牌，或者每分鐘50個。注意這裡的發放速度是勻速，也就是說這50個令牌并非是在每個時間視窗剛開始的時候一次性發放，而是會在這個時間視窗内勻速發放。

在令牌發放器就是一個水龍頭，假如在下面接水的桶子滿了，那麼自然這個水（令牌）就流到了外面。在令牌發放過程中也一樣，令牌桶的容量是有限的，如果目前已經放滿了額定容量的令牌，那麼新來的令牌就會被丢棄掉。

令牌擷取

每個通路請求到來後，必須擷取到一個令牌才能執行後面的邏輯。假如令牌的數量少，而通路請求較多的情況下，一部分請求自然無法擷取到令牌，那麼這個時候我們可以設定一個“緩沖隊列”來暫存這些多餘的令牌。

緩沖隊列其實是一個可選的選項，并不是所有應用了令牌桶算法的程式都會實作隊列。當有緩存隊列存在的情況下，那些暫時沒有擷取到令牌的請求将被放到這個隊列中排隊，直到新的令牌産生後，再從隊列頭部拿出一個請求來比對令牌。

當隊列已滿的情況下，這部分通路請求将被丢棄。在實際應用中我們還可以給這個隊列加一系列的特效，比如設定隊列中請求的存活時間，或者将隊列改造為PriorityQueue，根據某種優先級排序，而不是先進先出。算法是死的，人是活的，先進的生産力來自于不斷地創造，在技術領域尤其如此。

2）漏桶算法

Leaky Bucket

漏桶算法的前半段和令牌桶類似，但是操作的對象不同，令牌桶是将令牌放入桶裡，而漏桶是将通路請求的資料包放到桶裡。同樣的是，如果桶滿了，那麼後面新來的資料包将被丢棄。

漏桶算法的後半程是有鮮明特色的，它永遠隻會以一個恒定的速率将資料包從桶内流出。打個比方，如果我設定了漏桶可以存放100個資料包，然後流出速度是1s一個，那麼不管資料包以什麼速率流入桶裡，也不管桶裡有多少資料包，漏桶能保證這些資料包永遠以1s一個的恒定速度被處理。

漏桶 vs 令牌桶的差別

根據它們各自的特點不難看出來，這兩種算法都有一個“恒定”的速率和“不定”的速率。令牌桶是以恒定速率建立令牌，但是通路請求擷取令牌的速率“不定”，反正有多少令牌發多少，令牌沒了就幹等。而漏桶是以“恒定”的速率處理請求，但是這些請求流入桶的速率是“不定”的。

從這兩個特點來說，漏洞的天然特性決定了它不會發生突發流量，就算每秒1000個請求到來，那麼它對背景服務輸出的通路速率永遠恒定。而令牌桶則不同，其特性可以“預存”一定量的令牌，是以在應對突發流量的時候可以在短時間内消耗所有令牌，其突發流量處理效率會比漏桶高，但是導向背景系統的壓力也會相應增多。

3、分布式限流的主流方案

這裡主要講nginx和lua的限流，gateway和hystrix放在後面springcloud中講

1）Guava RateLimiter用戶端限流

1.引入maven

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>18.0</version>
</dependency>
           

2.編寫Controller

@RestController
@Slf4j
public class Controller{
    //每秒鐘可以建立兩個令牌
    RateLimiter limiter = RateLimiter.create(2.0);
    
    //非阻塞限流
    @GetMapping("/tryAcquire")
    public String tryAcquire(Integer count){
        //count 每次消耗的令牌
        if(limiter.tryAcquire(count)){
            log.info("成功，允許通過，速率為{}",limiter.getRate());
            return "success";
        }else{
            log.info("錯誤，不允許通過，速率為{}",limiter.getRate());
            return "fail";
        }
    }
    
    //限定時間的非阻塞限流
    @GetMapping("/tryAcquireWithTimeout")
    public String tryAcquireWithTimeout(Integer count, Integer timeout){
        //count 每次消耗的令牌  timeout 逾時等待的時間
        if(limiter.tryAcquire(count,timeout,TimeUnit.SECONDS)){
            log.info("成功，允許通過，速率為{}",limiter.getRate());
            return "success";
        }else{
            log.info("錯誤，不允許通過，速率為{}",limiter.getRate());
            return "fail";
        }
    }
    
    //同步阻塞限流
    @GetMapping("/acquire")
    public String acquire(Integer count){
        limiter.acquire(count);
        log.info("成功，允許通過，速率為{}",limiter.getRate());
        return "success";
    }
}
           

2）基于Nginx的限流

1.iP限流

1.編寫Controller

@RestController
@Slf4j
public class Controller{
    //nginx測試使用
    @GetMapping("/nginx")
    public String nginx(){
        log.info("Nginx success");
    }
}
           

2.修改host檔案，添加一個網址域名

127.0.0.1   www.test.com
           

3.修改nginx，将步驟2中的域名，添加到路由規則當中

打開nginx的配置檔案

vim /usr/local/nginx/conf/nginx.conf
           

添加一個服務

#根據IP位址限制速度
#1）$binary_remote_addr   binary_目的是縮寫記憶體占用，remote_addr表示通過IP位址來限流
#2）zone=iplimit:20m   iplimit是一塊記憶體區域（記錄通路頻率資訊），20m是指這塊記憶體區域的大小
#3）rate=1r/s  每秒放行1個請求
limit_req_zone $binary_remote_addr zone=iplimit:20m rate=1r/s;

server{
    server_name www.test.com;
    location /access-limit/ {
        proxy_pass http://127.0.0.1:8080/;
        
        #基于ip位址的限制
        #1）zone=iplimit 引用limit_rep_zone中的zone變量
        #2）burst=2  設定一個大小為2的緩沖區域，當大量請求到來，請求數量超過限流頻率時，将其放入緩沖區域
        #3）nodelay   緩沖區滿了以後，直接傳回503異常
        limit_req zone=iplimit burst=2 nodelay;
    }
}
           

4.通路位址，測試是否限流

www.test.com/access-limit/nginx

2.多元度限流

1.修改nginx配置

#根據IP位址限制速度
limit_req_zone $binary_remote_addr zone=iplimit:20m rate=10r/s;
#根據伺服器級别做限流
limit_req_zone $server_name zone=serverlimit:10m rate=1r/s;
#根據ip位址的連結數量做限流
limit_conn_zone $binary_remote_addr zone=perip:20m;
#根據伺服器的連接配接數做限流
limit_conn_zone $server_name zone=perserver:20m;


server{
    server_name www.test.com;
    location /access-limit/ {
        proxy_pass http://127.0.0.1:8080/;
        
        #基于ip位址的限制
        limit_req zone=iplimit burst=2 nodelay;
        #基于伺服器級别做限流
        limit_req zone=serverlimit burst=2 nodelay;
        #基于ip位址的連結數量做限流  最多保持100個連結
        limit_conn zone=perip 100;
        #基于伺服器的連接配接數做限流 最多保持100個連結
        limit_conn zone=perserver 1;
        #配置request的異常傳回504（預設為503）
        limit_req_status 504;
        limit_conn_status 504;
    }
    
     location /download/ {
        #前100m不限制速度
        limit_rate_affer 100m;
        #限制速度為256k
        limit_rate 256k;
     }
}
           

3）基于Redis+Lua的分布式限流

1.Lua腳本

Lua是一個很小巧精緻的語言，它的誕生（1993年）甚至比JDK 1.0還要早。Lua是由标準的C語言編寫的，它的源碼部分不過2萬多行C代碼，甚至一個完整的Lua解釋器也就200k的大小。

Lua往大了說是一個新的程式設計語言，往小了說就是一個腳本語言。對于有程式設計經驗的同學，拿到一個Lua腳本大體上就能把業務邏輯猜的八九不離十了。

Redis内置了Lua解釋器，執行過程保證原子性

2.Lua安裝

安裝Lua：

1.參考http://www.lua.org/ftp/教程，下載下傳5.3.5_1版本，本地安裝

如果你使用的是Mac，那建議用brew工具直接執行brew install lua就可以順利安裝，有關brew工具的安裝可以參考https://brew.sh/網站，使用brew安裝後的目錄在/usr/local/Cellar/lua/5.3.5_1

2.安裝IDEA插件，在IDEA->Preferences面闆，Plugins，裡面Browse repositories，在裡面搜尋lua，然後就選擇同名插件lua。安裝好後重新開機IDEA

3.配置Lua SDK的位置：IDEA->File->Project Structure,選擇添加Lua，路徑指向Lua SDK的bin檔案夾

4.都配置好之後，在項目中右鍵建立Module，左側欄選擇lua，點下一步，選擇lua的sdk，下一步，輸入lua項目名，完成

3.編寫hello lua

print 'Hello Lua'
           

4.編寫模拟限流

-- 模拟限流

-- 用作限流的key
local key = 'my key'

-- 限流的最大門檻值
local limit = 2

-- 目前限流大小
local currentLimit = 2

-- 是否超過限流标準
if currentLimit + 1 > limit then
    print 'reject'
    return false
else
    print 'accept'
    return true
end
           

5.限流元件封裝

1.添加maven

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>18.0</version>
</dependency>
           

2.添加Spring配置

不是重要内容就随便寫點，主要就是把reids配置一下

server.port=8080

spring.redis.database=0
spring.redis.host=localhost
spring.redis.port=6376
           

3.編寫限流腳本

lua腳本放在resource目錄下就可以了

-- 擷取方法簽名特征
local methodKey = KEYS[1]
redis.log(redis.LOG_DEBUG,'key is',methodKey)

-- 調用腳本傳入的限流大小
local limit = tonumber(ARGV[1])

-- 擷取目前流量大小
local count = tonumber(redis.call('get',methodKey) or "0")

--是否超出限流值
if count + 1 >limit then
    -- 拒絕通路
    return false
else
    -- 沒有超過門檻值
    -- 設定目前通路數量+1
    redis.call('INCRBY',methodKey,1)
    -- 設定過期時間
    redis.call('EXPIRE',methodKey,1)
    -- 放行
    return true
end
           

4.使用spring-data-redis元件內建Lua和Redis

建立限流類

@Service
@Slf4j
public class AccessLimiter{
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisScript<Boolean> rateLimitLua;

    public void limitAccess(String key,Integer limit){
        boolean acquired = stringRedisTemplate.execute(
            rateLimitLua,//lua腳本的真身
            Lists.newArrayList(key),//lua腳本中的key清單
            limit.toString()//lua腳本的value清單
        );

        if(!acquired){
            log.error("Your access is blocked,key={}",key);
            throw new RuntimeException("Your access is blocked");
        }
    }
}
           

建立配置類

@Configuration
public class RedisConfiguration{
    public RedisTemplate<String,String> redisTemplate(RedisConnectionFactory factory){
        return new StringRedisTemplate(factory);
    }
    
    public DefaultRedisScript loadRedisScript(){
        DefaultRedisScript redisScript = new DefaultRedisScript();
        redisScript.setLocation(new ClassPathResource("rateLimiter.lua"));
        redisScript.setResultType(java.lang.Boolean.class);
        return redisScript;
    }
}
           

5.在Controller中添加測試方法驗證限流效果

@RestController
@Slf4j
public class Controller{
    @Autowired
    private AccessLimiter accessLimiter;
    
    @GetMapping("test")
    public String test(){
        accessLimiter.limitAccess("ratelimiter-test",1);
        return "success";
    }
} 
           

6.編寫限流注解

1.新增注解

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface AccessLimiterAop{
    int limit();
    
    String methodKey() default "";
}
           

2.新增切面

@Slf4j
@Aspect
@Component
public class AccessLimiterAspect{
    @Autowired
    private AccessLimiter  accessLimiter;

    //根據注解的位置，自己修改
    @Pointcut("@annotation(com.gyx.demo.annotation.AccessLimiter)")
    public void cut(){
        log.info("cut");
    }
    
    @Before("cut()")
    public void before(JoinPoint joinPoint){
        //擷取方法簽名，作為methodkey
        MethodSignature signature =(MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        AccessLimiterAop annotation = method.getAnnotation(AccessLimiterAop.class);
        
        if(annotation == null){
            return;
        }
        String key = annotation.methodKey();
        Integer limit = annotation.limit();
        //如果沒有設定methodKey，就自動添加一個
        if(StringUtils.isEmpty(key)){
            Class[] type = method.getParameterType();
            key = method.getName();
            if (type != null){
                String paramTypes=Arrays.stream(type)
                    .map(Class::getName)
                    .collect(Collectors.joining(","));
                    key += "#"+paramTypes;
            }
        }
        
        //調用redis
        return accessLimiter.limitAccess(key,limit);
    }
}
           

3.在Controller中添加測試方法驗證限流效果

@RestController
@Slf4j
public class Controller{
    @Autowired
    private AccessLimiter accessLimiter;
    
    @GetMapping("test")
    @AccessLImiterAop(limit =1)
    public String test(){
        return "success";
    }
} 
           

 感謝閱讀，希望對你有所幫助 :)