Spring Boot JPA应用Flyway迁移报错：EntityManagerFactory已关闭

问题现象 #

基于PostgreSQL的Spring Boot Data JPA应用，Flyway迁移脚本在测试环境正常，但部署时抛出EntityManagerFactory is closed异常。将Repository返回Stream<MessageEntity>的方法替换为返回集合的findAll()后，迁移正常执行。因数据量较大，需避免一次性加载所有数据导致超时，因此需要解决Stream方式的问题，同时寻求其他大数据量迁移方案。

报错堆栈 #

org.springframework.beans.factory.BeanCreationException: 
Error creating bean with name 'flywayInitializer' defined in class path resource 
[org/springframework/boot/autoconfigure/flyway/FlywayAutoConfiguration$FlywayConfiguration.class]: 
Invocation of init method failed; nested exception is org.flywaydb.core.internal.command.DbMigrate$FlywayMigrateException: 
Migration failed !

Caused by: org.springframework.transaction.CannotCreateTransactionException: Could not open JPA EntityManager for transaction; nested exception is java.lang.IllegalStateException: EntityManagerFactory is closed
    at org.springframework.orm.jpa.JpaTransactionManager.doBegin(JpaTransactionManager.java:467)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.startTransaction(AbstractPlatformTransactionManager.java:400)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:373)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:568)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208)
    at jdk.proxy2/jdk.proxy2.$Proxy183.getTransaction(Unknown Source)
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137)
    at org.springframework.transaction.support.TransactionOperations.executeWithoutResult(TransactionOperations.java:67)
    at com.life.adapter.out.jpa.migration.V1689943999__Migrate_Documents.migrate(V1689943999__Migrate_Documents.kt:23)
    at org.flywaydb.core.internal.resolver.java.JavaMigrationExecutor.executeOnce(JavaMigrationExecutor.java:55)
    ... 47 common frames omitted
Caused by: java.lang.IllegalStateException: EntityManagerFactory is closed
    at org.hibernate.internal.SessionFactoryImpl.validateNotClosed(SessionFactoryImpl.java:547)
    at org.hibernate.internal.SessionFactoryImpl.createEntityManager(SessionFactoryImpl.java:636)
    at org.hibernate.internal.SessionFactoryImpl.createEntityManager(SessionFactoryImpl.java:158)
    at org.springframework.orm.jpa.AbstractEntityManagerFactoryBean.createNativeEntityManager(AbstractEntityManagerFactoryBean.java:585)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:568)
    at org.springframework.orm.jpa.AbstractEntityManagerFactoryBean.invokeProxyMethod(AbstractEntityManagerFactoryBean.java:487)
    at org.springframework.orm.jpa.AbstractEntityManagerFactoryBean$ManagedEntityManagerFactoryInvocationHandler.invoke(AbstractEntityManagerFactoryBean.java:734)
    at jdk.proxy2/jdk.proxy2.$Proxy227.createNativeEntityManager(Unknown Source)
    at org.springframework.orm.jpa.JpaTransactionManager.createEntityManagerForTransaction(JpaTransactionManager.java:485)
    at org.springframework.orm.jpa.JpaTransactionManager.doBegin(JpaTransactionManager.java:410)
    ... 60 common frames omitted

相关代码 #

迁移类 #

@Component
class V1689943999__Migrate_Documents(
    @Lazy private val transactionManager: PlatformTransactionManager,
    @Lazy private val repository: MessageEntityRepository
) : BaseJavaMigration() {

    override fun migrate(context: Context) {
        TransactionTemplate(transactionManager).executeWithoutResult {
            messageRepository.findAllByOrderByCreatedAt().use { entities ->
                entities.forEach { migrate(it) }
            }
        }
    }

    private fun migrate(entity: MessageEntity) {
        ... // 业务处理逻辑
        repository.save(entity)
    }
}

Repository接口 #

@Repository
interface MessageEntityRepository : JpaRepository<MessageEntity, UUID> {
    fun findAllByOrderByCreatedAt(): Stream<MessageEntity>
}

依赖版本 #

implementation("org.flywaydb:flyway-core:8.5.13")

原因分析 #

