批次存储透明

Question

我们正在使用以下框架和版本：

• jOOQ 3.11.1
• Spring Boot 2.3.1.RELEASE
• Spring 5.2.7.RELEASE

在我的问题中，我们的一些业务逻辑被划分为逻辑单元，如下所示：

• 接收包含用户事务的请求。
• 此请求包含各种信息，如交易类型、该交易中的哪些产品、进行了何种付款等。
• 然后将这些属性单独存储在数据库中。

在代码中，这大致如下所示：

TransactionRecord transaction = transactionRepository.create();
transaction.create(creationCommand);`

在Transaction#create (通过事务方式运行)中，会发生类似的情况：

storeTransaction();
storePayments();
storeProducts();
// ... other relevant information

给定的事务可以有许多不同类型的产品和属性，所有这些都是存储的。这些属性中有许多会导致UPDATE语句，而有些属性可能会导致INSERT语句--很难事先完全知道。

例如，storeProducts方法看起来大致如下：

products.forEach(product -> {
    ProductRecord record = productRepository.findProductByX(...);
    if (record == null) {
        record = productRepository.create();
        record.setX(...);
        record.store();
    } else {
      // do something else
    }
});

如果产品是新的，他们是INSERT编辑。否则，可能会进行其他计算。根据事务的大小，这个单个用户事务显然会导致O(n)数据库调用/往返，甚至更多地取决于存在哪些其他属性。在存在大量属性的事务中，这可能导致对单个请求(!)的数百个数据库调用。我希望尽可能接近O(1)，以便在我们的数据库上有更多可预测的负载。

当然，这里会想到批处理和批量插入/更新。我想做的是使用jOOQ将所有这些语句批次到一个批处理中，并在提交之前在成功的方法调用之后执行。我已经找到了几个帖子(所以波斯特，jOOQ API，jOOQ GitHub特性请求)，其中隐式地提到了这个主题，而一个用户组发布似乎与我的问题有着明确的联系。

由于我将Spring与jOOQ结合使用，我相信理想的解决方案(最好是声明式解决方案)如下所示：

@Batched(100) // batch size as parameter, potentially
@Transactional
public void createTransaction(CreationCommand creationCommand) {
    // all inserts/updates above are added to a batch and executed on successful invocation
}

要使其工作，我想我需要管理一个作用域(ThreadLocal/Transactional/Session范围)资源，该资源可以跟踪当前批处理，以便：

1. 在输入该方法之前，如果该方法为@Batched，则创建一个空批，
2. 通过DI提供的自定义DSLContext (可能是扩展DefaultDSLContext)有一个ThreadLocal标志，它跟踪当前是否应该批处理任何语句，如果应该的话。
3. 拦截调用并将它们添加到当前批处理中，而不是立即执行它们。

然而，第3步将需要重写我们从(海事组织)相对可读的代码的很大一部分：

records.forEach(record -> {
    record.setX(...);
    // ...
    record.store();
}

至：

userObjects.forEach(userObject -> {
    dslContext.insertInto(...).values(userObject.getX(), ...).execute();
}

因为第二种形式也可以使用DSLContext#batchStore或DSLContext#batchInsert重写，这就违背了一开始就具有这种抽象的目的。然而，海事组织的批次和散装插入不应由个别开发人员决定，而应能够在更高层次(例如通过框架)透明地处理。

我发现jOOQ API的可读性是使用它的一个惊人的好处，然而，就我所能知道的情况而言，它似乎不能很好地用于这种情况下的拦截/扩展。使用jOOQ 3.11.1 (甚至是当前的) API，是否有可能获得与前者类似的透明批/批量处理行为？这会带来什么后果？

编辑：

一个可能的但非常麻烦的解决方案出现在脑海中，以支持透明的批次商店将如下所示：

1. 在启用批处理时，创建一个RecordListener并将其作为缺省值添加到Configuration中。
2. 在RecordListener#storeStart中，将查询添加到当前事务的批处理中(例如，在ThreadLocal<List>中)
3. AbstractRecord有一个changed标志，在存储之前检查它(org.jooq.impl.UpdatableRecordImpl#store0，org.jooq.impl.TableRecordImpl#addChangedValues)。重置此操作(并将其保存以供以后使用)使存储操作成为不操作。
4. 最后，在成功调用方法时但在提交之前：

• 将各记录的changes标志重置为正确的值
• 调用org.jooq.UpdatableRecord#store，这次不使用RecordListener或者跳过storeStart方法(可能使用另一个ThreadLocal标志来检查是否已经执行了批处理)。

据我所知，这种方法在理论上应该可行。显然，它非常麻烦，而且容易崩溃，因为如果代码依赖于反射才能工作，库内部可能随时会发生变化。

有没有人知道一个更好的方法，只使用公共jOOQ API？

Answer 1

jOOQ 3.14溶液

您已经发现了相关的特性请求#3419，它将在JDBC级别上解决这个问题，从jOOQ 3.14开始。您可以直接使用BatchedConnection，包装自己的连接以实现以下功能，也可以使用以下API：

ctx.batched(c -> {

    // Make sure all records are attached to c, not ctx, e.g. by fetching from c.dsl()
    records.forEach(record -> {
        record.setX(...);
        // ...
        record.store();
    }
});

jOOQ 3.13及解决方案前

目前，在#3419实现之前(在jOOQ 3.14中将是这样)，您可以自己实现它作为一个解决方案。你必须代理一个JDBC Connection和PreparedStatement并且..。

...拦截全部：

• 调用Connection.prepareStatement(String)，如果string与最后一个准备语句相同，则返回一个缓存的代理语句，或者批处理执行最后一个准备语句并创建一个新语句。
• 调用PreparedStatement.executeUpdate()和execute()，代之以对PreparedStatement.addBatch()的调用

...委托all:

• 调用其他API，例如Connection.createStatement()，它应该刷新上述缓冲批，然后调用委托API。

我不建议您绕过jOOQ的RecordListener和其他SPIs，我认为这是缓冲数据库交互的错误抽象级别。此外，您还需要对其他语句类型进行批处理。

请注意，在默认情况下，jOOQ的UpdatableRecord试图获取生成的标识值(请参阅Settings.returnIdentityOnUpdatableRecord)，这会阻止批处理。这样的store()调用必须立即执行，因为您可能期望标识值是可用的。