Java blocking queue download process scenario

Question

I have the following scenario .

I have incoming file download requests and each download is happening in different thread until pool size is exceeded. And after a download completed, a processor processes the downloaded item. So I created the following. I wonder if uses of thread and executors make sense

DownloadTaskProcessor

public class DownloadTaskEnqueuer {
    private static final BlockingQueue<Task> downloadQueue = new LinkedBlockingQueue<>();
    private static final BlockingQueue<Task> processQueue = new LinkedBlockingQueue<>();
    private static final ExecutorService executor = Executors.newCachedThreadPool();

    public void offer(Task task) {
        return downloadQueue.offer(task);
    }

    public void createPool(int size) {
        for (int i = 0; i < size; i++) {
            executor.execute(new DownloadTask(downloadQueue, processQueue);
            executor.execute(new ProcessTask(processQueue));
        }
    }
}

Download task

public class DownloadTask implements Runnable {
    private BlockingQueue<Task> downloadQueue;
    private BlockingQueue<Task> processQueue;
    
    // constructor for initing two queue

    public void offer(Task task) {
        return processQueue.offer(task);
    }
    
    @Override
    public void run() {
        while (true) {
           Task task = downloadQueue.poll();
           if (task != null) {
               task.getDownloadTask().download();
               offer(task);    
           } else {
               // sleep 250 ms 
           }
        }
    }
}

Process task

public class ProcessTask implements Runnable {
    private BlockingQueue<Task> processQueue;
    
    // constructor for initing queue
    
    @Override
    public void run() {
        while (true) {
           Task task = processQueue.poll();
           if (task != null) {
                task.getProcessTask().process();
           } else {
               // sleep 250 ms 
           }
        }
    }
}

Use case (pseudo)

createPool(10);

listener.listen((task) -> {
    downloadTaskEnqueuer.offer(task);
}

Answer 1

In your DownloadTaskEnqueuer it might be a little bit weird to mix instance methods with static fields (unless there is more code which you omitted). It might be better to either use instance fields (i.e. remove static from downloadQueue, processQueue and executor) or make the methods static.

You DownloadTaskEnqueuer.offer(...) method has void as return type but contains a return statement returning a value.

Executors.newCachedThreadPool() is not that well suited for your use case. You are creating a fixed number of work tasks which run indefinitely so an executor which creates a new thread per task would work equally well. Normally you also don't submit an indefinitely running work task which itself performs tasks to an executor, but instead directly submit the tasks to the executor.

This leads to the next question: How are you shutting down workers? It appears with your current code the DownloadTask and ProcessTask run forever. Or is there no need to shut down workers in your use case?

Using BlockingQueue.poll() and then waiting seems a little bit inefficient. Since you are already dealing with BlockingQueue you could use the blocking method BlockingQueue.take().

If I may make a suggestion, it appears CompletableFuture would be well suited for your use case:

public class DownloadProcessor {
    private final ExecutorService downloadExecutor;
    private final ExecutorService processingExecutor;

    public DownloadProcessor(int poolsSize) {
        downloadExecutor = Executors.newFixedThreadPool​(poolsSize);​
        processingExecutor = Executors.newFixedThreadPool​(poolsSize);​
    }

    public CompletableFuture<Void> downloadAndProcess(DownloadTask task) {
        return CompletableFuture.supplyAsync(() -> task.download(), downloadExecutor)
            // Assumes that DownloadTask.download() returns ProcessingTask which
            // has method `process()`
            .thenAcceptAsync​(task -> task.process(), processingExecutor);
    }
    
    public void shutdown() {
        downloadExecutor.shutdown();
        processExecutor.shutdown();
        // Maybe also call awaitTermination​
    }
}