1. Flyway执行时机问题：Flyway迁移在Spring上下文初始化阶段执行，此时Spring Data JPA的EntityManagerFactory尚未完全完成初始化，状态不稳定。
2. Stream的生命周期特性：Spring Data JPA的Stream返回类型会在Stream关闭时（use块执行完毕）自动关闭对应的EntityManager。在迁移场景下，这会导致EntityManagerFactory被提前关闭，后续的save操作无法获取新的EntityManager，从而抛出异常。
3. 集合类型的兼容性：使用集合类型返回数据时，EntityManager不会被提前关闭，因此迁移能正常执行，但大数据量下会导致内存占用过高或超时。

解决方案 #

方案1：改用Flyway JDBC连接直接操作（推荐） #

绕开Spring Data JPA，直接使用Flyway提供的JDBC连接进行分批读写，完全避免EntityManager相关问题，适合大数据量迁移：

class V1689943999__Migrate_Documents : BaseJavaMigration() {

    override fun migrate(context: Context) {
        val connection = context.connection
        connection.autoCommit = false
        val batchSize = 1000
        var offset = 0

        try {
            while (true) {
                // 分批查询数据
                val selectStmt = connection.prepareStatement(
                    "SELECT id, content, created_at FROM message_entity ORDER BY created_at LIMIT ? OFFSET ?"
                )
                selectStmt.setInt(1, batchSize)
                selectStmt.setInt(2, offset)
                val rs = selectStmt.executeQuery()

                var hasData = false
                // 批量更新
                val updateStmt = connection.prepareStatement(
                    "UPDATE message_entity SET content = ? WHERE id = ?"
                )

                while (rs.next()) {
                    hasData = true
                    val id = rs.getObject("id", UUID::class.java)
                    val originalContent = rs.getString("content")
                    // 业务处理逻辑
                    val updatedContent = originalContent.trim()

                    updateStmt.setString(1, updatedContent)
                    updateStmt.setObject(2, id)
                    updateStmt.addBatch()
                }

                if (!hasData) break

                updateStmt.executeBatch()
                connection.commit()
                offset += batchSize

                rs.close()
                selectStmt.close()
                updateStmt.close()
            }
        } finally {
            connection.autoCommit = true
            connection.close()
        }
    }
}

方案2：分页查询替代Stream #

通过分页分批加载数据，既避免一次性加载所有数据，又兼容当前Spring Data JPA的使用方式：

@Component
class V1689943999__Migrate_Documents(
    @Lazy private val transactionManager: PlatformTransactionManager,
    @Lazy private val repository: MessageEntityRepository
) : BaseJavaMigration() {

    private val batchSize = 1000

    override fun migrate(context: Context) {
        TransactionTemplate(transactionManager).executeWithoutResult {
            var pageNum = 0
            while (true) {
                val page = repository.findAll(
                    PageRequest.of(pageNum, batchSize, Sort.by(Sort.Direction.ASC, "createdAt"))
                )
                if (page.isEmpty) break

                page.content.forEach { entity ->
                    // 业务处理逻辑
                    repository.save(entity)
                }
                pageNum++
            }
        }
    }
}

方案3：手动管理EntityManager生命周期 #

通过直接操作EntityManagerFactory创建EntityManager，控制其生命周期，避免Stream自动关闭导致的问题：

@Component
class V1689943999__Migrate_Documents(
    @Lazy private val transactionManager: PlatformTransactionManager,
    @Lazy private val entityManagerFactory: EntityManagerFactory
) : BaseJavaMigration() {

    override fun migrate(context: Context) {
        TransactionTemplate(transactionManager).executeWithoutResult {
            val entityManager = entityManagerFactory.createEntityManager()
            entityManager.joinTransaction()
            try {
                val query = entityManager.createQuery(
                    "SELECT m FROM MessageEntity m ORDER BY m.createdAt", 
                    MessageEntity::class.java
                )
                query.stream.use { stream ->
                    stream.forEach { entity ->
                        // 业务处理逻辑
                        entityManager.merge(entity)
                    }
                }
            } finally {
                entityManager.close()
            }
        }
    }
}

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