我们目前在工作中遇到一个性能问题。我们有一个计划任务需要处理大量的数据。为了提高吞吐量，我们部署了很多机器，但是这个任务在运行之前，需要从其他服务中拉取大量的数据。随着数据量的增加，如果多台机器同时并发拉取数据，会对下游服务造成很大的压力。之前加了单机限流，但是不能解决问题，因为只有不到10%的数据任务在拉取数据。如果单机限流过严，虽然集群的总请求量得到了控制，但是任务Throughput又下降了。如果限流阈值过高，在多机并发运行时仍有可能压垮下游。所以目前唯一可行的方案就是分布式限流。我目前选择直接使用Redisson库中的RRateLimiter来实现分布式限流。很多人可能听说过Redission。它实际上是一个基于Redis能力构建的开发库。除了支持Redis的基本操作外，还封装了Bloom过滤器、分布式锁、限流器等工具，进行了引入。今天我们要说的RRateLimiter就是它实现的限流器。接下来本文将详细介绍RRateLimiter的具体使用方法、实现原理以及一些注意事项。最后简单说说我对分布式限流底层原理的理解。RRateLimiter的使用RRateLimiter的使用非常简单，参数也不多。只要创建RedissonClient，就可以从客户端获取RRateLimiter对象，直接看代码示例。RedissonClientredissonClient=Redisson.create();RRateLimiterrateLimiter=redissonClient.getRateLimiter("xindoo.limiter");rateLimiter.trySetRate(RateType.OVERALL,100,1,RateIntervalUnit.HOURS);复制代码rateLimiter.trySetRate是设置限流参数，有两种RateType，OVERALL是全局限流，PER_CLIENT是单Client限流（可以认为是单机限流），这里只讨论全局模式。最后三个参数的作用是设置时间窗口（rateInterval+IntervalUnit），总许可数不超过（rate）。上面代码我设置的值是1小时内的license总数不超过100个，然后调用rateLimiter的tryAcquire()或acquire()方法获取license。rateLimiter.acquire(1);//申请1个license直到成功booleanres=rateLimiter.tryAcquire(1,5,TimeUnit.SECONDS);//申请1个license，如果5s内没有，放弃复制代码use还是很简单的。上面代码中的两个方法是同步调用，但是Redisson也提供了异步方法acquireAsync()和tryAcquireAsync()，返回的RFuture可以用来异步获取license。RRateLimiter的实现接下来，我们跟随tryAcquire()方法来看看它是如何实现的。在RedissonRateLimiter类中，我们可以看到底层的tryAcquireAsync()方法。privateRFuturetryAcquireAsync(RedisCommandcommand,Longvalue){byte[]random=newbyte[8];ThreadLocalRandom.current().nextBytes(随机);返回commandExecutor.evalWriteAsync(getRawName(),LongCodec.INSTANCE,命令,"————————————————————————————————————————”+“这里是一大段lua代码”+“________________________________________”,Arrays.asList(getRawName(),getValueName(),getClientValueName(),getPermitsName(),getClientPermitsName()),value,System.currentTimeMillis(),random);}复制代码映入眼帘的是一大段Lua代码。其实这段Lua代码就是限流实现的核心。我提取了这段Lua代码并添加了一些注释。让我们详细看一下。localrate=redis.call("hget",KEYS[1],"rate")#100localinterval=redis.call("hget",KEYS[1],"interval")#3600000localtype=redis.call("hget",KEYS[1],"type")#0assert(rate~=falseandinterval~=falseandtype~=false,"RateLimiterisnotinitialized")localvalueName=KEYS[2]#{xindoo.limiter}:value用于存储剩余的许可数。localpermitsName=KEYS[4]#{xindoo.limiter}:permits记录所有颁发的许可证的时间戳。区分iftype=="1"thenvalueName=KEYS[3]#{xindoo.limiter}:value:b474c7d5-862c-4be2-9656-f4011c269d54permitsName=KEYS[5]#{xindoo.limiter}:permits:b474c7d5-862c-4be2-9656-f4011c269d54end检查参数assert(tonumber(rate)>=tonumber(ARGV[1]),"Requestedpermitsamountcouldnotexceeddefinedrate")获取当前还剩多少许可证localcurrentValue=redis.call("get",valueName)localres如果有记录当前还剩多少个licenseifcurrentValue~=falsethen#回收过期的licenses数量localexpiredValues=redis.call("zrangebyscore",permitsName,0,tonumber(ARGV[2])-interval)localreleased=0fori,vinipairs(expiredValues)dolocalrandom,permits=struct.unpack("Bc0I",v)released=released+permitsend#清理过期许可记录ifreleased>0thenredis.call("zremrangebyscore",permitsName,0,tonumber(ARGV[2])-interval)如果tonumber(currentValue)+released>tonumber(rate)thencurrentValue=tonumber(rate)-redis.call("zcard",permitsName)elsecurrentValue=tonumber(currentValue)+releasedendredis.call("set",valueName,currentValue)end#ARGVpermittimestamprandom,random是一个随机的8字节#如果剩余权限不够，iftonumber(currentValue)0thenredis.call("pexpire",valueName,ttl)redis.call("pexpire",permitsName,ttl)endreturnres复制代码即使加了注释，相信这段代码你还是很难一下子看懂，接下来我会利用它在Redis中的数据存储形式，补充一个流程一张图让大家看个透彻。实现原理首先用RRateLimiter有一个名字，在我的代码中是xindoo.limiter。以此为KEY，可以在Redis中找到一个map，里面存储了limiter的工作模式（type），数量（rate），timewindowsize（interval），在创建limiter的时候写入redis，也是在上面的lua代码中使用。其次，有两个很重要的key，valueName和permitsName，在我的代码中实现了其中的valueName是{xindoo.limiter}:value，存放的是当前可用的license数量。我的代码中permitsName的具体值是{xindoo.limiter}:permits，是一个zset，存储了当前所有的license授权记录（包括授权时间戳），其中SCORE直接使用时间戳，VALUE包含8字节随机值和许可数量，如下图所示：{xindoo.limiter}:permits这个zset存储了所有的历史授权记录。了解了这些信息，相信你就明白了RRateLimiter的实现原理了。我们还是把上面大段Lua代码的流程图画出来，整个执行过程会更加直观。看到这里，大家应该能明白这段Lua代码的逻辑了。可以看到Redis使用了多个字段来存储限流信息，还有各种操作。Redis如何保证这些限制在分布式的情况下是分布式的呢？流量信息数据的一致性？答案是没有保证。在这种场景下，信息自然是一致的。原因是Redis的单进程数据处理模型。同一个Key下，所有的eval请求都是串行的，不需要考虑并发数据操作的问题。这里，Redisson也使用了HashTag来保证所有的限流信息都存储在同一个Redis实例上。使用RRateLimiter的注意事项在了解了RRateLimiter的底层原理后，结合Redis本身的特点，想到了RRateLimiter的几个局限性（问题）。RRateLimiter是一个不公平的限流器这个是我从资料上了解到的，在我自己的代码实践过程中也得到了验证。具体表现为，如果多个实例（机器）竞争这些license，很可能有一些实例会拿到。他们中的大多数，以及其他一些可怜的例子只获得了少量的许可证，这意味着干旱和洪水容易发生旱涝灾害。在使用的过程中，你必须要考虑你是否可以接受这种情况，如果你不能接受，你必须考虑使用一些方法让它尽可能的公平。Rate不要设置太大从RRateLimiter的实现原理也可以看出，它是采用滑动窗口的方式限流，并且记录了所有的许可信息，所以如果Rate值设置太大，在Redis中文件中存放的信息（zset对应permitsName）越多，每次执行那个lua脚本的性能就越差，这也是对Redis实例的一种压力。我个人的建议是，如果要设置更大的限流阈值，倾向于使用小Rate+小时间窗的方式，这种设置方式要求会更加统一。限流上限取决于Redis单实例的性能原则上RRateLimiter在Redis中存储的信息必须在一个Redis实例上，所以它的限流QPS上限就是Redis单实例的上限，比如作为你的Redis的例子是1wQPS。如果想用RRateLimiter实现一个2wQPS的限流器，那肯定不行。有没有办法突破Redis单实例的性能限制？单个限流器肯定是不行的。我们可以拆分多个限流器。比如我做10个限流器，名词不一样，然后每台机器随机用一个限流器来限流。实际不是把流量分配到不同的限流器上了吗？总限流不上去吗？分布式限流的本质分布式限流的本质其实是协作，协作的本质是信息交换。信息交流最重要的是信息的准确性和一致性。更简单粗暴地理解，分布式限流的本质原则其实就是分布式数据一致性原则，限流只是对数据结果的决策。所以只要能以任何方式同步信息，并保证信息的正确性，就可以实现分布式限流器。这是我理解的基本思想。其实从上面RRateLimiter的实现原理可以看出，它只是存储了一些信息！那我不用Redis，那我用mysql可以吗？其实肯定是可以的，只要将上面Lua代码中的所有操作都放到一个事务中，将事务级别改为序列化，仍然可以实现同样的RRateLimiter的功能。如果有Mysql相关知识，肯定可以基于Mysql封装RRateLimiterAPI，但是封装的限流器的上限取决于Mysql实例的性能上限。最近，chatGPT相当流行。我还询问了它对分布式限流基本原理的理解。以下是它的回答。你怎么认为？分布式限流的本质原理是通过在分布式系统中共享限流状态来限制系统在单位时间内的请求数，从而避免系统因流量过大而崩溃。这是通过使用一些共享存储组件如数据库、缓存、分布式锁等来实现的。在每次请求时，系统都会检查当前的请求数是否超过了预设的限制。如果超过限制，请求将被拒绝；如果不超过限制，请求将被允许。通过采用分布式限流技术，系统可以在高并发下保持稳定的性能，避免因流量过大导致系统崩溃。
