我有一个系统，它一次使用大约 30 个线程将大任务分解为小任务。当每个单独的线程完成时，它会将其计算结果保存到数据库中。我想要实现的是让每个线程将其结果传递给一个新的持久性类，该类将在其自己的线程中运行时执行一种双缓冲和数据持久性。
例如，在 100 个线程将其数据移动到持久性类的缓冲区后，持久性类会交换缓冲区并将所有 100 个条目持久化到数据库中。这将允许使用准备好的语句，从而减少程序和数据库之间的 I/O。
有这种类型的多线程双缓冲的模式或很好的例子吗？
【问题讨论】：
标签： java multithreading design-patterns
我见过这种模式被称为异步数据库写入或后写模式。这是分布式缓存产品（Teracotta、Coherence、GigaSpaces 等）支持的典型模式，因为您不希望缓存更新还包括将更改写入底层数据库。
此模式的复杂性取决于您对丢失数据库更新的容忍度。由于完成工作和将结果写入数据库之间的延迟，您可能会因错误、电源故障等而丢失更新...（您看懂了）。
我建议使用某种队列将完成的结果写入数据库，然后分批处理 100 个（使用您的示例）或一段时间后处理它们。还使用时间延迟的原因是为了处理不能被 100 整除的结果集。
如果您对弹性/耐用性没有要求，那么您可以在同一个过程中完成所有这些。但是，如果您不能容忍任何损失，那么您可以将 in-vm 队列替换为持久 JMS 队列（更慢但更安全）。
【讨论】：
为了降低同步开销，请使用本地线程（针对每个计算线程）来构建批量结果。一旦达到一定数量的结果，将批处理排入阻塞队列。使用 ArrayBlockingQueue 来支持您的持久性类，因为您可能不希望您的内存使用变得无限。您可以让多个数据库编写器线程获取结果组并将它们保存到数据库中。
class WriteBehindPersister {
ThreadLocal<List<Result>> internalBuffer;
static ArrayBlockingQueue<List<Result>> persistQueue;
static {
persistQueue = new ArrayBlockingQueue(10);
new WriteThread().start();
}
public WriteBehindPersister() {
internalBuffer = new ThreadLocal<List<Result>>();
}
public void persist(Result r) {
List<Result> localResult = internalBuffer.get();
localResult.add(r);
if (localResult.size() > max) {
persistQueue.put(new ArrayList(localResult));
localResult.clear();
}
}
class WriteThread extends Thread {
public void run() {
while (true) {
List<Result> batch = persistQueue.take();
beginTransaction();
for (Result r : batch) {
batchInsert(r);
}
endTransaction();
}
}
}
}
此外，您可以使用执行器服务（而不是单个写入线程）将多个批次同时保存到数据库，但要权衡使用多个数据库连接。如果您的驱动程序支持，请确保使用 JDBC 批处理 API。
【讨论】：