java concurrency - keep order between certain tasks

user_1977315 0
Viewed 178

My app receives messages from several users. Messages per user must be executed in order, but otherwise messages can be executed in parallel. How to implement such logic?

E.g. messages arrive in this order: u1:m1 u1:m2 u2:m1 u1:m3 u2:m2. Execution should be parallel like this:

  • thread1: u1:m1 u1:m2 u1:m3
  • thread2: u2:m1 u2:m2

Number of users can be huge, therefore I can't just create a single threaded executor per user.

private ExcutorService executorService = newFixedThreadPool(10);

public void onMessage(String user, String message) {
  // TODO schedule tasks per user in order
  executorService.schedule(() -> processMessage(message));
}
concurrency java
2 Answers

You can do it like this:

        //CREATE EXECUTORS
        int numberOfThreads = 10;
        ExecutorService[] executors = new ExecutorService[numberOfThreads];
        for (int i = 0; i < numberOfThreads; i++) {
            executors[i] = Executors.newSingleThreadExecutor();
        }

Then in your method use a specific executor for the user with the help of hashCode and modulo:

public void onMessage(String user, String message) {
    //same user will always get the same executor, hashCode will evenly distribute the load among the executors
    int executorToUse = Math.abs(user.hashCode()) % numberOfThreads; 
    ExecutorService executorService = executors[executorToUse];
    executorService.execute(() -> processMessage(message));
}
user_7749700

You can decouple message producer and processor by kind of grouping queue:

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class GroupingBlockingQueue<G, E> {

    private static final int UNBOUNDED = -1;

    private final Lock lock = new ReentrantLock();

    private final Condition notFull  = lock.newCondition();

    private final Condition notEmpty = lock.newCondition();

    private final Map<G, List<E>> map = new LinkedHashMap<>();

    private final int bound;

    public GroupingBlockingQueue() {
        this(UNBOUNDED);
    }

    public GroupingBlockingQueue(int bound) {
        this.bound = bound;
    }

    public void put(G group, E element) throws InterruptedException {
        lock.lock();
        try {
            if (bound > 0) {
                while (bound == map.keySet().size()) {
                    notFull.await();
                }
            }
            map.computeIfAbsent(group, k -> new ArrayList<>()).add(element);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Item<G, E> take() throws InterruptedException {
        lock.lock();
        try {
            Set<G> keys = map.keySet();
            while (keys.isEmpty()) {
                notEmpty.await();
            }
            G group = keys.iterator().next();
            List<E> elements = map.remove(group);
            notFull.signal();
            return new Item<>(group, elements);
        } finally {
            lock.unlock();
        }
    }

    public static class Item<G, E> {
        private final G group;
        private final List<E> elements;

        public Item(G group, List<E> elements) {
            this.group = group;
            this.elements = elements;
        }

        public G getGroup() {
            return group;
        }

        public List<E> getElements() {
            return elements;
        }
    }
}

Then message processor may look like:

package ru.dkovalev;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import static java.util.concurrent.Executors.newFixedThreadPool;

public class UserMessageProcessor implements AutoCloseable {

    private GroupingBlockingQueue<String, String> queue = new GroupingBlockingQueue<>();
    private ExecutorService executorService;

    public UserMessageProcessor(int threadCount) {
        executorService = newFixedThreadPool(threadCount);
        for (int i = 0; i < threadCount; i++) {
            executorService.submit((Runnable) this::processMessages);
        }
    }

    private void processMessages() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                GroupingBlockingQueue.Item<String, String> item = queue.take();
                processMessages(item.getGroup(), item.getElements());
            }
        } catch (InterruptedException ignore) {}
    }

    private void processMessages(String user, List<String> messages) throws InterruptedException {
        for (String message : messages) {
            System.out.println(String.format("[%s] %s:%s", Thread.currentThread().getName(), user, message));
            Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        }
    }

    public void onMessage(String user, String message) throws InterruptedException {
        queue.put(user, message);
    }

    @Override
    public void close() {
        try {
            executorService.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException ignore) {
        }
        executorService.shutdown();
    }

    public static void main(String[] args) throws InterruptedException {
        try (UserMessageProcessor processor = new UserMessageProcessor(10)) {
            processor.onMessage("u1", "m1");
            processor.onMessage("u2", "m1");
            processor.onMessage("u1", "m2");
            processor.onMessage("u2", "m2");
            processor.onMessage("u1", "m3");
        }
    }
}
user_13720390
Related