前言

在一個高并發(fā)系統(tǒng)中對流量的把控是非常重要的，當巨大的流量直接請求到我們的服務器上沒多久就可能造成接口不可用，不處理的話甚至會造成整個應用不可用。

比如最近就有個這樣的需求，我作為客戶端要向kafka生產(chǎn)數(shù)據(jù)，而kafka的消費者則再源源不斷的消費數(shù)據(jù)，并將消費的數(shù)據(jù)全部請求到web服務器，雖說做了負載(有4臺web服務器)但業(yè)務數(shù)據(jù)的量也是巨大的，每秒鐘可能有上萬條數(shù)據(jù)產(chǎn)生。如果生產(chǎn)者直接生產(chǎn)數(shù)據(jù)的話極有可能把web服務器拖垮。

對此就必須要做限流處理，每秒鐘生產(chǎn)一定限額的數(shù)據(jù)到kafka，這樣就能極大程度的保證web的正常運轉。

其實不管處理何種場景，本質都是降低流量保證應用的高可用。

常見算法

對于限流常見有兩種算法:

• 漏桶算法
• 令牌桶算法

漏桶算法比較簡單，就是將流量放入桶中，漏桶同時也按照一定的速率流出，如果流量過快的話就會溢出(漏桶并不會提高流出速率)。溢出的流量則直接丟棄。

如下圖所示：

這種做法簡單粗暴。

漏桶算法雖說簡單，但卻不能應對實際場景，比如突然暴增的流量。

這時就需要用到令牌桶算法：

令牌桶會以一個恒定的速率向固定容量大小桶中放入令牌，當有流量來時則取走一個或多個令牌。當桶中沒有令牌則將當前請求丟棄或阻塞。

相比之下令牌桶可以應對一定的突發(fā)流量。

RateLimiter實現(xiàn)

對于令牌桶的代碼實現(xiàn)，可以直接使用Guava包中的RateLimiter。

@Override 
public BaseResponse<UserResVO> getUserByFeignBatch(@RequestBody UserReqVO userReqVO) { 
 //調(diào)用遠程服務 
 OrderNoReqVO vo = new OrderNoReqVO() ; 
 vo.setReqNo(userReqVO.getReqNo()); 
 RateLimiter limiter = RateLimiter.create(2.0) ; 
 //批量調(diào)用 
 for (int i = 0 ;i< 10 ; i++){ 
 double acquire = limiter.acquire(); 
 logger.debug("獲取令牌成功!,消耗=" + acquire); 
 BaseResponse<OrderNoResVO> orderNo = orderServiceClient.getOrderNo(vo); 
 logger.debug("遠程返回:"+JSON.toJSONString(orderNo)); 
 } 
 UserRes userRes = new UserRes() ; 
 userRes.setUserId(123); 
 userRes.setUserName("張三"); 
 userRes.setReqNo(userReqVO.getReqNo()); 
 userRes.setCode(StatusEnum.SUCCESS.getCode()); 
 userRes.setMessage("成功"); 
 return userRes ; 
} 

詳見此。

調(diào)用結果如下：

代碼可以看出以每秒向桶中放入兩個令牌，請求一次消耗一個令牌。所以每秒鐘只能發(fā)送兩個請求。按照圖中的時間來看也確實如此(返回值是獲取此令牌所消耗的時間，差不多也是每500ms一個)。

使用RateLimiter有幾個值得注意的地方:

允許先消費，后付款，意思就是它可以來一個請求的時候一次性取走幾個或者是剩下所有的令牌甚至多取，但是后面的請求就得為上一次請求買單，它需要等待桶中的令牌補齊之后才能繼續(xù)獲取令牌。

總結

針對于單個應用的限流 RateLimiter 夠用了，如果是分布式環(huán)境可以借助 Redis 來完成。

來做演示。

在 Order 應用提供的接口中采取了限流。首先是配置了限流工具的 Bean：

@Configuration 
public class RedisLimitConfig { 
 @Value("${redis.limit}") 
 private int limit; 
 @Autowired 
 private JedisConnectionFactory jedisConnectionFactory; 
 @Bean 
 public RedisLimit build() { 
 RedisClusterConnection clusterConnection = jedisConnectionFactory.getClusterConnection(); 
 JedisCluster jedisCluster = (JedisCluster) clusterConnection.getNativeConnection(); 
 RedisLimit redisLimit = new RedisLimit.Builder<>(jedisCluster) 
 .limit(limit) 
 .build(); 
 return redisLimit; 
 } 
} 

