Java 2 grouped thread scenario

Question

I have the following multi thread scenario and the codes. It looks kind of messy and want to know if I can make it better or if there is any flaw to fix.

When a request arrives, one thread downloads it and the remaining operations are dependent on download so it needs to wait for download to finish. But there can be many request coming and download tasks must run at the same time but each process operation need to wait for its own download task. So I did the following.

ProcessCompletedListener

public interface ProcessCompletedListener {
    void onComplete(Object object);
}

RequestListener

public interface RequestListener {
    void onRequest(Object request);
}

Receiver class where I send request for testing purpose.

public class Receiver {
private RequestListener requestListener;

public void setRequestListener(RequestListener requestListener) {
    this.requestListener = requestListener;
}

public void requestBomb() {
    String[] names = new String[]{"a", "b", "c"};

    int i = 0;
    while (i < 3) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        requestListener.onRequest(names[i]);
        i++;
    }
}
}

Custom blocking queue

public class CustomBlockingQueue {
private BlockingQueue<Object> blockingQueue = new LinkedBlockingDeque<>();
private ProcessCompletedListener processCompletedListener;

public boolean offerAndProcess(Object object) {
    if (blockingQueue.offer(object)) {
        process(object);
        return true;
    }
    return false;
}

public void setProcessCompletedListener(ProcessCompletedListener processCompletedListener) {
    this.processCompletedListener = processCompletedListener;
}

private void process(Object object) {
    new Thread(() -> {
        System.err.println(object + ": Request process started.");

        try {
            Thread.sleep(6000); // mock for real operation
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.err.println(object + ": Request process completed.");

        if (processCompletedListener != null) {
            processCompletedListener.onComplete(object);
        }
    }).start();
}
}

Demo class

public class Demo {
public CustomBlockingQueue customBlockingQueue;

public void setCustomBlockingQueue(CustomBlockingQueue customBlockingQueue) {
    this.customBlockingQueue = customBlockingQueue;
}

public CustomBlockingQueue getCustomBlockingQueue() {
    return customBlockingQueue;
}

public static void main(String[] args) {
    Demo demo = new Demo();
    demo.setCustomBlockingQueue(new CustomBlockingQueue());

    Receiver receiver = new Receiver();
    Thread t = new Thread(() -> {
        receiver.setRequestListener((request) -> {
            new Thread(() ->
                    demo.getCustomBlockingQueue().setProcessCompletedListener((Object object) -> {
                        System.err.println(object + ": post process started");
                        try {
                            Thread.sleep(1500); // mock for real operation
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.err.println(object + ": post process completed");
                    })).start();
            demo.getCustomBlockingQueue().offerAndProcess(request);
        });
    });
    t.start();
    receiver.requestBomb();
}
}

Answer 1

When a request arrives, one thread downloads it and the remaining operations are dependent on download so it needs to wait for download to finish. But there can be many request coming and download tasks must run at the same time but each process operation need to wait for its own download task.

That does sound like you want it in the same thread, like a "standard" webserver actually. A request is routed to a thread which handles that request from start to finish. No waiting or synchronizing involved. These threads can also easily be pooled.

---

    int i = 0;
    while (i < 3) {

I find it curios that you've chosen a while over for here.

---

        try {
            Thread.sleep(6000); // mock for real operation
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

AS a note, you most likely want to exit here. The InterruptedException is being thrown when the Thead has been asked to stop doing whatever it does, so you don't want to ignore it most of the time.

---

    t.start();
    receiver.requestBomb();

Mind you that the Thread might not have started at this point. start does start the thread, obviously, but whether it has already executed something when your logic reaches the next instruction is undefined. Whether receiver.setRequestListener(...) has been called when you call threadBomb is undefined, might not have been.

---

Your code looks and reads fine, actually. It's well written, however, it is unnecessarily complicated, or at least I'm failing to see why you need to do it this way.

What would make sense is to have a single thread pool for processing, which is being fed by a single queue. A little pseudo-code for that:

Queue processingQueue = new BlockingQueue();

class RequestReceiver {
    public void onRequestReceived(Reqquest request) {
        processingQueue.add(request);
    }
}

class ProcessingThread {
    public void run() {
        while (aliveCondition) {
            Request request = processingQueue.poll();
            download(request);
            process(request);
        }
    }
}

This scheme is in, some form or another (rather another), also used by servers and webservers to process requests. You spin up multiple ProcessingThreads and have them wait for something to do, and after being done, they go back into the idle state.

If you now require different threads for download and processing, you're just adding a another queue, like this:

Queue downloadingQueue = new BlockingQueue();
Queue processingQueue = new BlockingQueue();

class RequestReceiver {
    public void onRequestReceived(Reqquest request) {
        downloadingQueue.add(request);
    }
}

class DownloadingThread {
    public void run() {
        while (aliveCondition) {
            Request request = downlaodingQueue.poll();
            download(request);
            processingQueue.add(request);
        }
    }
}

class ProcessingThread {
    public void run() {
        while (aliveCondition) {
            Request request = processingQueue.poll();
            process(request);
        }
    }
}

---

    t.start();
    receiver.requestBomb();

Mind you that the Thread might not have started at this point. start does start the thread, obviously, but whether it has already executed something is undefined. The