Java用多线程持久化一个列表答案

【问题标题】：Java persist a list with multi threadJava用多线程持久化一个列表
【发布时间】：2019-02-08 17:22:10
【问题描述】：

我需要创建一个列表来执行以下操作：

我每微秒从外部队列/主题接收一个对象。
对对象进行一些操作后，我需要将这些对象持久化到数据库中。
我正在进行 100 或 1000 批次的持久化。唯一的问题是，持久化率低于传入消息率。现在我不想将其保留在单个线程中，因为持久化会减慢消息消耗。
我的想法是继续接受消息对象并将它们添加到集合中（如链表）
并继续以 100 或 1000 个为一组从集合的另一端删除并持久化到数据库中。
什么是正确的集合使用？如何同步这个，避免并发修改异常？

下面是我尝试使用 ArrayList 实现的代码，该 ArrayList 在持久化时每隔几秒清除一次列表。

class myclass{
List persistList;
ScheduledExecutorService persistExecutor;
ScheduledFuture scheduledFuture;
PersistOperation persistOperation;
//Initialize delay, interval
void init(){
scheduledFuture=persistExecutor.scheduleAtFixedRate(new persistOperation(persistList), delay, interval, TimeUnit.SECONDS);
}
void execute(msg){
//process the message and add to the persist list
}
class PersistOperation implements Runnable{
List persistList
PersistOperation(List persistList){
//Parameterized constructor
}
run(){
//Copy persistList to new ArrayList and clear persistList
//entity manager persist/update/merge
}
}
}

【问题讨论】：

标签： java database multithreading collections persist

【解决方案1】：

并继续以 100 或 1000 的批次从集合的另一端删除并持久化到数据库中。

只要有多个线程从集合中轮询，这是合理的。

下面是我尝试用 ArrayList 实现的代码

ArrayList 在这里是一个糟糕的选择，因为它不是线程安全的，并且在删除索引 0 处的元素时，必须将其右侧的每个元素都移过来（O(n) 操作） .

您要查找的集合称为Deque，也称为双端队列。但是，因为您需要集合是线程安全的，所以我建议使用 ConcurrentLinkedDeque。

【讨论】：

【解决方案2】：

我想你会想在这里使用LMAX Disruptor framework。我设想两个 RingBuffers。您将使用第一个来接受传入的消息。您的工人将从 RingBuffer 中读取数据。您可以将 RingBuffer 的大小设置为等于您的持久性块大小（例如 100 或 1000）。在工作人员从 RingBuffer 获取事件并对其进行处理后，它将对持久对象的引用放入队列集合中。每次第一个 RingBuffer 被圈出一次，你分配一个新的 Queue 并将旧的 Queue 放入第二个 RingBuffer。第二个 RingBuffer 的 worker(s) 从 RingBuffer 中获取一个 Queue 对象，持久化 Queue 中的所有对象，然后移动到下一个队列。您可以调整第二个 RingBuffer 和工作线程的大小，以适应数据库持久化数据块的速度。

【讨论】：

【解决方案3】：

如果您有 100 条消息已收到但未保存，并且您的应用程序死掉了，您能否承受丢失这些消息的损失？主题/队列的类型在这里很重要，主题具有管理这种背压控制的优势，队列通常在那里，因为需要有序处理。如果您的队列/主题是 kafka，并且您提取消息，则 kafka 可以提取批次，并且您可能也可以将批次保存到数据库中，只有在保存后才向 kafka 确认消息。如果您的处理需要订购，您可能可以处理一些反应性方法之王并调整数据库。通常，队列系统可以控制流量。

【讨论】：