Redisson实现分布式限流

首页发现

博客园 - GL_BKY

AI笔记 hippo4j线程池监控落地 Archery SQL 审核查询平台 mybatis-typeHandler解决表字段中json映射对象问题 ConcurrentSkipListMap实现随机数命中权重占比 WebLogAspect 面试常问题自动代码质量分析示例搭建 Springboot监听多个端口 Git使用规范一些问题 netty MQ Spring dubbo+zookeeper JVM 数据结构 redis JS前端加密JAVA后端解密详解

Redisson实现分布式限流

GL_BKY · 2024-06-25 · via 博客园 - GL_BKY

前提：

最近公司在做有需求在做分布式限流，调研的限流框架大概有

1、spring cloud gateway集成redis限流,但属于网关层限流
2、阿里Sentinel,功能强大、带监控平台
3、srping cloud hystrix，属于接口层限流，提供线程池与信号量两种方式
4、其他：redission、手撸代码

实际需求情况属于业务端限流，redission更加方便，使用更加灵活，下面介绍下redission分布式限流如何使用及原理：

一、使用

使用很简单、如下

// 1、 声明一个限流器
RRateLimiter rateLimiter = redissonClient.getRateLimiter(key);
// 2、 设置速率，5秒中产生3个令牌
rateLimiter.trySetRate(RateType.OVERALL, 3, 5, RateIntervalUnit.SECONDS);
// 3、试图获取一个令牌，获取到返回true
rateLimiter.tryAcquire(1)

二、原理

1、getRateLimiter

// 声明一个限流器 名称 叫key
redissonClient.getRateLimiter(key)

2、trySetRate

trySetRate方法跟进去底层实现如下：

@Override
public RFuture<Boolean> trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "redis.call('hsetnx', KEYS[1], 'rate', ARGV[1]);"
              + "redis.call('hsetnx', KEYS[1], 'interval', ARGV[2]);"
              + "return redis.call('hsetnx', KEYS[1], 'type', ARGV[3]);",
                Collections.<Object>singletonList(getName()), rate, unit.toMillis(rateInterval), type.ordinal());
    }

举个例子，更容易理解：

比如下面这段代码，5秒中产生3个令牌，并且所有实例共享（RateType.OVERALL所有实例共享、RateType.CLIENT单实例端共享）

trySetRate(RateType.OVERALL, 3, 5, RateIntervalUnit.SECONDS);

那么redis中就会设置3个参数：

　　hsetnx,key,rate,3

　　hsetnx,key,interval,5

　　hsetnx,key,type,0

接着看tryAcquire(1)方法：底层源码如下

private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                "local rate = redis.call('hget', KEYS[1], 'rate');"  //1
              + "local interval = redis.call('hget', KEYS[1], 'interval');"  //2
              + "local type = redis.call('hget', KEYS[1], 'type');" //3
              + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')" //4
              
              + "local valueName = KEYS[2];" //5
              + "if type == 1 then "
                  + "valueName = KEYS[3];" //6
              + "end;"
              
              + "local currentValue = redis.call('get', valueName); " //7
              + "if currentValue ~= false then " 
                     + "if tonumber(currentValue) < tonumber(ARGV[1]) then " //8
                         + "return redis.call('pttl', valueName); "
                     + "else "
                         + "redis.call('decrby', valueName, ARGV[1]); " //9
                         + "return nil; "
                     + "end; "
              + "else " //10
                     + "redis.call('set', valueName, rate, 'px', interval); " 
                     + "redis.call('decrby', valueName, ARGV[1]); "
                     + "return nil; "
              + "end;",
                Arrays.<Object>asList(getName(), getValueName(), getClientValueName()), 
                value, commandExecutor.getConnectionManager().getId().toString());
    }

第1、2、3备注行是获取上一步set的3个值：rate、interval、type，如果这3个值没有设置，直接返回rateLimiter没有被初始化。
第5备注行声明一个变量叫valueName 值为KEYS[2],KEYS[2]对应的值是getValueName()方法，getValueName()返回的就是上面第一步getRateLimiter我们设置的key；如果type=1，表示全局共享，那么valueName 的值改为取KEYS[3]，KEYS[3]对应的值为getClientValueName()，查看getClientValueName()源码：

String getClientValueName() {
        return suffixName(getValueName(), commandExecutor.getConnectionManager().getId().toString());
   }

ConnectionManager().getId()如下：

public interface ConnectionManager {
    
    UUID getId();
 
    省略...
}

这个getId()是每个客户端初始化的时候生成的UUID，即每个客户端的getId是唯一的，这也就验证了trySetRate方法中RateType.ALL与RateType.PER_CLIENT的作用。

接着看第7标准行，获取valueName对应的值currentValue；首次获取肯定为空，那么看第10标准行else的逻辑
set valueName 3 px 5，设置key=valueName value=3 过期时间为5秒
decrby valueName 1，将上面valueName的值减1
那么如果第二次访问，第7标注行返回的值存在，将会走第8标注行，紧接着走如下判断
如果当前valueName的值也就是3，小于要获得的令牌数量(tryAcquire方法中的入参)，那么说明当前时间内（key的有效期5秒内），令牌的数量已经被用完，返回pttl(key的剩余过期时间)；反之说明桶中有足够的令牌，获取之后将会把桶中的令牌数量减1，至此结束。

总结：

redission分布式限流采用令牌桶思想和固定时间窗口，trySetRate方法设置桶的大小，利用redis key过期机制达到时间窗口目的，控制固定时间窗口内允许通过的请求量。

更多扩展如下：

　　随着互联网的快速发展，高并发场景下的性能问题愈发凸显。特别是在处理大量数据的定时任务中，如何防止多机并发对下游服务造成过大压力成为了一个亟待解决的问题。传统的单机限流方法虽然能够在一定程度上控制请求量，但在分布式环境下往往无法达到预期效果。这时，Redisson分布式限流的出现为我们提供了一种新的解决方案。

　　Redisson是一个在Redis能力上构建的开发库，它不仅支持Redis的基础操作，还封装了布隆过滤器、分布式锁、限流器等工具。其中，RRateLimiter是Redisson实现分布式限流的核心组件。它基于Redis的分布式特性，能够在多台机器之间协同工作，实现对请求量的精确控制。

　　那么，Redisson分布式限流的实现原理是什么呢？

　　首先，我们需要了解Redisson分布式限流的核心思想：通过Redis的原子操作来保证多个节点之间的限流策略一致。RRateLimiter内部使用了Redis的Lua脚本功能，确保了在执行限流操作时的原子性。

　　其次，RRateLimiter采用了滑动窗口算法来计算请求速率。在滑动窗口内，RRateLimiter会记录通过的请求数量，并根据窗口大小和时间间隔来判断是否超出限流阈值。当请求速率超过阈值时，后续请求将被拒绝，从而实现限流效果。

　　此外，Redisson还提供了丰富的限流策略，如固定窗口限流、漏桶算法限流等。这些策略可以根据具体场景进行灵活选择，以满足不同业务需求。

　　在实际应用中我们可以通过以下步骤来使用Redisson实现分布式限流：

引入Redisson依赖：首先，我们需要在项目中引入Redisson的依赖，以便能够使用其提供的分布式限流功能。
创建Redisson客户端：然后，我们需要创建一个Redisson客户端实例，用于与Redis服务器进行通信。
配置RRateLimiter：接下来，我们需要配置RRateLimiter的参数，包括限流阈值、时间窗口大小等。这些参数将决定限流策略的具体实施。
使用RRateLimiter进行限流：最后，在业务代码中，我们可以使用RRateLimiter实例来进行限流操作。当接收到请求时，先通过RRateLimiter判断是否超出限流阈值，若超出则拒绝请求，否则继续处理。

　　通过以上步骤，我们可以轻松地在分布式环境下实现高效的限流功能。Redisson分布式限流不仅解决了多机并发对下游服务造成过大压力的问题，还提供了灵活的限流策略和丰富的功能支持。在实际应用中，我们可以根据具体需求选择合适的限流策略，并结合Redisson的其他功能来提升系统的性能和稳定性。

总之，Redisson分布式限流是一种高效、灵活且易于实现的解决方案，对于处理高并发场景下的性能瓶颈问题具有重要意义。通过深入了解其实现原理和应用方法，我们可以更好地利用Redisson库来优化系统性能，提升用户体验。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - GL_BKY

前提：

一、使用

二、原理

总结：