redis実現限流

エンドシステムの性能の問題のため、ストリーム制限機能を設計する必要があり、redis listを採用してスライドウィンドウのストリーム制限戦略を行う.

 local function addToQueue(x,time)
     local count=0
     for i=1,x,1 do
         redis.call('lpush',KEYS[1],time)
         count=count+1
     end
     return count
 end
 local result=0
 local timeBase = redis.call('lindex',KEYS[1], tonumber(ARGV[2])-tonumber(ARGV[1]))
 if (timeBase == false) or (tonumber(ARGV[4]) - tonumber(timeBase)>tonumber(ARGV[3])) then
   result=result+addToQueue(tonumber(ARGV[1]),tonumber(ARGV[4]))
 end
 if (timeBase~=false) then
    redis.call('ltrim',KEYS[1],0,tonumber(ARGV[2])*3)
 end
 return result

package com.bestpay.messagecenter.product.core.redis.impl;

import com.bestpay.messagecenter.product.common.constant.RedisProductKeys;
import com.bestpay.messagecenter.product.common.util.StreamUtil;
import com.bestpay.messagecenter.product.core.redis.ConfigRedisService;
import com.bestpay.messagecenter.product.core.redis.QosRedisService;
import com.google.common.base.Throwables;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import redis.clients.jedis.Jedis;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

/**
 * redis list         
 * @author lxn
 * @version Id: QosRedisServiceImpl.java, v 0.1 2017/6/30 17:36 lxn Exp $$
 */
@Slf4j
@Service
public class QosRedisServiceImpl implements QosRedisService {

    /**
     * Jedis    
     */
    @Resource
    private JedisConnectionFactory jedisConnectionFactory;

    /**
     *    sha1
     */
    private String scriptShal;

    @Autowired
    private ConfigRedisService configRedisService;

    /**
     *     lua   redis
     */
    @PostConstruct
    public void loadScript() {
        JedisConnection connection = jedisConnectionFactory.getConnection();
        Jedis jedis = connection.getNativeConnection();
        String script = StreamUtil.convertStreamToString(FreqRedisServiceImpl.class.getClassLoader().
                getResourceAsStream("qosScript.lua"));
        this.scriptShal = jedis.scriptLoad(script);
        log.info("            ，sha1：{}", this.scriptShal);
        connection.close();
    }
    /**
     *      
     * @param count      
     * @param rateCount     
     * @param rateTime        
     * @return
     */
    @Override
    public long acquirePromise(String redisKey,long count, long rateCount, long rateTime) {
        Assert.hasText(redisKey,"  key    ");
        Assert.isTrue(count>0,"         0");
        Assert.isTrue(rateCount>0,"        0");
        Assert.isTrue(rateTime>0,"        0");
        List keys = new ArrayList<>();
        keys.add(redisKey);//   
        List values = new ArrayList<>();
        values.add(String.valueOf(count)); //        1
        values.add(String.valueOf(rateCount));//     2
        values.add(String.valueOf(rateTime));//    （  ）3
        values.add(String.valueOf(System.currentTimeMillis()));//     4
        JedisConnection connection=null;
        try {
            connection = jedisConnectionFactory.getConnection();
            Jedis jedis = connection.getNativeConnection();

            Object evalResult = jedis.evalsha(scriptShal, keys, values);
            return Long.parseLong(evalResult.toString());
        }finally {
            if(connection!=null) {
                connection.close();
            }
        }
    }

    /**
     *     
     * @param count
     * @param rateCount
     * @param rateTime
     */
    @Override
    public void acquirePromiseBlock(String redisKey,long count, long rateCount, long rateTime) {
        while (acquirePromise(redisKey,count, rateCount, rateTime)<=0){
            int sleepTime = configRedisService.getConfigInt(RedisProductKeys.getCfgQosLimitThreadSleepTime());
            try {
                Thread.sleep(sleepTime);
            }catch (InterruptedException e){
                log.error("          {}", Throwables.getStackTraceAsString(e));
                //       
                Thread.currentThread().interrupt();
            }
        }
    }
}