2. 程式人生 > 其它 >redis實現分散式介面限流

redis實現分散式介面限流

阿新 • • 發佈：2021-01-07

技術標籤：日常開發遇到的問題

限流的目的是通過對併發訪問/請求進行限速或者一個時間視窗內的的請求進行限速來保護系統，一旦達到限制速率則可以拒絕服務。

限流包括兩種：
1.Nginx接入層限流
按照一定的規則如帳號、IP、系統呼叫邏輯等在Nginx層面做限流
2.業務應用系統限流
通過業務程式碼控制流量這個流量可以被稱為訊號量，可以理解成是一種鎖，它可以限制一項資源最多能同時被多少程序訪問。

這裡主要說說第二種使用redis在業務層進行限流。

公司做電商業務的，經常會發放優惠券，在某個點優惠券領取開放後，會有大量的併發流量進入，導致系統壓力過大，為了限制併發流量，這裡使用redis在業務層做了流量控制。

  /**
         * 如果使用者在限流之外，則提示使用者稍後再試
         */
        int limitNum = 100;
        long limitSecond = 1000;
        if (redisUtil.acquireTokenFromBucket(limitNum, limitSecond) == null) {
            logger.error("使用者在限流之外，稍後再試");
            return CommonResult.failed(-3,"使用者在限流之外，稍後再試"
 
);
        }

下面主要介紹一下redisUtil類中的這個方法（這個方法參考了別人的，原文請點選）


    private static final String BUCKET = "BUCKET";
    private static final String BUCKET_COUNT = "BUCKET_COUNT";
    private static final String BUCKET_MONITOR = "BUCKET_MONITOR";
    
 /**
     * Redis實現分散式應用限流的方法
     * @param limit     請求訊號量
     * @param timeout   請求時間
     * @return
     */
 

    public String acquireTokenFromBucket(int limit, long timeout) {
        //這裡獲取jeids連線池，具體方法，可以自己實現
        Jedis jedis = getJedis();
        if(jedis == null){
            return null;
        }
        //這裡開始，是限流的方法
        try{
            String identifier = UUIDGenerator.generate32UUID();
            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);

            //計數器自增
            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 中垃圾資料
                transaction = jedis.multi();
                transaction.zrem(BUCKET_MONITOR, identifier);
                transaction.zrem(BUCKET, identifier);
                transaction.exec();
            }
        } catch (Exception e) {
            log.error("在列表key的尾部插入元素失敗：" + e.getMessage(), e);
            returnBrokenResource(jedis);
        } finally {
            returnResource(jedis);
        }
        return null;
    }

    /**
     * 釋放jedis資源
     *
     * @param jedis
     */
    @SuppressWarnings("deprecation")
    public  void returnResource(final Jedis jedis) {
        if (jedis != null && jedisPool != null) {
            jedisPool.returnResource(jedis);
        }
    }

    @SuppressWarnings("deprecation")
    public  void returnBrokenResource(final Jedis jedis) {
        if (jedis != null && jedisPool != null) {
            jedisPool.returnBrokenResource(jedis);
        }
    }

業務層用法，如最開始的程式碼，通過返回判斷是否為空，判斷當前流量是否限流

以下是從網上找到的詳細方法和測試

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.ZParams;
import java.util.List;
import java.util.UUID;
 
/**
 * 實現類
 */
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);
 
    //計數器自增
    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 中垃圾資料
      transaction = jedis.multi();
      transaction.zrem(BUCKET_MONITOR, identifier);
      transaction.zrem(BUCKET, identifier);
      transaction.exec();
    }
    return null;
  }
}

測試呼叫類

@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 你的業務邏輯
  }
  jedisPool.returnResource(jedis);
}

此外還可以使用註解類+切面攔截來處理

@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("/*");
  }
}

定義註解類

/**
 * 限流注解
 */
 
@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() {
}

測試結果
併發測試

工具：apache-jmeter-3.2
說明： 沒有獲取到訊號量的介面返回500，status是紅色，獲取到訊號量的介面返回200，status是綠色。
當限制請求訊號量為2，併發5個執行緒：
在這裡插入圖片描述

當限制請求訊號量為5，併發10個執行緒：
在這裡插入圖片描述

總結：
1.對於訊號量的操作，使用事務操作。
2.不要使用時間戳作為訊號量的排序分數，因為在分散式環境中，各個節點的時間差的原因，會出現不公平訊號量的現象。
3.可以使用把這塊程式碼抽成@rateLimiter註解，然後再方法上使用就會很方便
4.不同介面的流控，可以參考原始碼的裡面RedisRateLimiterPlus,無非是每個介面生成一個監控引數

相關推薦

redis實現分散式介面限流

技術標籤：日常開發遇到的問題 限流的目的是通過對併發訪問/請求進行限速或者一個時間視窗內的的請求進行限速來保護系統，一旦達到限制速率則可以拒絕服務。

go redis實現滑動視窗限流的方式(redis版)

之前給大家介紹過單機當前程序的滑動視窗限流,這一個是使用go redis list結構實現的滑動視窗限流,原理都一樣,但是支援分散式

Redis實現訪問次數限流，這有難點嗎？

大家好，我是【架構擺渡人】，一隻十年的程式猿，這是流量治理系列的第11篇原創文章，如果有收穫，還請分享給更多的朋友。

基於Redis實現分散式限流

目錄前言模板方法程式碼方法一 固定時間段限流優點缺點核心邏輯tryAcquire測試用例tryAcquire測試用例結果方法二 時間段不固定限流Redis官網虛擬碼優點缺點核心邏輯tryAcquire測試用例tryAcquire測試用例結果方法四

redis實現介面限流

慣例，先看下我們的實戰目錄結構： 首先是pom.xml 核心依賴： <!--用於redis資料庫連線-->

SpringBoot+Redis實現介面限流

原文連結：https://mp.weixin.qq.com/s/xa7oLSiSnYlVPEyMReCCaQ 1.使用maven新增依賴庫，本專案中使用的是：

也來說說redis+lua實現高併發限流

我們的靈活用工系統呼叫優付渠道介面做使用者簽約或資金下發時，優付系統增加了API介面請求的限流策略。

SpringBoot服務上實現介面限流的方法

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

SpringBoot基於Sentinel在服務上實現介面限流

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

詳解Springboot整合sentinel實現介面限流入門

Sentinel是阿里巴巴開源的限流器熔斷器，並且帶有視覺化操作介面。 在日常開發中，限流功能時常被使用，用於對某些介面進行限流熔斷，譬如限制單位時間內介面訪問次數；或者按照某種規則進行限流，如限制ip的單位時間

redis做介面限流

技術標籤：springbootredis redis做介面限流（eladmin） LimitAspect package me.zhengjie.aspect;

自定義註解實現介面限流

自定義註解實現介面限流 場景：限制驗證碼在單位時間內的訪問次數 實現流程：自定義一個註解，註解內包含訪問的次數與單位時間。通過AOP進行切面攔截，獲取註解內的次數和時間，獲取請求的uri與訪問者ip。組成redis的

Redis 實現分散式鎖

在單節點情況下，實現執行緒安全需要靠同步狀態來控制。而在分散式應用中，使程式正確執行不被併發問題影響，就需要分散式鎖來控制。

Redis實現分散式資料夾鎖

緣起 最近做一個專案，類似某度雲盤，另外附加定製功能，本人負責雲盤相關功能實現，這個專案跟雲盤不同的是，以專案為分配許可權的單位，同一個專案及子目錄所有有許可權的使用者可以同時操作所有檔案，這樣就很容

redis實現分散式的方法總結

一 為什麼使用 Redis 在專案中使用 Redis，主要考慮兩個角度：效能和併發。如果只是為了分散式鎖這些其他功能，還有其他中介軟體 Zookpeer 等代替，並非一定要使用 Redis。

Redis實現分散式鎖和等待序列的方法示例

在叢集下，經常會因為同時處理髮生資源爭搶和併發問題，但是我們都知道同步鎖synchronized 、cas 、ReentrankLock 這些鎖的作用範圍都是JVM ，說白了在叢集下沒啥用。這時我們就需要能在多臺JVM 之間決定執行順序的

Redis實現分散式鎖的方法示例

之前我們使用的定時任務都是隻部署在了單臺機器上，為了解決單點的問題，為了保證一個任務，只被一臺機器執行，就需要考慮鎖的問題，於是就花時間研究了這個問題。到底怎樣實現一個分散式鎖呢？

Java RPC框架如何實現客戶端限流配置

*:first-child { margin-top: 0 !important;}body>*:last-child { margin-bottom: 0 !important;}/* BLOCKS=============================================================================*/p,pre { margin: 15px

詳解Java分散式IP限流和防止惡意IP攻擊方案

前言 限流是分散式系統設計中經常提到的概念，在某些要求不嚴格的場景下，使用Guava RateLimiter就可以滿足。但是Guava RateLimiter只能應用於單程序，多程序間協同控制便無能為力。本文介紹一種簡單的處理方式，用於

Java基於redis實現分散式鎖程式碼例項

為什麼會有這個需求： 例如一個簡單使用者的操作，一個執行緒去修改使用者狀態，首先在在記憶體中讀出使用者的狀態，然後在記憶體中進行修改，然後在存到資料庫中。在單執行緒中，這是沒有問題的。但是在多執行緒中由