快捷導航

基于Redis實現(xiàn)分布式應(yīng)用限流的方法

更新時間：2017年12月15日 10:30:49 作者：冷冷gg

本篇文章主要介紹了基于 Redis 實現(xiàn)分布式應(yīng)用限流的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

限流的目的是通過對并發(fā)訪問/請求進行限速或者一個時間窗口內(nèi)的的請求進行限速來保護系統(tǒng)，一旦達到限制速率則可以拒絕服務(wù)。

前幾天在DD的公眾號，看了一篇關(guān)于使用瓜娃實現(xiàn)單應(yīng)用限流的方案 --》原文，參考《redis in action》實現(xiàn)了一個jedis版本的，都屬于業(yè)務(wù)層次限制。實際場景中常用的限流策略：

Nginx接入層限流

按照一定的規(guī)則如帳號、IP、系統(tǒng)調(diào)用邏輯等在Nginx層面做限流

業(yè)務(wù)應(yīng)用系統(tǒng)限流

通過業(yè)務(wù)代碼控制流量這個流量可以被稱為信號量，可以理解成是一種鎖，它可以限制一項資源最多能同時被多少進程訪問。

代碼實現(xiàn)

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.ZParams;
import java.util.List;
import java.util.UUID;

/**
 * @email wangiegie@gmail.com
 * @data 2017-08
 */
public class RedisRateLimiter {
  private static final String BUCKET = "BUCKET";
  private static final String BUCKET_COUNT = "BUCKET_COUNT";
  private static final String BUCKET_MONITOR = "BUCKET_MONITOR";

  static String acquireTokenFromBucket(
      Jedis jedis, int limit, long timeout) {
    String identifier = UUID.randomUUID().toString();
    long now = System.currentTimeMillis();
    Transaction transaction = jedis.multi();

    //刪除信號量
    transaction.zremrangeByScore(BUCKET_MONITOR.getBytes(), "-inf".getBytes(), String.valueOf(now - timeout).getBytes());
    ZParams params = new ZParams();
    params.weightsByDouble(1.0,0.0);
    transaction.zinterstore(BUCKET, params, BUCKET, BUCKET_MONITOR);

    //計數(shù)器自增
    transaction.incr(BUCKET_COUNT);
    List<Object> results = transaction.exec();
    long counter = (Long) results.get(results.size() - 1);

    transaction = jedis.multi();
    transaction.zadd(BUCKET_MONITOR, now, identifier);
    transaction.zadd(BUCKET, counter, identifier);
    transaction.zrank(BUCKET, identifier);
    results = transaction.exec();
    //獲取排名，判斷請求是否取得了信號量
    long rank = (Long) results.get(results.size() - 1);
    if (rank < limit) {
      return identifier;
    } else {//沒有獲取到信號量，清理之前放入redis 中垃圾數(shù)據(jù)
      transaction = jedis.multi();
      transaction.zrem(BUCKET_MONITOR, identifier);
      transaction.zrem(BUCKET, identifier);
      transaction.exec();
    }
    return null;
  }
}

調(diào)用

測試接口調(diào)用

@GetMapping("/")
public void index(HttpServletResponse response) throws IOException {
  Jedis jedis = jedisPool.getResource();
  String token = RedisRateLimiter.acquireTokenFromBucket(jedis, LIMIT, TIMEOUT);
  if (token == null) {
    response.sendError(500);
  }else{
    //TODO 你的業(yè)務(wù)邏輯
  }
  jedisPool.returnResource(jedis);
}

優(yōu)化

使用攔截器 + 注解優(yōu)化代碼

攔截器

@Configuration
static class WebMvcConfigurer extends WebMvcConfigurerAdapter {
  private Logger logger = LoggerFactory.getLogger(WebMvcConfigurer.class);
  @Autowired
  private JedisPool jedisPool;

  public void addInterceptors(InterceptorRegistry registry) {
    registry.addInterceptor(new HandlerInterceptorAdapter() {
      public boolean preHandle(HttpServletRequest request, HttpServletResponse response,
                   Object handler) throws Exception {
        HandlerMethod handlerMethod = (HandlerMethod) handler;
        Method method = handlerMethod.getMethod();
        RateLimiter rateLimiter = method.getAnnotation(RateLimiter.class);
        if (rateLimiter != null){
          int limit = rateLimiter.limit();
          int timeout = rateLimiter.timeout();
          Jedis jedis = jedisPool.getResource();
          String token = RedisRateLimiter.acquireTokenFromBucket(jedis, limit, timeout);
          if (token == null) {
            response.sendError(500);
            return false;
          }
          logger.debug("token -> {}",token);
          jedis.close();
        }
        return true;
      }
    }).addPathPatterns("/*");
  }
}

定義注解

/**
 * @email wangiegie@gmail.com
 * @data 2017-08
 * 限流注解
 */

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
  int limit() default 5;
  int timeout() default 1000;
}

使用

@RateLimiter(limit = 2, timeout = 5000)
@GetMapping("/test")
public void test() {
}

并發(fā)測試

工具：apache-jmeter-3.2

說明：沒有獲取到信號量的接口返回500，status是紅色，獲取到信號量的接口返回200，status是綠色。

當限制請求信號量為2，并發(fā)5個線程：

當限制請求信號量為5，并發(fā)10個線程：

資料

基于reids + lua的實現(xiàn)

總結(jié)

對于信號量的操作，使用事務(wù)操作。
不要使用時間戳作為信號量的排序分數(shù)，因為在分布式環(huán)境中，各個節(jié)點的時間差的原因，會出現(xiàn)不公平信號量的現(xiàn)象。
可以使用把這塊代碼抽成@rateLimiter注解，然后再方法上使用就會很方便啦
不同接口的流控，可以參考源碼的里面RedisRateLimiterPlus,無非是每個接口生成一個監(jiān)控參數(shù)
源碼：boding1-pig-cloud-jb51.rar

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: