java - Mybatis批处理

Question

我有一个包含 3 多百万条记录的表。我需要从数据库读取所有这些记录，将它们发送到 kafka 队列进行处理，以供其他系统处理。然后从输出kafka队列中读取结果并写回数据库。
我需要以合理的部分进行读写，否则我会立即收到 OOM 异常。

用mybatis实现批量读写操作有哪些可行的技术方案？
整洁的工作示例将不胜感激。

Answer 1

我会写伪代码，因为我对 Kafka 不太了解。

首先在读取时，Mybatis默认行为是在List中返回结果，但你不想将300万个对象加载到内存中。必须使用org.apache.ibatis.session.ResultHandler<T>的自定义实现来覆盖它。

public void handleResult(final ResultContext<YourType> context) {
    addToKafkaQueue(context.getResultObject());
}

同时设置fetchSize如果 Mybatis 全局设置中没有定义值，则语句的值(当使用基于注释的映射器时: @Option(fetchSize=500) )。如果未设置，则此选项默认依赖于驱动程序值，每个数据库供应商都不同。这定义了一次将在结果集中缓冲多少记录。例如:对于 Oracle，该值 10:通常太低，因为涉及从应用程序到数据库的大量读取操作；对于 Postgresql，这是无限的(整个结果集)，然后太多了。您必须在速度和内存使用之间找到适当的平衡。

更新:

do {
    YourType object = readFromKafkaQueue();
    mybatisMapper.update(object);
} while (kafkaQueueHasMoreElements());
sqlSession.flushStatement(); // only when using ExecutorType.BATCH

最重要的是ExecutorType (这是 SessionFactory.openSession() 中的参数)或者 ExecutorType.REUSE这将允许只准备一次语句，而不是在每次迭代时都使用默认值ExecutorType.SIMPLE或ExecutorType.BATCH这将堆叠语句并仅在刷新时实际执行它们。

现在仍然需要考虑事务:这可能涉及提交 300 万次更新，或者可以分段。