接著在 Controller 使用組件：

@Autowired 
private RedisLimit redisLimit ; 
@Override 
@CheckReqNo 
public BaseResponse<OrderNoResVO> getOrderNo(@RequestBody OrderNoReqVO orderNoReq) { 
 BaseResponse<OrderNoResVO> res = new BaseResponse(); 
 //限流 
 boolean limit = redisLimit.limit(); 
 if (!limit){ 
 res.setCode(StatusEnum.REQUEST_LIMIT.getCode()); 
 res.setMessage(StatusEnum.REQUEST_LIMIT.getMessage()); 
 return res ; 
 } 
 res.setReqNo(orderNoReq.getReqNo()); 
 if (null == orderNoReq.getAppId()){ 
 throw new SBCException(StatusEnum.FAIL); 
 } 
 OrderNoResVO orderNoRes = new OrderNoResVO() ; 
 orderNoRes.setOrderId(DateUtil.getLongTime()); 
 res.setCode(StatusEnum.SUCCESS.getCode()); 
 res.setMessage(StatusEnum.SUCCESS.getMessage()); 
 res.setDataBody(orderNoRes); 
 return res ; 
} 

為了方便使用，也提供了注解：

@Override 
@ControllerLimit 
public BaseResponse<OrderNoResVO> getOrderNoLimit(@RequestBody OrderNoReqVO orderNoReq) { 
 BaseResponse<OrderNoResVO> res = new BaseResponse(); 
 // 業(yè)務邏輯 
 return res ; 
} 

該注解攔截了 http 請求，會再請求達到閾值時直接返回。

普通方法也可使用：

@CommonLimit 
public void doSomething(){} 

會在調(diào)用達到閾值時拋出異常。

為了模擬并發(fā)，在 User 應用中開啟了 10 個線程調(diào)用 Order(限流次數(shù)為5) 接口(也可使用專業(yè)的并發(fā)測試工具 JMeter 等)。

@Override 
public BaseResponse<UserResVO> getUserByFeign(@RequestBody UserReqVO userReq) { 
 //調(diào)用遠程服務 
 OrderNoReqVO vo = new OrderNoReqVO(); 
 vo.setAppId(1L); 
 vo.setReqNo(userReq.getReqNo()); 
 for (int i = 0; i < 10; i++) { 
 executorService.execute(new Worker(vo, orderServiceClient)); 
 } 
 UserRes userRes = new UserRes(); 
 userRes.setUserId(123); 
 userRes.setUserName("張三"); 
 userRes.setReqNo(userReq.getReqNo()); 
 userRes.setCode(StatusEnum.SUCCESS.getCode()); 
 userRes.setMessage("成功"); 
 return userRes; 
} 
private static class Worker implements Runnable { 
 private OrderNoReqVO vo; 
 private OrderServiceClient orderServiceClient; 
 public Worker(OrderNoReqVO vo, OrderServiceClient orderServiceClient) { 
 this.vo = vo; 
 this.orderServiceClient = orderServiceClient; 
 } 
 @Override 
 public void run() { 
 BaseResponse<OrderNoResVO> orderNo = orderServiceClient.getOrderNoCommonLimit(vo); 
 logger.info("遠程返回:" + JSON.toJSONString(orderNo)); 
 } 
} 

為了驗證分布式效果啟動了兩個 Order 應用。

效果如下：

實現(xiàn)原理

實現(xiàn)原理其實很簡單。既然要達到分布式全局限流的效果，那自然需要一個第三方組件來記錄請求的次數(shù)。

其中 Redis 就非常適合這樣的場景。

• 每次請求時將當前時間(精確到秒)作為 Key 寫入到 Redis 中，超時時間設置為 2 秒，Redis 將該 Key 的值進行自增。
• 當達到閾值時返回錯誤。
• 寫入 Redis 的操作用 Lua 腳本來完成，利用 Redis 的單線程機制可以保證每個 Redis 請求的原子性。

