4
\$\begingroup\$

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();
}
}
\$\endgroup\$
  • \$\begingroup\$ You are creating too much threads! Not all ops(like field set) should be done async here. I could not find the use-case so clear, so you mean only one download at-a-time? And once it's finished go for next queued download? \$\endgroup\$ – user230399 Sep 12 at 20:23
  • \$\begingroup\$ many downloads at the same time but each process should wait for its relative download to finish \$\endgroup\$ – Davis Sep 13 at 3:46
  • \$\begingroup\$ And a request for file a could be asked while, it has been started already? So the process/callback should be queued? \$\endgroup\$ – user230399 Sep 13 at 5:42
3
\$\begingroup\$

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

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ This does not download async? Should I do sth like new Thread(()->{download(request); processingQueue.add(request);}).start(); wrapping them in new thread? \$\endgroup\$ – Davis Sep 14 at 15:08
  • \$\begingroup\$ This does not download on the main thread, which pretty much means it is asynchronous. The download runs in a separate thread. What exactly do you mean? \$\endgroup\$ – Bobby Sep 14 at 17:15
  • \$\begingroup\$ each download run async from each other. Downloads should block each other. But each process needs to wait relative download operation. \$\endgroup\$ – Davis Sep 14 at 18:45
  • \$\begingroup\$ They don't block each other as long as the thread pool is not exhausted. What your confusion most likely is, is that I"m talking about a thread pool, not a single instance of DownloadingThread, but multiple running in parallel. \$\endgroup\$ – Bobby Sep 14 at 19:36
  • \$\begingroup\$ Can I just use CompletionService instead of these? I think it can provide the same logic? \$\endgroup\$ – Davis Sep 15 at 17:09

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.