当某些交易的顺序重要时,我如何多线程化队列消费者？

Question

我有一个对一组对象进行操作的任务队列(例如,为了示例,这些对象是地址簿中的条目).

示例任务可能是"将Joe的电话号码更新为888-555-1212".

同时在队列中有多个"更新Joe的电话号码......"任务,但电话号码不同.在这种情况下,必须应用更新以确保最后状态是正确的(并且,为了参数,不能将时间戳放在通讯簿条目上的任务和时间戳上并扔掉陈旧的任务).

使用Joe的更新对Jane进行无序更新是安全的.

我想多线程处理队列,但我需要按人员同步访问.

这种东西有一个方便的库吗？或者我降级为使用Executor并在Runnable的run()方法中对"name"进行自己的同步？

Answer 1

一种可能的解决方案

假设某个任务由某个类描述

class Task {
  Integer taskGroup;
  // other
}

其中taskGroup是一个ID，它标识必须按到达顺序处理的任务（在您的示例中，每个“名称”可以定义自己的taskGroup - 或者更一般地说 - 具有相同名称的任务属于同一taskGroup）。

让mainTaskQueue表示ListTask 对象。然后

• 创建一个Map<Integer,List<Task>>，说taskGroupsQueues
• 为每个任务组创建一个按顺序操作的线程taskGroupsQueues.get(taskGroup)。
• 主线程task从主任务列表中删除 amainTaskQueue并将其附加到taskGroups.get(task.taskGroup)
• 将任务从主队列移动到单个队列以及从单个队列中获取任务必须同步。

换句话说：属于相同名称的任务在同一线程上执行。

注意，如果主线程执行任务的分配，那么他还可以执行某种负载平衡，即，如果一个任务由于顺序一致性而没有被强制到特定队列，则该任务应该进入shortes队列。然而，您的问题固有的是它可能会变成单线程 - 即当您只有属于同一任务组（以您的案例名称）的任务时。

另一个可能的解决方案 （未经测试，只是一个建议）

正如increment1s 帖子和Aurands 评论中所指出的，tread 内taskGroup（名称）的同步存在一些问题。基本上：为时已晚，因为执行器可能已经启动了两个尝试同步同名的线程。但是，您可以尝试确保执行器级别的执行顺序。例如，请参阅这篇文章：Java Executors：如何设置任务优先级？（它引用传递给执行器的 PriorityBlockingQueue）。