基于redis实现最简单的计数器限流算法

   
        org.springframework.boot
        spring-boot-starter-data-redis
   

   
        org.springframework.boot
        spring-boot-starter-test
        test
   

   
        com.fasterxml.jackson.core
        jackson-databind
        2.12.3
   

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory){
        //使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值（默认使用JDK的序列化方式）
        Jackson2JsonRedisSerializer

@Component
public class RedisManager {
    protected static final Logger logger = LoggerFactory.getLogger(RedisManager.class);

    @Autowired
    private RedisTemplate redisTemplate;

    public String get(String key){
        Object value = redisTemplate.opsForValue().get(key);
        if (value instanceof String){
            return (String) value;
        }
        return null;
    }

    public void set(String key,String value){
        redisTemplate.opsForValue().set(key,value);
    }

    public long counter(String key, Integer dela,Integer initial,Long expiry){
        return (long) redisTemplate.execute((RedisCallback) connection -> {
            try {
                long result = 0L;
                if (expiry == 0L) {
                    if (connection.setNX(key.getBytes(), (initial + "").getBytes())) {
                        result = initial;
                    } else {
                        result = connection.incrBy(key.getBytes(), dela);
                    }
                } else if (expiry > 0L) {
                    if (connection.setNX(key.getBytes(), (initial + "").getBytes())) {
                        result = initial;
                        connection.expire(key.getBytes(), expiry);
                    } else {
                        result = connection.incrBy(key.getBytes(), dela);
                        long ttl = connection.ttl(key.getBytes());
                        if (ttl == -1L) {
                            connection.expire(key.getBytes(), expiry);
                        }
                    }
                } else if (connection.exists(key.getBytes())) {
                    result = connection.incrBy(key.getBytes(), dela);
                }

                return result;
            } catch (Exception var6) {
                logger.error("count: ", var6);
                return -1L;
            }
        });
    }

    public void updateExpiry(String key,Long expiry){
        redisTemplate.expire(key,expiry, TimeUnit.SECONDS);
    }
}

@Component
public class RedisRateLimiter {
    protected static final Logger logger = LoggerFactory.getLogger(RedisRateLimiter.class);

    @Autowired
    public RedisManager redisManager;

    public boolean acquire(String key){
        if (StringUtil.isNullOrEmpty(key)){
            return false;
        }
        ExecutorService executorService = getExecutorService();
        String persecond = redisManager.get("DISTT_CRDT_ACC_PERSECOND");
        Integer maximum = persecond==null?0:Integer.valueOf(persecond)  ;
        if (maximum<1){
            return false;
        }
        while (getCounter(key, executorService)>maximum){
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
               logger.error("现场休眠异常{}",e.getMessage());
            }
        }
        return true;
    }

    private long getCounter(String key, ExecutorService executorService) {
        long time = System.currentTimeMillis() / 1000;
        long counter = redisManager.counter(key + time, 1,1,3L);
        executorService.execute(()->{
            try {
                // 因不是lua脚本无法保证原子性，故兜底设置限流key自动过期
                redisManager.updateExpiry(key + time,3L);
            }catch (Exception e){
                logger.debug("更新过期时间异常",e);
            }
        });
        return counter;
    }

    public ExecutorService getExecutorService(){
        return ExecutorServiceHolder.executorService;
    }

    static class ExecutorServiceHolder{
        private static ExecutorService  executorService=null;
        static {
            executorService = new ThreadPoolExecutor(8, 8,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue(100000), new DefaultThreadFactory("ExecutorServiceHolder")
                    , new ThreadPoolExecutor.AbortPolicy());
        }
    }
}

@SpringBootTest
class RedisApplicationTests {
    protected static final Logger logger = LoggerFactory.getLogger(RedisApplicationTests.class);

    @Autowired
    private RedisManager redisManager;
   @Autowired
    RedisRateLimiter redisRateLimiter;
    @Test
    void single() {
        redisManager.set("DISTT_CRDT_ACC_PERSECOND","10");
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            redisRateLimiter.acquire("wdz");
        }
        long end = System.currentTimeMillis();
        logger.info("请求开始时间：{}，请求结束时间:{}，请求耗时:{}",start/1000,end/1000,end-start);
    }

    @Test
    void multi() {
        redisManager.set("DISTT_CRDT_ACC_PERSECOND","10");
        long start = System.currentTimeMillis();
        ExecutorService executorService = Executors.newCachedThreadPool();

        for (int i = 0; i < 100; i++) {
            executorService.execute(()->redisRateLimiter.acquire("wdz"));
        }
        executorService.shutdown();
        while (true){
            if (executorService.isTerminated()){
                break;
            }
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        long end = System.currentTimeMillis();
        logger.info("请求开始时间：{}，请求结束时间:{}，请求耗时:{}",start/1000,end/1000,end-start);
    }

}