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

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为null。我想知道为什么会这样，因为在第(1)行检查entryArray为空，并在第(2)行将其初始化。

任何想法都将不胜感激。

Answer 1

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

请参阅文档：https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html

对于一些常见的批处理用例，使用多线程步骤实现有一些实际限制。步骤中的许多参与者(例如读取器和写入器)都是有状态的。如果状态不是由线程隔离的，那么这些组件在多线程步骤中是不可用的。特别是，来自Spring Batch的大多数现成的读取器和写入器都不是为多线程使用而设计的。

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

Java/Jakarta Batch做到了这一点。读取器、写入器、处理器实例总是按分区计算的，因此您永远不会遇到问题。