本文講述一個由 ShardingJDBC 使用不當(dāng)引起的悲慘故事。

一. 問題重現(xiàn)

有一天運營反饋我們部分訂單狀態(tài)和第三方訂單狀態(tài)無法同步。

根據(jù)現(xiàn)象找到了不能同步訂單狀態(tài)是因為 order 表的 thirdOrderId 為空導(dǎo)致的，但是這個字段為啥為空，排查過程比較波折。

過濾掉復(fù)雜的業(yè)務(wù)邏輯，當(dāng)時的代碼可以簡化為這樣：

Order order;
// 業(yè)務(wù)在特定情況會生成新的訂單
if (特定條件) {
    order = buildOrders();
	orderService.saveBatch(Lists.newArrayList(order));
}


// 省略復(fù)雜的業(yè)務(wù)邏輯
// ...

// 調(diào)用第三方下單
ThirdOrder thirdOrder = callThirdPlaceOrder()
// 設(shè)置order表 thirdOrderId 字段
order.setThirdOrderId(thirdOrder.getOrderId());
// 設(shè)置 order_item 表 thirdOrderId 字段
orderItems.foreach(e -> e.setThirdOrderId(thirdOrder.getOrderId()));

// 更新 order 表
orderService.updateById(order);
// 更新 order_item 表
orderItemService.updateBatchById(itemUpdateList);

我們發(fā)現(xiàn)這類有問題的訂單，order 表 thirdOrderId 為空，但是 order_item 表 thirdOrderId 更新成功了，使我們直接排除了這里母單更新“失敗”的問題，因為兩張表的更新操作在一個事務(wù)里面，子單更新成功了說明這里的代碼邏輯應(yīng)該沒有問題。

就是這里的錯覺，讓我們走了很多彎路。我們排查了所有可能存在并發(fā)更新、先讀后寫、數(shù)據(jù)覆蓋的地方，結(jié)合業(yè)務(wù)日志，翻遍了業(yè)務(wù)代碼仍然無法確認(rèn)問題具體在哪里。最后只能在可能出現(xiàn)問題的地方補充了日志，同時我們也在此處更新 order 表的地方加上了日志，最后發(fā)現(xiàn)在執(zhí)行 orderService.saveBatch 后 order 的 id 為空，導(dǎo)致 order 的更新并沒有成功。

說實話找到問題的那一刻有點顛覆我的認(rèn)知，在我的印象中，MyBatisPlus批量插入的方法是可以返回ID，經(jīng)過實驗，在當(dāng)前項目環(huán)境中，save()方法會返回主鍵ID，但是saveBatch()方法不會。這種顛覆認(rèn)知的新

二. 源碼分析

2.1 JDBC如何獲取批量插入數(shù)據(jù)的ID

要想摸清楚批量插入后為什么沒有獲取到主鍵ID，我們得先了解一下JDBC如何批量插入數(shù)據(jù)，以及在批量插入操作后，獲取數(shù)據(jù)庫的主鍵值。

// 創(chuàng)建一個 PreparedStatement 對象，并指定獲取自動生成的主鍵
PreparedStatement pstmt = conn.prepareStatement("INSERT INTO order_info (column1, column2) VALUES (?, ?)", Statement.RETURN_GENERATED_KEYS);

// 執(zhí)行批量插入操作
pstmt.setString(1, "value1"); // 設(shè)置參數(shù)值
pstmt.setString(2, "value2"); // 設(shè)置參數(shù)值
pstmt.addBatch(); // 添加批量操作

// ... 添加更多批量操作

// 執(zhí)行批量操作
pstmt.executeBatch();

// 獲取生成的主鍵
ResultSet generatedKeys = pstmt.getGeneratedKeys();
while (generatedKeys.next()) {
    int primaryKey = generatedKeys.getInt(1); // 假設(shè)主鍵為整數(shù)類型，如果是其他類型，請根據(jù)實際情況調(diào)整
    System.out.println("Generated Primary Key: " + primaryKey);
}

// 關(guān)閉相關(guān)資源
generatedKeys.close();
pstmt.close();
conn.close();