Lua 腳本如下:

--lua 下標從 1 開始-- 限流 keylocal key = KEYS[1]-- 限流大小local limit = tonumber(ARGV[1])-- 獲取當前流量大小local curentLimit = tonumber(redis.call('get', key) or "0")if curentLimit + 1 > limit then -- 達到限流大小 返回 return 0;else -- 沒有達到閾值 value + 1 redis.call("INCRBY", key, 1) redis.call("EXPIRE", key, 2) return curentLimit + 1end

Java 中的調(diào)用邏輯:

--lua 下標從 1 開始 
-- 限流 key 
local key = KEYS[1] 
-- 限流大小 
local limit = tonumber(ARGV[1]) 
-- 獲取當前流量大小 
local curentLimit = tonumber(redis.call('get', key) or "0") 
if curentLimit + 1 > limit then 
 -- 達到限流大小 返回 
 return 0; 
else 
 -- 沒有達到閾值 value + 1 
 redis.call("INCRBY", key, 1) 
 redis.call("EXPIRE", key, 2) 
 return curentLimit + 1 
end 

所以只需要在需要限流的地方調(diào)用該方法對返回值進行判斷即可達到限流的目的。

當然這只是利用 Redis 做了一個粗暴的計數(shù)器，如果想實現(xiàn)類似于上文中的令牌桶算法可以基于 Lua 自行實現(xiàn)。

Builder 構建器

在設計這個組件時想盡量的提供給使用者清晰、可讀性、不易出錯的 API。

比如第一步，如何構建一個限流對象。

最常用的方式自然就是構造函數(shù)，如果有多個域則可以采用重疊構造器的方式:

public A(){} 
public A(int a){} 
public A(int a,int b){} 

缺點也是顯而易見的：如果參數(shù)過多會導致難以閱讀，甚至如果參數(shù)類型一致的情況下客戶端顛倒了順序，但不會引起警告從而出現(xiàn)難以預測的結果。

第二種方案可以采用 JavaBean 模式，利用 setter 方法進行構建:

A a = new A(); 
a.setA(a); 
a.setB(b); 

這種方式清晰易讀，但卻容易讓對象處于不一致的狀態(tài)，使對象處于線程不安全的狀態(tài)。

所以這里采用了第三種創(chuàng)建對象的方式，構建器：

public class RedisLimit { 

 private JedisCommands jedis; 

 private int limit = 200; 

 private static final int FAIL_CODE = 0; 

 /** 

 * lua script 

 */ 

 private String script; 

 private RedisLimit(Builder builder) { 

 this.limit = builder.limit ; 

 this.jedis = builder.jedis ; 

 buildScript(); 

 } 

 /** 

 * limit traffic 

 * @return if true 

 */ 

 public boolean limit() { 

 String key = String.valueOf(System.currentTimeMillis() / 1000); 

 Object result = null; 

 if (jedis instanceof Jedis) { 

 result = ((Jedis) this.jedis).eval(script, Collections.singletonList(key), Collections.singletonList(String.valueOf(limit))); 

 } else if (jedis instanceof JedisCluster) { 

 result = ((JedisCluster) this.jedis).eval(script, Collections.singletonList(key), Collections.singletonList(String.valueOf(limit))); 

 } else { 

 //throw new RuntimeException("instance is error") ; 

 return false; 

 } 

 if (FAIL_CODE != (Long) result) { 

 return true; 

 } else { 

 return false; 

 } 

 } 

 /** 

 * read lua script 

 */ 

 private void buildScript() { 

 script = ScriptUtil.getScript("limit.lua"); 

 } 

 /** 

 * the builder 

 * @param <T> 

 */ 

 public static class Builder<T extends JedisCommands>{ 

 private T jedis = null ; 

 private int limit = 200; 

 public Builder(T jedis){ 

 this.jedis = jedis ; 

 } 

 public Builder limit(int limit){ 

 this.limit = limit ; 

 return this; 

 } 

標題名稱：分布式限流，你想知道的都在這里
URL網(wǎng)址：http://redsoil1982.com.cn/news15/101365.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站導航、網(wǎng)頁設計公司、定制網(wǎng)站、App設計、建站公司、關鍵詞優(yōu)化

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)