固定频率调用接口方案

本文来源于一次工作中，对某第三方平台进行二次开发。这个第三方平台提供的接口有一个很扯淡的限制：限制每秒调用频率为最多2次。解决这个问题，有下面几个问题需要解决：

• 第三方调用频率策略修改，能快速修改，适应其策略
• 在具体的业务逻辑不需要关注调用频率的变化

由此，想到使用自定义线程池来解决，调用时候，只需要忘线程池提交任务。其他接口调用频率交给线程池来处理。我使用单线程池，同时修改等待队列为DelayQueue,DelayQueue支持延迟固定时间take一个元素。

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                E first = q.peek();
                if (first == null)
                    available.await();
                else {
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    first = null; // don't retain ref while waiting
                    if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

其中有最关键的是：

 long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();

所以最为关键的试试实现自定义的getDelay方法。由于该对列只支持Delay类型，而普通的线程提交只支持Runnable和Callble，需要做一个适配。

public class DelayedThreadPoolFactoryBean extends ThreadPoolExecutorFactoryBean {
    Logger logger = LoggerFactory.getLogger(this.getClass());
    private TimeUnit timeUnit;
    private long delayed;

    public void setTimeUnit(TimeUnit timeUnit) {
        this.timeUnit = timeUnit;
    }

    public void setDelayed(long delayed) {
        this.delayed = delayed;
    }
    @Override
    protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
        return new DelegateDelayedQueue();
    }

    @Override
    protected ThreadPoolExecutor createExecutor(int corePoolSize,int maxPoolSize,int keepAliveSeconds,final BlockingQueue<Runnable> queue,ThreadFactory threadFactory,RejectedExecutionHandler rejectedExecutionHandler) {

        return new ThreadPoolExecutor(corePoolSize, maxPoolSize,
                keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler){
            @Override
            public void execute(Runnable command) {
                final DelegateDelayedQueue queue1 = (DelegateDelayedQueue)queue;
                DelegateDelaye delegateDelaye = new DelegateDelaye(command,queue1,timeUnit,delayed);
                super.execute(delegateDelaye);
            }
        };
    }
    private class DelegateDelaye implements Delayed,Runnable{
        private Runnable runnable;
        private DelegateDelayedQueue delegateDelayedQueue;
        private TimeUnit timeUnit;
        private long delayed;
        public DelegateDelaye(Runnable runnable,DelegateDelayedQueue delegateDelayedQueue,TimeUnit unit,long delayed){
            this.runnable = runnable;
            this.delegateDelayedQueue = delegateDelayedQueue;
            this.timeUnit = timeUnit;
            this.delayed = delayed;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            if(delegateDelayedQueue.getLast() == 0){
                return -1;
            }
            return unit.convert(delegateDelayedQueue.getLast() +
                    TimeUnit.MILLISECONDS.convert(5,TimeUnit.SECONDS) -
                    System.currentTimeMillis(),TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return 0;
        }

        @Override
        public void run() {
            runnable.run();
        }
    }
    private class DelegateDelayedQueue extends DelayQueue{
        private long last = 0;//默认为0

        @Override
        public Delayed poll() {
            Delayed delayed =  super.poll();
            if(delayed != null){
                this.last = System.currentTimeMillis();
            }
            return delayed;
        }

        @Override
        public Delayed take() throws InterruptedException {
            Delayed delayed =  super.take();
            if(delayed != null){
                this.last = System.currentTimeMillis();
            }
            return delayed;
        }

        @Override
        public Delayed poll(long timeout,TimeUnit unit) throws InterruptedException {
            Delayed delayed = super.poll(timeout,unit);
            if(delayed != null){
                this.last = System.currentTimeMillis();
            }
            return delayed;
        }

        public long getLast() {
            return last;
        }
    }
}

简单测试：

@Test
    public void testSinglePool(){
        ExecutorService executorService = executoryServiceFactory.executorService(ExecutoryServiceFactory.Pool.DELAYED);

        executorService.submit(new Callable<String>() {
           @Override
           public String call() throws Exception {
               System.out.println(TimeUnit.SECONDS.convert(System.currentTimeMillis(),TimeUnit.MILLISECONDS));
               return null;
           }
       });
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(TimeUnit.SECONDS.convert(System.currentTimeMillis(),TimeUnit.MILLISECONDS));
            }
        });
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(TimeUnit.SECONDS.convert(System.currentTimeMillis(),TimeUnit.MILLISECONDS));
            }
        });
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(TimeUnit.SECONDS.convert(System.currentTimeMillis(),TimeUnit.MILLISECONDS));
            }
        });
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(TimeUnit.SECONDS.convert(System.currentTimeMillis(),TimeUnit.MILLISECONDS));
            }
        });
        executorService.shutdown();
        try {
            executorService.awaitTermination(5,TimeUnit.MINUTES);
        } catch (Exception e){}
    }

结果符合预期的效果，固定频率执行任务。