多线程jpa读取时休眠空指针异常

Question

我有一个 spring-batch 作业，它使用 RepositoryItemReader 从数据库中读取，然后将结果转换为 Map，然后将结果写入 elasticsearch。它工作正常，虽然有点慢。所以现在我想添加一个池大小为 4 的 taskExecutor 以加快速度：

        return stepBuilders.get("search-export").<AbstractEntityDefinition, Map<String, Object>>chunk(nemesisSearchProperties.getExport().getChunkSize())
                        .reader(reader)
                        .processor(processor)
                        .writer(writer)
                        .stream(reader)
                        .transactionAttribute(transactionAttribute)
                        .listener(chunkSessionReplicatorExecutionListener)
                        .listener(new NemesisChunkLoggingListener(indexName + " search export", nemesisSearchProperties.getExport().getChunkSize()))
                        .taskExecutor(searchExportTaskExecutor).throttleLimit(4) // <-- I add this
                        .build();

但是，当我添加任务执行器时，我得到了这个：

java.lang.NullPointerException
    at org.hibernate.internal.util.collections.IdentityMap.entryArray(IdentityMap.java:162)
    at org.hibernate.internal.util.collections.IdentityMap.concurrentEntries(IdentityMap.java:58)
    at org.hibernate.engine.internal.StatefulPersistenceContext.forEachCollectionEntry(StatefulPersistenceContext.java:1135)
    at org.hibernate.event.internal.AbstractFlushingEventListener.prepareCollectionFlushes(AbstractFlushingEventListener.java:193)
    at org.hibernate.event.internal.AbstractFlushingEventListener.flushEverythingToExecutions(AbstractFlushingEventListener.java:85)

看着IdentityMap我可以看到这个：

(1)     if ( entryArray == null ) {
(2)         entryArray = new Map.Entry[ map.size() ];
            final Iterator<Entry<IdentityKey<K>, V>> itr = map.entrySet().iterator();
            int i = 0;
            while ( itr.hasNext() ) {
                final Entry<IdentityKey<K>, V> me = itr.next();
(3)             entryArray[i++] = new IdentityMapEntry( me.getKey().key, me.getValue() ); // Here entryArray is NULL!!!
            }
        }

异常发生在第 (3) 行，其中entryArray 为空。而且我想知道它是如何做到的，因为entryArray 在第 (1) 行被检查为空并在第 (2) 行初始化。

任何想法将不胜感激。

Answer 1

我不知道您的阅读器、处理器和编写器是什么样的，但这看起来像是一个并发问题，因为在线程之间共享 Session。确保每个线程都有单独的会话。

查看文档：https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html

对于一些常见的批处理用例，使用多线程 Step 实现存在一些实际限制。 Step 中的许多参与者（例如 reader 和 writer）都是有状态的。如果状态不是按线程隔离的，那么这些组件在多线程步骤中不可用。特别是 Spring Batch 中的大多数现成的 reader 和 writer 都不是为多线程使用而设计的。

最好为所有对象创建一个包装器，将它们各自的读取器/写入器等存储在本地线程中。

Java/Jakarta Batch 做到了这一点。读取器、写入器、处理器实例始终位于每个分区，因此您永远不会遇到问题。