Jooq 中的原子批量插入/更新

Question

我有一个程序可以监视服务并保存其当前状态。我的程序会定期检查每个服务并将有关它们的信息存储在 postgres 中。大多数时候，这涉及更新数据库中的现有行。但有时，也需要添加新服务。

我用一个看起来像这样的模式来存储它们：

CREATE TABLE services IF NOT EXISTS (
  id          SERIAL    NOT NULL,
  name        TEXT      NOT NULL,
  status      TEXT    ,
  PRIMARY KEY (id, name)
)

其中 ID 指定一台机器，名称指定该机器上的一项服务。 POJO 非常简单：

public class Service {
  int id;
  String name;
  String status;
}

例如，我的表格可能只有一行：

(22, "api", "active")

。在预定的时间间隔，我的程序确定一台机器上现在有 2 个服务正在运行，并且当前服务的状态已更改：

• （22，“API”，“错误”）
• （22，“消息传递”，“活动”）

我的方法如下：

  Set<ServicesRecord> records = listServices
    .stream()
    .map(service -> {
      ServicesRecord record = new ServicesRecord();
      record.setId(service.id);
      record.setName(service.name);
      record.setStatus(service.status);
      return record;
    })
    .collect(Collectors.toSet());
    DSLContext dsl = DSL.using(this.configuration);
    dsl.batchStore(records).execute();
}

但是，当我尝试运行它时，这给了我一个错误：

Caused by: org.jooq.exception.DataAccessException: SQL [insert into "foobar"."services" ("id", "name", "status", values (?, ?, ?)]; Batch entry 0 insert into "foobar"."services" ("id", "name", "status") values (1, 'testName', 'baz') was aborted: ERROR: duplicate key value violates unique constraint "services_pkey"

我可以看到

batchStore

失败了，因为它使用了

UpdatableRecord

的

store

方法，而该方法又失败了，因为我正在创建一条新记录，而不是从 jooq 获取一条记录。

我正在考虑这样的替代方法：

1. 从数据库中获取所有记录。
2. 对于每条匹配

   records

   的记录，我将

   changed

   设置为

   true

   。
3. 然后我就跑

   batchStore

或者，我可以这样做吗？

1. 从数据库中获取所有记录。
2. 制作要插入的记录列表并使用

   batchInsert

3. 制作要更新的记录列表，修改其记录，并使用

   batchStore

但这两者都存在非原子性的风险。如果在执行步骤 2 时数据库发生更改，我对

batchStore

的使用仍然可能会失败。我想要的是一种在单个操作中执行这种批量存储的方法，而不是先获取然后执行。有办法做到这一点吗？

Answer 1

您正在寻找的是

batchMerge()

，它将随 jOOQ 3.14 一起发布：https://github.com/jOOQ/jOOQ/issues/10046

或者，您可以使用数据导入 API。然后，您可以将

onDuplicateKeyUpdate()

子句与适当的 批量大小结合起来。

或者，您可以编写显式的

INSERT .. ON CONFLICT

语句 并显式地对它们进行批处理。

Answer 2

JOOQ 3.17 的

batchMerge()示例

List<UpdatableRecord<ServiceRecord>> list = new ArrayList<>();
UpdatableRecord<ServiceRecord> record1 = new UpdatableRecordImpl<>(SERVICE);
record1.setValue(SERVICE.ID, 1);
record1.setValue(SERVICE.NAME, "1");
record1.setValue(SERVICE.STATUS, "1");
list.add(record1);
...
list.add(recordN);
dsl.batchMerge(list).execute();