异步批量插入大文件异常求助：Large Objects不可用在自动提交模式

遇到的异常：

nested exception is org.postgresql.util.PSQLException: Large Objects may not be used in auto-commit mode.

代码示例 #

业务类代码

@Autowired
private JdbcTemplate jdbcTemplate;

@Autowired
private NamedParameterJdbcTemplate namedJdbcTemplate;

@Value("${jdbc.batch_insert_size}")
private int batchSize;

private static final ExecutorService executor = Executors.newFixedThreadPool(10);

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void batchInsertAsync(List<User> users) throws InterruptedException, ExecutionException {
  
  String sql = "INSERT INTO `USER` (USERNAME, PASSWORD, FILE, UPDATEDTIME, USERTYPE, DATEOFBIRTH)"
      + " VALUES(?,?,?,?,?,?)";
  final AtomicInteger sublists = new AtomicInteger();
  CompletableFuture[] futures = users.stream()
      .collect(Collectors.groupingBy(t -> sublists.getAndIncrement() / batchSize))
      .values()
      .stream()
      .map(ul -> runBatchInsert(ul, sql))
      .toArray(CompletableFuture[]::new);
  CompletableFuture<Void> run = CompletableFuture.allOf(futures);
  
  run.get();
  
  log.info("batchInsertAsync -> Total time in seconds: " + timer.getTotalTimeSeconds());
}
public CompletableFuture<Void> runBatchInsert(List<User> users, String sql) {
  return CompletableFuture.runAsync(() -> {
    jdbcTemplate.batchUpdate(sql, new UserBatchPreparedStatementSetter(users));
  }, executor);
}

批量参数设置类代码

public class UserBatchPreparedStatementSetter implements BatchPreparedStatementSetter{
  private List<User> users;
  
  public UserBatchPreparedStatementSetter(List<User> users) {
    super();
    this.users = users;
  }
  @Override
  public void setValues(PreparedStatement ps, int i) {
    
    try {
      User user = users.get(i);
      ps.setString(1, user.getUserName());
      ps.setString(2, user.getPassword());
      ps.setClob(3,org.hibernate.engine.jdbc.ClobProxy.generateProxy(user.getContent()));
      ps.setDate(4, user.getUpdatedTime() == null ? null : new java.sql.Date(user.getUpdatedTime().getTime()));
      ps.setString(5, user.getUserType().toString());
      ps.setDate(6, new java.sql.Date(user.getDateofBirth().getTime()));
    } catch (SQLException e) {
      e.printStackTrace();
    }
  }
  @Override
  public int getBatchSize() {
    return users.size();
  }
}

解决方案 #

• 修复事务与异步线程的绑定问题
  当前@Transactional注解加在batchInsertAsync方法上，但Spring事务绑定在当前线程，异步任务提交到自定义线程池后无法继承原事务上下文，导致JDBC操作处于自动提交模式，这是PostgreSQL大对象操作的禁忌场景。

  改用TransactionTemplate手动管理异步线程内的事务，确保每个批量任务都在独立事务中执行：

  @Autowired
  private TransactionTemplate transactionTemplate;
  
  public CompletableFuture<Void> runBatchInsert(List<User> users, String sql) {
    return CompletableFuture.runAsync(() -> {
      transactionTemplate.execute(status -> {
        jdbcTemplate.batchUpdate(sql, new UserBatchPreparedStatementSetter(users));
        return null;
      });
    }, executor);
  }
  

• 替换Hibernate ClobProxy为原生JDBC实现
  ClobProxy是Hibernate代理类，在纯JDBC模板场景下易引发事务兼容性问题，改用原生JDBC方式创建Clob：

  @Override
  public void setValues(PreparedStatement ps, int i) throws SQLException {
    User user = users.get(i);
    ps.setString(1, user.getUserName());
    ps.setString(2, user.getPassword());
    
    if (user.getContent() != null) {
      Clob clob = ps.getConnection().createClob();
      clob.setString(1, user.getContent());
      ps.setClob(3, clob);
    } else {
      ps.setNull(3, Types.CLOB);
    }
    
    ps.setDate(4, user.getUpdatedTime() == null ? null : new java.sql.Date(user.getUpdatedTime().getTime()));
    ps.setString(5, user.getUserType().toString());
    ps.setDate(6, new java.sql.Date(user.getDateofBirth().getTime()));
  }
  

• 改用Spring管理的线程池（可选）
  替换静态线程池为Spring托管的TaskExecutor，提升线程资源的可控性：

  @Autowired
  private TaskExecutor taskExecutor;
  
  public CompletableFuture<Void> runBatchInsert(List<User> users, String sql) {
    return CompletableFuture.runAsync(() -> {
      transactionTemplate.execute(status -> {
        jdbcTemplate.batchUpdate(sql, new UserBatchPreparedStatementSetter(users));
        return null;
      });
    }, taskExecutor);
  }
  

  配套配置类定义：

  @Bean
  public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("batch-insert-");
    executor.initialize();
    return executor;
  }
  

内容的提问来源于stack exchange，提问作者Sanket Gupta