假設我們要做一個業務需求，這個需求就是限制用戶的訪問頻次。比如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() {
    @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還不能用循環，代碼得寫多遍，這是一個惡心的點。

總結