在執(zhí)行批量插入操作后，我們可以通過 Statement.getGeneratedKeys() 方法獲取數(shù)據(jù)庫主鍵值。

2.2 MyBatis 批量插入原理

MyBatis-Plus 是對 MyBatis 的一種增強，底層還是依賴于MyBatis SqlSession API對數(shù)據(jù)庫進(jìn)行的操作，而SqlSession執(zhí)行批量插入大概分為如下幾步：

try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
    YourMapper mapper = sqlSession.getMapper(YourMapper.class);

    List<YourEntity> entities = new ArrayList<>();
    // 添加要插入的實體對象到列表中

    for (YourEntity entity : entities) {
        // 調(diào)用插入方法，但此時還未真正執(zhí)行
        mapper.insert(entity); 
    }

    // 批量執(zhí)行SQL
    sqlSession.flushStatements(); 

    sqlSession.commit(); // 提交事務(wù)
} catch (Exception e) {
    sqlSession.rollback(); // 發(fā)生異常時回滾事務(wù)
}

2.3 Myabtis-Plus + ShardingJDBC 批量插入數(shù)據(jù)為什么無法獲取ID

MyBatis-Plus 執(zhí)行批量插入操作本質(zhì)上和MyBatis是一致的，Myabtis-Plus saveBtach方法：

/**
 * 插入（批量）
 *
 * @param entityList 實體對象集合
 */
@Transactional(rollbackFor = Exception.class)
default boolean saveBatch(Collection<T> entityList) {
    return saveBatch(entityList, DEFAULT_BATCH_SIZE);
}

/**
 * 批量插入
 *
 * @param entityList ignore
 * @param batchSize  ignore
 * @return ignore
 */
@Transactional(rollbackFor = Exception.class)
@Override
public boolean saveBatch(Collection<T> entityList, int batchSize) {
    String sqlStatement = getSqlStatement(SqlMethod.INSERT_ONE);
    return executeBatch(entityList, batchSize, (sqlSession, entity) -> sqlSession.insert(sqlStatement, entity));
}

進(jìn)入executeBatch：

   /**
     * 執(zhí)行批量操作
     *
     * @param entityClass 實體類
     * @param log         日志對象
     * @param list        數(shù)據(jù)集合
     * @param batchSize   批次大小
     * @param consumer    consumer
     * @param <E>         T
     * @return 操作結(jié)果
     * @since 3.4.0
     */
    public static <E> boolean executeBatch(Class<?> entityClass, Log log, Collection<E> list, int batchSize, BiConsumer<SqlSession, E> consumer) {
        Assert.isFalse(batchSize < 1, "batchSize must not be less than one");
        return !CollectionUtils.isEmpty(list) && executeBatch(entityClass, log, sqlSession -> {
            int size = list.size();
            int i = 1;
            for (E element : list) {
                // 執(zhí)行 sqlSession 的 insert 方法
                consumer.accept(sqlSession, element);
                if ((i % batchSize == 0) || i == size) {
                    // 每達(dá)到 batchSize 就執(zhí)行SQL
                    sqlSession.flushStatements();
                }
                i++;
            }
        });
    }

在 executeBatch 中 MyBatis-Plus 會循環(huán)調(diào)用 SqlSession.insert 緩存插入語句，每 batchSize 提交一次SQL。

