分布式限流，你想知道的都在這里

作者：儒雅程序員 2019-04-24 08:31:43

網絡

通信技術

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

網站建設哪家好，找創(chuàng)新互聯建站！專注于網頁設計、網站建設、微信開發(fā)、成都微信小程序、集團企業(yè)網站建設等服務項目。為回饋新老客戶創(chuàng)新互聯還提供了霍邱免費建站歡迎大家使用！

前言

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

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

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

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

常見算法

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

• 漏桶算法
• 令牌桶算法

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

如下圖所示：

這種做法簡單粗暴。

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

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

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

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

RateLimiter實現

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

  
 
 
 

   
  
  
  
@Override 
   
  
  
  
public BaseResponse getUserByFeignBatch(@RequestBody UserReqVO userReqVO) { 
   
  
  
  
 //調用遠程服務 
   
  
  
  
 OrderNoReqVO vo = new OrderNoReqVO() ; 
   
  
  
  
 vo.setReqNo(userReqVO.getReqNo()); 
   
  
  
  
 RateLimiter limiter = RateLimiter.create(2.0) ; 
   
  
  
  
 //批量調用 
   
  
  
  
 for (int i = 0 ;i< 10 ; i++){ 
   
  
  
  
 double acquire = limiter.acquire(); 
   
  
  
  
 logger.debug("獲取令牌成功!,消耗=" + acquire); 
   
  
  
  
 BaseResponse 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 ; 
   
  
  
  
} 
  
 
 
 

詳見此。

調用結果如下：

代碼可以看出以每秒向桶中放入兩個令牌，請求一次消耗一個令牌。所以每秒鐘只能發(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 getOrderNo(@RequestBody OrderNoReqVO orderNoReq) { 
   
  
  
  
 BaseResponse 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 getOrderNoLimit(@RequestBody OrderNoReqVO orderNoReq) { 
   
  
  
  
 BaseResponse res = new BaseResponse(); 
   
  
  
  
 // 業(yè)務邏輯 
   
  
  
  
 return res ; 
   
  
  
  
} 
  
 
 
 

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

普通方法也可使用：

  
 
 
 

   
  
  
  
@CommonLimit 
   
  
  
  
public void doSomething(){} 
  
 
 
 

會在調用達到閾值時拋出異常。

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

  
 
 
 

   
  
  
  
@Override 
   
  
  
  
public BaseResponse getUserByFeign(@RequestBody UserReqVO userReq) { 
   
  
  
  
 //調用遠程服務 
   
  
  
  
 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 orderNo = orderServiceClient.getOrderNoCommonLimit(vo); 
   
  
  
  
 logger.info("遠程返回:" + JSON.toJSONString(orderNo)); 
   
  
  
  
 } 
   
  
  
  
} 
  
 
 
 

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

效果如下：

實現原理

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

其中 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 中的調用邏輯:

  
 
 
 

   
  
  
  
--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 
  
 
 
 

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

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

Builder 構建器

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

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

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

  
 
 
 

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

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

第二種方案可以采用 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  
   
  
  
  
 */ 
   
  
  
  
 public static class Builder{ 
   
  
  
  
 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; 
   
  
  
  
 } 
   
  
  
  
 public RedisLimit build(){ 
   
  
  
  
 return new RedisLimit(this) ; 
   
  
  
  
 } 
   
  
  
  
 } 
   
  
  
  
} 
  
 
 
 

這樣客戶端在使用時：

  
 
 
 

   
  
  
  
RedisLimit redisLimit = new RedisLimit.Builder<>(jedisCluster) 
   
  
  
  
 .limit(limit) 
   
  
  
  
 .build(); 
  
 
 
 

更加的簡單直接，并且避免了將創(chuàng)建過程分成了多個子步驟。

這在有多個構造參數，但又不是必選字段時很有作用。

因此順便將分布式鎖的構建器方式也一并更新了：

https://github.com/crossoverJie/distributed-redis-tool#features

API

從上文可以看出，使用過程就是調用 limit 方法。

  
 
 
 

   
  
  
  
//限流 
   
  
  
  
 boolean limit = redisLimit.limit(); 
   
  
  
  
 if (!limit){ 
   
  
  
  
 //具體限流邏輯 
   
  
  
  
 } 
  
 
 
 

為了減少侵入性，也為了簡化客戶端提供了兩種注解方式。

