问题

private ExecutorService executorService = Executors.newFixedThreadPool(10);
private ConcurrentHashMap<String, BlockingQueue<String>> userMessageQueues = new ConcurrentHashMap<>();

public void onMessage(String user, String message) {
    userMessageQueues.computeIfAbsent(user, key -> new LinkedBlockingQueue<>()).offer(message);
    
    if (!executorService.isShutdown()) {
        executorService.execute(() -> processMessagesForUser(user));
    }
}

private void processMessagesForUser(String user) {
    BlockingQueue<String> messageQueue = userMessageQueues.get(user);
    
    while (!messageQueue.isEmpty()) {
        String message = messageQueue.poll();
        if (message != null) {
            processMessage(message);
        }
    }
}

public void shutdown() {
    executorService.shutdown();
}

private void processMessage(String message) {
    // Process the message logic here
}

请注意，上述代码是基于您的要求编写的，用于按顺序执行每个用户的消息，同时允许并行处理不同用户的消息。代码创建一个线程池 executorService 来处理消息。每个用户都有一个关联的消息队列，其中消息会被顺序地添加。在 onMessage 方法中，消息被添加到相应用户的消息队列中，然后检查线程池是否正在关闭，如果没有，则会提交一个任务来处理该用户的消息。processMessagesForUser 方法从消息队列中取出消息并处理，确保按照顺序执行。最后，shutdown 方法用于关闭线程池。

请注意，代码可能需要根据您的实际应用进行调整和优化。

英文:

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(() -&gt; processMessage(message));
}

</details>


# 答案1
**得分**: 3

你可以这样做：

            // 创建执行器
            int 线程数量 = 10;
            ExecutorService[] 执行器们 = new ExecutorService[线程数量];
            for (int i = 0; i < 线程数量; i++) {
                执行器们[i] = Executors.newSingleThreadExecutor();
            }

然后在你的方法中，使用特定的执行器来处理用户的消息，利用哈希码和取模的方式：

    public void onMessage(String 用户, String 消息) {
        // 同一个用户将始终使用相同的执行器，哈希码会将负载均匀地分布在执行器之间
        int 要使用的执行器 = Math.abs(用户.hashCode()) % 线程数量;
        ExecutorService 执行器服务 = 执行器们[要使用的执行器];
        执行器服务.execute(() -> 处理消息(消息));
    }

<details>
<summary>英文:</summary>

You can do it like this:

            //CREATE EXECUTORS
            int numberOfThreads = 10;
            ExecutorService[] executors = new ExecutorService[numberOfThreads];
            for (int i = 0; i &lt; 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(() -&gt; processMessage(message));
    }



</details>



# 答案2
**得分**: 0

以下是翻译好的内容：

```java
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;
        }
    }
}

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");
        }
    }
}

英文:

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&lt;G, E&gt; {
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&lt;G, List&lt;E&gt;&gt; map = new LinkedHashMap&lt;&gt;();
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 &gt; 0) {
while (bound == map.keySet().size()) {
notFull.await();
}
}
map.computeIfAbsent(group, k -&gt; new ArrayList&lt;&gt;()).add(element);
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Item&lt;G, E&gt; take() throws InterruptedException {
lock.lock();
try {
Set&lt;G&gt; keys = map.keySet();
while (keys.isEmpty()) {
notEmpty.await();
}
G group = keys.iterator().next();
List&lt;E&gt; elements = map.remove(group);
notFull.signal();
return new Item&lt;&gt;(group, elements);
} finally {
lock.unlock();
}
}
public static class Item&lt;G, E&gt; {
private final G group;
private final List&lt;E&gt; elements;
public Item(G group, List&lt;E&gt; elements) {
this.group = group;
this.elements = elements;
}
public G getGroup() {
return group;
}
public List&lt;E&gt; 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&lt;String, String&gt; queue = new GroupingBlockingQueue&lt;&gt;();
private ExecutorService executorService;
public UserMessageProcessor(int threadCount) {
executorService = newFixedThreadPool(threadCount);
for (int i = 0; i &lt; threadCount; i++) {
executorService.submit((Runnable) this::processMessages);
}
}
private void processMessages() {
try {
while (!Thread.currentThread().isInterrupted()) {
GroupingBlockingQueue.Item&lt;String, String&gt; item = queue.take();
processMessages(item.getGroup(), item.getElements());
}
} catch (InterruptedException ignore) {}
}
private void processMessages(String user, List&lt;String&gt; messages) throws InterruptedException {
for (String message : messages) {
System.out.println(String.format(&quot;[%s] %s:%s&quot;, 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(&quot;u1&quot;, &quot;m1&quot;);
processor.onMessage(&quot;u2&quot;, &quot;m1&quot;);
processor.onMessage(&quot;u1&quot;, &quot;m2&quot;);
processor.onMessage(&quot;u2&quot;, &quot;m2&quot;);
processor.onMessage(&quot;u1&quot;, &quot;m3&quot;);
}
}
}

通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。

java并发 – 在特定任务之间保持顺序

问题

创建一个对象时使用超类引用和使用子类引用有什么区别？

无法在启动Spring应用程序期间注入MapStruct接口。

如何在Spring 5的Xml中配置WebContentInterceptor中的cacheControlMappings

302重定向加载空白页面。

What's the correct way to type hint an empty list as a literal in python?

如何在Highcharts Gantt中更改本地化的星期名称

如何在同一个流中使用多个过滤器和映射函数？

如何使用Map/Set来将代码优化到O(n)？

.NET MAUI Android在GitHub Actions上构建失败，错误代码为1。

如何在Playwright视觉比较中屏蔽多个定位器？

在C++中，可以使用可变模板参数来检索类型的内部类型。

selenium.common.exceptions.StaleElementReferenceException: Message: stale element reference: stale element not found

Creating and opening a URL to log in to Website via Basic Auth with Robot Framework/Selenium (Python)

AG Grid 在上下文菜单中以大文本形式打开

发表评论