本文作者是来自国外一家名为ClassDojo的科技公司，其分享了在企业构建推送通知服务的限流实践。该服务要求具备以下标准的限流功能：
- 分布式：限流器可以在多个进程之间共享。这就需要使用外部键值存储，我们选择了Redis，因为系统的其他地方也使用了redis。
- 滚动窗口：如果我们设置每分钟最多10条消息，我们不希望用户在0:59能收到10条消息，在1:01又收到10条消息。
- 消息之间要保持最小间隔时间：不管限流rate选择设置多大，强制要求连续的消息之间保持最小间隔，这样接收者忙碌的时候不会有烦人的声音或提醒。
我们查看了可用的限流器选择，但在NPM（node.js包管理，他们的服务应该是基于node.js开发）未找到符合上述要求的包。所以决定自己写，你可以在这里下载。
方案一：token buckets(令牌桶)
使用滚动窗口限流的经典算法是令牌桶(或漏桶算法)。下面是它的工作原理:
- 每个用户都有一个关联的桶，其中包含许多令牌（tokens）。
- 用户发起推送操作，就检查对应桶里tokens的数量。
- 如果桶是空的，用户已经达到限流条件，操作就被阻止。
- 否则，就从桶中拿掉一个token（对于特殊操作可以多拿几个tokens）然后正常执行操作，例如推送通知。
- 随着时间的推移，我们会以设定的速率将所有桶重新装满tokens，直到桶满为止。
这是一种非常聪明的算法，只占用很少的空间(每个用户只占用一个整数计数器)，但是这种直接的实现有一个很大的缺陷——一些进程需要不断地填充桶。由于有数百万用户，而且每次填充操作都需要写操作，这对我们的Redis实例来说是一个不可持续的负载。这里有一个更高级的方法:
- 每个用户都有两个与之相关的键:令牌桶，以及桶最后一次被重新填充的时间戳。
- 当用户试图执行一个操作时，我们获取存储的时间戳。
- 我们计算自上次请求以来应该向用户授予多少tokens。
- 然后，我们可以使用这个新的token计数继续算法。
不幸的是，这个算法也有问题：当两个进程需要共享限流器时，它将失败。Redis可以将操作批处理为一个原子操作，但要计算给用户多少令牌，我们至少需要两次访问Redis：一次获取最后的时间戳，一次设置新令牌计数。如果不深入研究Redis Lua脚本(我们没有任何经验，而且很难在测试中模拟)，我们无法找到一种方法将所有需要的操作组合成一个单一的Redis原子操作。正因为如此，如果两个使用限流器的客户端都试图以同一个用户操作，我们可以得到以下操作序列: - 用户有足够的token。
- 自上次操作以来，还没有足够的时间来授予更多的令牌。
- 客户端1获取存储的时间戳和令牌计数。
- 客户端2获取存储的时间戳和令牌计数。
- 客户端1计算不需要添加令牌，允许操作，并告诉redis设置令牌计数为0。
- 客户端2计算也不需要添加令牌，允许操作，并告诉redis设置令牌计数为0。
不用说，这并不理想。如果我们有几十个进程在处理推送通知负载，那么用户可能会同时收到数十个推送。
更好方法：sorted sets（有序集合）
幸运的是，Redis有另一个数据结构，我们可以用来防止这些竞争条件-有序集合。这是我们想出的算法:
- 每个用户有一个有序集合。key和value相同，等于执行操作时的(微秒)时间。
- 当用户试图执行操作，我们首先删除集合中所有在一个间隔之前发生的元素。这可以通过Redis的ZREMRANGEBYSCORE命令来完成。
- 使用ZRANGE(0， -1)来获取集合中的所有元素
- 使用ZADD将当前时间戳添加到集合中
- 设置TTL等于限流间隔(节省空间)。
- 在所有操作完成后，我们计算获取的元素的数量。如果它超过了限制，不允许操作执行。
- 还可以将获取的最后添加的元素与当前的时间戳进行比较。如果他们离得太近，我们也不允许操作。
- 这种方法的优点是，所有的Redis操作都可以作为一个原子操作执行，使用MULTI命令。这意味着，如果两个进程都试图以同一用户执行一个操作，它们就不可能没有最新的信息，从而防止上述问题的发生。它还允许我们对想要跟踪的两种速率使用一个限流器(即每分钟不超过10条消息或每3秒不超过2条消息)。
然而，这种方法有一个瑕疵——我们不受影响，但可能不适合他人的要求。在我们的算法中，你可以看到一个操作是否被阻塞是在所有Redis操作完成后决定的。这意味着被阻止的操作仍然算作操作。所以，如果用户持续超过速率限制，他们的任何操作都将不被允许(在前几次之后)，而不是允许偶尔的操作通过。
模块
我们在npm上开源了我们的限制器，作为rolling-rate-limiter模块。它可以使用Redis后端，或者，如果你不需要在多进程中运行你的限流器，在内存中操作(使用一个简单的数组而不是排序集)。这里有一个例子:
/* Setup: */
var RateLimiter = require("rolling-rate-limiter");
var Redis = require("redis");
var client = Redis.createClient(config);
var limiter = RateLimiter({
redis: client,
namespace: "UserLoginLimiter"
interval: 1000
maxInInterval: 10
minDifference: 100
});
/* Action: */
function attemptAction(userId, cb) {
limiter(userId, function(err, timeLeft) {
if (err) {
// redis failed or similar.
} else if (timeLeft > 0) {
// limit was exceeded, action should not be allowed
// timeLeft is the number of ms until the next action will be allowed
// note that this can be treated as a boolean, since 0 is falsy
} else {
// limit was not exceeded, action should be allowed
}
});
}
你也可以很容易地使用它与中间件结合对请求限速，如下所示:
var limiter = RateLimiter({
redis: redisClient,
namespace: "requestRateLimiter",
interval: 60000,
maxInInterval: 100,
minDifference: 100
});
app.use(function(req, res, next) {
// "req.ipAddress" could be replaced with any unique user identifier
limiter(req.ipAddress, function(err, timeLeft) {
if (err) {
return res.status(500).send();
} else if (timeLeft) {
return res.status(429).send("You must wait " + timeLeft + " ms before you can make requests.");
} else {
return next();
}
});
});
你可以在Github上查看源代码。我们希望这将有助于您编写Node.js应用程序!
总结
本文介绍了一种基于Redis有序集合实现的分布式限流器，该方法特别适合软件通知服务限流。当然每种算法都不是放之四海而皆准的，需要根据自己的场景来选择或者改造。虽然文章使用的是node.js实现，读者也可以尝试用Go来实践下。
