Thread pool to process messages in parallel, but preserve order within conversations

user_1847482 0
Viewed 3078

I need to process messages in parallel, but preserve the processing order of messages with the same conversation ID.

Example:
Let's define a Message like this:

class Message {
    Message(long id, long conversationId, String someData) {...}
}

Suppose the messages arrive in the following order:
Message(1, 1, "a1"), Message(2, 2, "a2"), Message(3, 1, "b1"), Message(4, 2, "b2").

I need the message 3 to be processed after the message 1, since messages 1 and 3 have the same conversation ID (similarly, the message 4 should be processed after 2 by the same reason).
I don't care about the relative order between e.g. 1 and 2, since they have different conversation IDs.

I would like to reuse the java ThreadPoolExecutor's functionality as much as possible to avoid having to replace dead threads manually in my code etc.

Update: The number of possible 'conversation-ids' is not limited, and there is no time limit on a conversation. (I personally don't see it as a problem, since I can have a simple mapping from a conversationId to a worker number, e.g. conversationId % totalWorkers).

Update 2: There is one problem with a solution with multiple queues, where the queue number is determined by e.g. 'index = Objects.hash(conversationId) % total': if it takes a long time to process some message, all messages with the same 'index' but different 'conversationId' will wait even though other threads are available to handle it. That is, I believe solutions with a single smart blocking queue would be better, but it's just an opinion, I am open to any good solution.

Do you see an elegant solution for this problem?

multithreading java
8 Answers

This library should help: https://github.com/jano7/executor

ExecutorService underlyingExecutor = Executors.newCachedThreadPool();
KeySequentialRunner<String> runner = new KeySequentialRunner<>(underlyingExecutor);

Message message = retrieveMessage();

Runnable task = new Runnable() {
    @Override
    public void run() {
        // process the message
    }
};

runner.run(message.conversationId, task);
user_135244900
Related