Rate-limited thread scheduler in Java

Question

Background: I'm planning to call different endpoints of a website's API quickly, but the usage policy implies that I may only make five calls in a second. I can't rely on the calls taking 200 ms on average, so I wrote the following.

The plan is to have any other threads that want to call an API submit a bulk of Callables to the executor below. Each Callable does a call to the API and the executor submits them according to the API's max rate. The results may then be collected later by the submitting thread through the FutureTask.

// RateLimitedExecutorService.java

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;

public class RateLimitedExecutorService extends TimerTask {

    private final long creditDelay;
    private ExecutorService exec;
    private Timer timer;
    private LinkedBlockingQueue<FutureTask<?>> tasks;

    public RateLimitedExecutorService(int ratePerSecs, int capacity) throws InvalidParameterException {

        if (ratePerSecs <= 0) {
            throw new InvalidParameterException("Rate is <= 0");
        }

        this.exec = Executors.newCachedThreadPool();
        this.tasks = new LinkedBlockingQueue<>(capacity);

        this.timer = new Timer(true);
        this.timer.scheduleAtFixedRate(this, 0, this.creditDelay);
    }

    public <T> FutureTask<T> submit(Callable<T> task) throws NullPointerException, InterruptedException {
        FutureTask<T> futureTask = new FutureTask<T>(task);
        tasks.put(futureTask);
        return futureTask;
    }

    @Override
    public void run() {
        FutureTask<?> task = tasks.poll();
        try {
            exec.submit(task);
        } catch (NullPointerException npe) {
            // swallow: to be expected when task list is empty
        } catch (RejectedExecutionException ree) {
            System.out.println("RLExec: can't schedule task!");
        }
    }

    public void shutdown() {
        this.exec.shutdownNow();
    }
}

Here is some example driver code:

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class TestMain {

    public static void main(String[] args) throws Exception {

        ArrayList<TestCallable> tasks = new ArrayList<>();
        ArrayList<FutureTask<Integer>> res = new ArrayList<>();
        RateLimitedExecutorService exec = new RateLimitedExecutorService(5, 100);

        for (int i = 0; i < 20; i++) {
            tasks.add(new TestCallable(i, (int) (Math.random() * 20)));
        }
        for (TestCallable tc : tasks) {
            res.add(exec.submit(tc));
        }
        for (FutureTask<Integer> ft : res) {
            System.out.println("res: " + ft.get());
        }
        exec.shutdown();
    }

    private static class TestCallable implements Callable<Integer> {

        private int x, y;

        public TestCallable(int a, int b) {
            this.x = a;
            this.y = b;
        }

        @Override
        public Integer call() throws InterruptedException {
            Thread.sleep(y * 10);
            return x + y;
        }
    }
}

I don't have too much experience with Java's multithreading apart from using the builtin executors. Any feedback (and hints on how to handle the InterruptedExceptions) is appreciated!

Answer 1

This code will work, but few suggestions :

1. Here we are using timer as well as executors, side by side, the problem here is that you are un-necessarily creating 2 threads, timer is just going to hand over the api call to another thread. Now there can be scenario where due to some reason, the connection establishment might take time and initially latency is higher than 200ms, then in that case, since timer will keep on queueing other tasks, more than 5 tasks might get piled up, and rate limiting might not get honoured, if the latency improves later on.
2. Even though you wanted 5 calls in 1 sec, we are effectively doing 1 call every 200ms,even though it's same, but it's putting un-necessary throttling at your end.

You to take a look at ScheduledThreadPoolExecutor, it does the same thing that you are doing, but this will also suffer from problem 1. In order to resolve that you can have tokenBucket sort of algorithm, you can leverage Semaphore, and have a timer return the token back to Semaphore after x delay.