@ControllerLimit

該注解可以作用于 @RequestMapping 修飾的接口中，并會在限流后提供限流響應。

實現如下：

  
 
 
 

   
  
  
  
@Component 
   
  
  
  
public class WebIntercept extends WebMvcConfigurerAdapter { 
   
  
  
  
 private static Logger logger = LoggerFactory.getLogger(WebIntercept.class); 
   
  
  
  
 @Autowired 
   
  
  
  
 private RedisLimit redisLimit; 
   
  
  
  
 @Override 
   
  
  
  
 public void addInterceptors(InterceptorRegistry registry) { 
   
  
  
  
 registry.addInterceptor(new CustomInterceptor()) 
   
  
  
  
 .addPathPatterns("/**"); 
   
  
  
  
 } 
   
  
  
  
 private class CustomInterceptor extends HandlerInterceptorAdapter { 
   
  
  
  
 @Override 
   
  
  
  
 public boolean preHandle(HttpServletRequest request, HttpServletResponse response, 
   
  
  
  
 Object handler) throws Exception { 
   
  
  
  
 if (redisLimit == null) { 
   
  
  
  
 throw new NullPointerException("redisLimit is null"); 
   
  
  
  
 } 
   
  
  
  
 if (handler instanceof HandlerMethod) { 
   
  
  
  
 HandlerMethod method = (HandlerMethod) handler; 
   
  
  
  
 ControllerLimit annotation = method.getMethodAnnotation(ControllerLimit.class); 
   
  
  
  
 if (annotation == null) { 
   
  
  
  
 //skip 
   
  
  
  
 return true; 
   
  
  
  
 } 
   
  
  
  
 boolean limit = redisLimit.limit(); 
   
  
  
  
 if (!limit) { 
   
  
  
  
 logger.warn("request has bean limit"); 
   
  
  
  
 response.sendError(500, "request limit"); 
   
  
  
  
 return false; 
   
  
  
  
 } 
   
  
  
  
 } 
   
  
  
  
 return true; 
   
  
  
  
 } 
   
  
  
  
 } 
   
  
  
  
} 
  
 
 
 

其實就是實現了 SpringMVC 中的攔截器，并在攔截過程中判斷是否有使用注解，從而調用限流邏輯。

前提是應用需要掃描到該類，讓 Spring 進行管理。

  
 
 
 

   
  
  
  
@ComponentScan(value = "com.crossoverjie.distributed.intercept") 
  
 
 
 

@CommonLimit

當然也可以在普通方法中使用。實現原理則是 Spring AOP (SpringMVC 的攔截器本質也是 AOP)。

  
 
 
 

   
  
  
  
@Aspect 
   
  
  
  
@Component 
   
  
  
  
@EnableAspectJAutoProxy(proxyTargetClass = true) 
   
  
  
  
public class CommonAspect { 
   
  
  
  
 private static Logger logger = LoggerFactory.getLogger(CommonAspect.class); 
   
  
  
  
 @Autowired 
   
  
  
  
 private RedisLimit redisLimit ; 
   
  
  
  
 @Pointcut("@annotation(com.crossoverjie.distributed.annotation.CommonLimit)") 
   
  
  
  
 private void check(){} 
   
  
  
  
 @Before("check()") 
   
  
  
  
 public void before(JoinPoint joinPoint) throws Exception { 
   
  
  
  
 if (redisLimit == null) { 
   
  
  
  
 throw new NullPointerException("redisLimit is null"); 
   
  
  
  
 } 
   
  
  
  
 boolean limit = redisLimit.limit(); 
   
  
  
  
 if (!limit) { 
   
  
  
  
 logger.warn("request has bean limit"); 
   
  
  
  
 throw new RuntimeException("request has bean limit") ; 
   
  
  
  
 } 
   
  
  
  
 } 
   
  
  
  
} 
  
 
 
 

很簡單，也是在攔截過程中調用限流。

當然使用時也得掃描到該包:

  
 
 
 

   
  
  
  
@ComponentScan(value = "com.crossoverjie.distributed.intercept") 
  
 
 
 

總結

限流在一個高并發(fā)大流量的系統(tǒng)中是保護應用的一個利器，成熟的方案也很多，希望對剛了解這一塊的朋友提供一些思路。

分享名稱：分布式限流，你想知道的都在這里
本文地址：http://www.dlmjj.cn/article/cogsppo.html