進(jìn)入 DefaultSqlSession.flushStatements():

  @Override
  public List<BatchResult> flushStatements() {
    try {
      return executor.flushStatements();
    } catch (Exception e) {
      throw ExceptionFactory.wrapException("Error flushing statements.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
  }

委托 BaseExecutor.flushStatements() 執(zhí)行：

  @Override
  public List<BatchResult> flushStatements() throws SQLException {
    return flushStatements(false);
  }

  public List<BatchResult> flushStatements(boolean isRollBack) throws SQLException {
    if (closed) {
      throw new ExecutorException("Executor was closed.");
    }
    return doFlushStatements(isRollBack);
  }

最終 doFlushStatements() 方法由各個子類去實現(xiàn)，BaseExecutor 有 BatchExecutor，ReuseExecutor，SimpleExecutor，ClosedExecutor，MybatisBatchExecutor，MybatisReuseExecutor，MybatisSimpleExecutor這幾種實現(xiàn)。

Mybatis 開頭的是 Mybatis-Plus 提供的實現(xiàn)，分別對應(yīng) MyBatis 的 simple、reuse、batch執(zhí)行器類別。不管哪個執(zhí)行器，里面都會有一個 StatementHandler 接口來負(fù)責(zé)具體執(zhí)行SQL。

而在 MyBatis-Plus 批量插入的場景中，是由 MybatisBatchExecutor#doFlushStatements 執(zhí)行的：

@Override
public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {
    try {
        List<BatchResult> results = new ArrayList<>();
        if (isRollback) {
            return Collections.emptyList();
        }
        for (int i = 0, n = statementList.size(); i < n; i++) {
            Statement stmt = statementList.get(i);
            applyTransactionTimeout(stmt);
            BatchResult batchResult = batchResultList.get(i);
            try {
                // 1. 此處調(diào)用JDBC PreparedStatement API，批量執(zhí)行SQL
                batchResult.setUpdateCounts(stmt.executeBatch());
                MappedStatement ms = batchResult.getMappedStatement();
                List<Object> parameterObjects = batchResult.getParameterObjects();
                KeyGenerator keyGenerator = ms.getKeyGenerator();
                if (Jdbc3KeyGenerator.class.equals(keyGenerator.getClass())) {
                    Jdbc3KeyGenerator jdbc3KeyGenerator = (Jdbc3KeyGenerator) keyGenerator;
                    // 2. 使用 jdbc3KeyGenerator 獲取批量執(zhí)行的所生成的 ID
                    jdbc3KeyGenerator.processBatch(ms, stmt, parameterObjects);
                } else if (!NoKeyGenerator.class.equals(keyGenerator.getClass())) { //issue #141
                    for (Object parameter : parameterObjects) {
                        keyGenerator.processAfter(this, ms, stmt, parameter);
                    }
                }
                // Close statement to close cursor #1109
                closeStatement(stmt);
            } catch (BatchUpdateException e) {
                StringBuilder message = new StringBuilder();
                message.append(batchResult.getMappedStatement().getId())
                    .append(" (batch index #")
                    .append(i + 1)
                    .append(")")
                    .append(" failed.");
                if (i > 0) {
                    message.append(" ")
                        .append(i)
                        .append(" prior sub executor(s) completed successfully, but will be rolled back.");
                }
                throw new BatchExecutorException(message.toString(), e, results, batchResult);
            }
            results.add(batchResult);
        }
        return results;
    } finally {
        for (Statement stmt : statementList) {
            closeStatement(stmt);
        }
        currentSql = null;
        statementList.clear();
        batchResultList.clear();
    }
}

在 1 處，執(zhí)行批量插入語句后，然后在2處調(diào)用 Jdbc3KeyGenerator.jdbc3KeyGenerator 獲取ID：

// org.apache.ibatis.executor.keygen.Jdbc3KeyGenerator#processBatch
public void processBatch(MappedStatement ms, Statement stmt, Object parameter) {
    final String[] keyProperties = ms.getKeyProperties();
    if (keyProperties == null || keyProperties.length == 0) {
      return;
    }
    // 本質(zhì)上，還是調(diào)用JDBC Statement.getGeneratedKeys() 方法獲取ID（參考文中2.1示例）
    try (ResultSet rs = stmt.getGeneratedKeys()) {
      final ResultSetMetaData rsmd = rs.getMetaData();
      final Configuration configuration = ms.getConfiguration();
      if (rsmd.getColumnCount() < keyProperties.length) {
        // Error?
      } else {
        assignKeys(configuration, rs, rsmd, keyProperties, parameter);
      }
    } catch (Exception e) {
      throw new ExecutorException("Error getting generated key or setting result to parameter object. Cause: " + e, e);
    }
}

但是我們項目中如果使用的 ShardingJDBC，那么此時調(diào)用的就是 ShardingPreparedStatement.getGeneratedKeys():

通過 DEBUG，我們發(fā)現(xiàn)在我們項目中 ShardingPreparedStatement.getGeneratedKeys() 返回的是null值:

這也就找到了為什么MyBatis-Plus 和 ShardingJDBC 一起使用時獲取不到ID值的問題，問題的根節(jié)并不在MyBatis這邊，而是 ShardingJDBC 實現(xiàn)的 PreparedStatement 獲取不到key。

2.4 為什么執(zhí)行MyBatis-Plus save方法可以獲取到主鍵

當(dāng)我們調(diào)用 MyBatis-Plus save() 方法保存單條數(shù)據(jù)時，底層實際上還是調(diào)用的 ShardingPreparedStatement.getGeneratedKeys() 方法，獲取插入后的主鍵key:

@Override
public ResultSet getGeneratedKeys() throws SQLException {
    Optional<GeneratedKeyContext> generatedKey = findGeneratedKey();
    if (preparedStatementExecutor.isReturnGeneratedKeys() && generatedKey.isPresent()) {
        return new GeneratedKeysResultSet(generatedKey.get().getColumnName(), generatedValues.iterator(), this);
    }
    if (1 == preparedStatementExecutor.getStatements().size()) {
        return preparedStatementExecutor.getStatements().iterator().next().getGeneratedKeys();
    }
    return new GeneratedKeysResultSet();
}

但是在執(zhí)行單條數(shù)據(jù)插入時，1 == preparedStatementExecutor.getStatements().size() 是成立的，就會返回底層被真實被代理的MySQL JDBC 的 Statement 獲取主鍵key：

至于 AbstractStatementExecutor.statements 為什么在執(zhí)行單一語句的時候statements里不為空，但是批量插入的時候，這個list為空，可以參考下面的回答：

AbstractStatementExecutor.statements是ShardingJDBC中的一個重要數(shù)據(jù)結(jié)構(gòu)，它用于存儲待執(zhí)行的SQL語句及其對應(yīng)的數(shù)據(jù)庫連接信息。在進(jìn)行SQL操作時，ShardingJDBC會根據(jù)你的分片策略將SQL語句路由到相應(yīng)的數(shù)據(jù)庫節(jié)點，并生成對應(yīng)的數(shù)據(jù)結(jié)構(gòu)存儲在statements這個列表里。

那么，為什么在執(zhí)行單一SQL語句時，statements不為空，而在批量插入時，這個列表卻為空呢？這主要是因為ShardingJDBC處理這兩種情況的方式有所不同。

1. 對于單一SQL語句，ShardingJDBC將其路由到正確的數(shù)據(jù)庫節(jié)點（可能是多個），然后創(chuàng)建對應(yīng)的PreparedStatement對象，這些對象被存儲在statements列表中，以便后續(xù)執(zhí)行和獲取結(jié)果。
2. 對于批量插入，ShardingJDBC采取了一種“延遲執(zhí)行”的策略。具體來說，ShardingJDBC首先會解析和拆分批量插入語句，然后將拆分后的單一插入語句暫存起來，而不是立即創(chuàng)建PreparedStatement對象。這就導(dǎo)致了在批量插入過程中，statements列表為空。這樣做的主要目的是為了提高批量插入的性能，因為創(chuàng)建PreparedStatement對象和管理數(shù)據(jù)庫連接都是需要消耗資源的。

三. 總結(jié)

本文由故障現(xiàn)象定位到了具體的問題點是因為 MyBatis-Plus 批量插入沒有返回數(shù)據(jù)庫組件，而跟蹤源碼后我們卻發(fā)現(xiàn)是因為ShardingJDBC不支持批量插入獲取主鍵值。

ShardingJDBC不支持批量插入后獲取主鍵，主要是因為在批量插入操作中，ShardingJDBC可能需要將數(shù)據(jù)插入到多個不同的數(shù)據(jù)庫節(jié)點，在這種情況下，每個節(jié)點都可能有自己的主鍵生成規(guī)則，并且這些節(jié)點可能并不知道其他節(jié)點的主鍵值。因此，如果你需要在批量插入后獲取自動生成的主鍵，可能需要通過其他方式實現(xiàn)，例如使用全局唯一ID作為主鍵。