序

雙向隊(duì)列(Deque),是Queue的一個(gè)子接口，雙向隊(duì)列是指該隊(duì)列兩端的元素既能入隊(duì)(offer)也能出隊(duì)(poll)。使用場(chǎng)景比如工作竊取，比如限流。

限流實(shí)例

使用deque來(lái)限流，其中timeIntervalInMs為事件窗口，maxLimit為該事件窗口的最大值。

public class MyRateLimiter {

    private static final Logger LOGGER = LoggerFactory.getLogger(DemoRateLimiter.class);

    private final Deque<Long> queue;

    private long timeIntervalInMs;

    public MyRateLimiter(long timeIntervalInMs, int maxLimit) {
        this.timeIntervalInMs = timeIntervalInMs;
        this.queue = new LinkedBlockingDeque<Long>(maxLimit);
    }

    public boolean incrAndReachLimit(){
        long currentTimeMillis = System.currentTimeMillis();
        boolean success = queue.offerFirst(currentTimeMillis);
        if(success){
            //沒(méi)有超過(guò)maxLimit
            return false;
        }

        synchronized (this){
            //queue is full
            long last = queue.getLast();

            //還在時(shí)間窗口內(nèi),超過(guò)maxLimit
            if (currentTimeMillis - last < timeIntervalInMs) {
                return true;
            }
            LOGGER.info("time window expired,current:{},last:{}",currentTimeMillis,last);
            //超過(guò)時(shí)間窗口了,超過(guò)maxLimit的情況下,重置時(shí)間窗口
            queue.removeLast();
            queue.addFirst(currentTimeMillis);

            return false;
        }

    }
}

測(cè)試

@Test
    public void testDeque() throws InterruptedException {
        DemoRateLimiter limiter = new DemoRateLimiter(5*1000,3);
        Callable<Void> test = new Callable<Void>(){

            @Override
            public Void call() throws Exception {
                for(int i=0;i<1000;i++){
                    LOGGER.info("result:{}",limiter.incrAndReachLimit());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return null;
            }
        };
        ExecutorService pool = Executors.newFixedThreadPool(10);
        pool.invokeAll(Arrays.asList(test,test,test,test,test));

        Thread.sleep(100000);
    }

小結(jié)

這里使用了Deque的容量來(lái)作為時(shí)間窗口的限流大小，利用兩端來(lái)判斷時(shí)間窗口，相對(duì)來(lái)講有點(diǎn)巧妙。