Spring 集成 JDBC 批量插入答案

【问题标题】：Spring Integration JDBC Batch InsertSpring 集成 JDBC 批量插入
【发布时间】：2023-03-18 01:04:01
【问题描述】：

我需要插入要插入数据库的 POJO 列表。我有一个存储过程，一次执行一个插入。在当前的实现中，我有一个拆分器，它在 POJO 上拆分并将该有效负载传递到存储过程出站网关以调用我的存储过程。

在实时场景中，我的列表大小可能高达 500K。那么有没有更好的实施方法？有没有办法在 SI 流程中执行批量插入？

谢谢

【问题讨论】：

我不知道你的衡量标准是什么，但这里有一些建议：（1）不要一次插入一个；批处理请求以减少网络流量，(2) 将列表分成较小的部分，作为一个工作单元提交，这样您就不会创建巨大的回滚段，(3) 看看 Java 8 并行流是否让它运行得更快.
您使用什么风格的 SQL？如果您实时插入 500K，您可能必须获得相当低的级别。例如，MySQL 有一个 LOAD INTO 调用，它采用比标准批量插入快得多的输入流。
目前，Spring Integration 不支持此功能。有功能要求 - jira.spring.io/browse/INT-3364。因此，我为此目的使用了自己编写的出站通道适配器。

标签： java spring-integration

【解决方案1】：

我写了提到自定义出站通道适配器，它使用spring-jdbc批处理能力：

public class ArrayListSqlBatchOBCA {

  private final static Logger           log = LoggerFactory.getLogger(ArrayListSqlBatchOBCA.class);
  private NamedParameterJdbcTemplate    template;
  private String                        sql;

  public void process(Message<? extends ArrayList<?>> message) throws Exception {

    try {
          ArrayList<?> list = (ArrayList<?>) message.getPayload();

          SqlParameterSource[] batchArgs = new SqlParameterSource[list.size()];
          for (int i = 0; i < list.size(); i++) {
              batchArgs[i] = new BeanPropertySqlParameterSource(list.get(i));
          }

          template.batchUpdate(sql, batchArgs);
      }
      catch (Exception e) {
          log.error("Exception while processing message", e);
          throw e;
      }
   }
}

这就是我使用它的方式（XML 配置）：

  <channel id="cc"/>
  <outbound-channel-adapter channel="cc" method="process">
      <beans:bean class="mypackage.ArrayListSqlBatchOBCA">
          <beans:property name="template" ref="jdbcTemplate"/>
          <beans:property name="sql" 
             value="INSERT INTO test (field1,field2,field3)  
                    VALUES (:f1,:f2,:f3)"/>
  </beans:bean>
  </outbound-channel-adapter>

此外，我在数据源的 jdbc url 中添加了 ?reWriteBatchedInserts=true（我使用 postgres JDBC 驱动程序）。

Documentation about the option:

reWriteBatchedInserts - 启用优化以重写和折叠批处理的兼容 INSERT 语句。如果启用，pgjdbc 将一批insert into ... values(?, ?) 重写为insert into ... values(?, ?), (?, ?), ...

这样，使用一个小的自定义代码，我从输入集合中一次插入多行。

【讨论】：