大家好，我是【架构摆渡人】，一只十年的程序猿，这是流量治理系列的第11篇原创文章，如果有收获，还请分享给更多的朋友。

假设我们要做一个业务需求，这个需求就是限制用户的访问频次。比如1分钟内只能访问20次，10分钟内只能访问200次。因为是用户维度的场景，性能肯定是要首先考虑，那么适合这个场景的非Redis莫属。

最简单的实现，莫过于只是用incr进行计数操作，于是有了下面的代码：

long count = redisTemplate.opsForValue().increment("user:1:60");
if (count > maxLimitCount) {
   throw new LimitException("访问太频繁");
}

count = redisTemplate.opsForValue().increment("user:1:600");
if (count > maxLimitCount) {
   throw new LimitException("访问太频繁");
}

来，我们对上面这段代码解读一下。需求有2个时间维度的限制，所以这边基于用户和时间维度构建了Redis的Key。然后对每个Key进行计数，计数后的结果用于跟限制的值进行判断，如果超出了限制的值就抛出异常。

假设限制的时间场景有10个呢？那上面的代码是不是得写10遍才可以。有人可能会说，这还不简单吗？循环呀，循环确实能够解决这个问题。但是大家有没有去思考，这是用户维度的请求，如果每个请求里面都去操作10次Redis的话，这耗时至少也得10来毫秒吧。所以问题在这，并不是说这个逻辑实现的有问题。

那我们就改成批量的吧，用pipeline来批量执行。

redisTemplate.execute(new RedisCallback<Long>() {
    @Override
    public Long doInRedis(RedisConnection connection) throws DataAccessException {
        connection.openPipeline();
        connection.incr("user:1:60".getBytes());
        connection.incr("user:1:600".getBytes());
        onnection.closePipeline();
        return null;
    }
});

用pipeline也有一个问题，那就是拿不到返回值，也就只能增加，但是没办法判断是否超过了限制的阀值。

所以需要在第一步先查询下，用查到的值进行判断，这样也就是只需要和Redis交互两次就可以了。

上面的代码在单节点下没问题，但是如果在集群下，其实每个Key都可能分配到不同的节点上去，只不过是底层帮你屏蔽掉了细节，并发执行，拿到了所有结果后合并返回的。所以我们需要让所有的Key都路由到一个节点上，本来就是用户维度的，直接使用userId路由即可。

这个时候Redis的HashTag功能就排上用场了，将Key user:1:600改写成user:{1}:600 。

虽然已经优化了，但是还是要发起两次网络请求才能完成这个逻辑，有没有可能再进一步优化下呢？一次请求行不行。

这个时候要放大招了，Lua脚本走起，将所有逻辑都放入Lua脚本中，一次网络交互即可完成。

local current
current = redis.call("incr",KEYS[1])
if current == 1 then
    redis.call("expire",KEYS[1],1)
end

if current > ARGV[1]
  return 1
end

return 0

上面脚本演示了如何对一个Key进行处理，返回1表示限流，返回0表示通过。不过使用lua脚本的时候要注意，某些云服务的Redis会对脚本进行校验，像Redis的Key不能使用变量，必须用KEYS[下标]的方式，所以这里操作多个Key还不能用循环，代码得写多遍，这是一个恶心的点。

原创：架构摆渡人（公众号ID：jiagoubaiduren），欢迎分享，转载请保留出处。

本文已收录至学习网站 http://cxytiandi.com/ ，里面有Spring Boot, Spring Cloud，分库分表，微服务，面试等相关内容。