EntityManagerFactory 在使用@DirtiesContext 重新加载上下文后关闭

Question

我有一个 Spring Boot 应用程序，它使用 JMS 连接到队列并侦听传入消息。在应用程序中，我有一个集成测试，它将一些消息发送到队列，然后确保当侦听器接收到新消息时应该发生的事情实际发生。

我已经用@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD) 注释了我的测试类，以确保我的数据库在每次测试后都是干净的。每个测试在单独运行时都会通过。但是，在第一个测试成功通过后一起运行它们时，下一个测试会失败，但当被测代码尝试将实体保存到数据库时，会出现以下异常：

    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:431) ~[spring-orm-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:373) ~[spring-tx-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:447) ~[spring-tx-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:277) ~[spring-tx-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96) ~[spring-tx-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) ~[spring-aop-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at com.sun.proxy.$Proxy95.handleWorkflowEvent(Unknown Source) ~[na:na]
    at com.mottmac.processflow.infra.jms.EventListener.onWorkflowEvent(EventListener.java:51) ~[classes/:na]
    at com.mottmac.processflow.infra.jms.EventListener.onMessage(EventListener.java:61) ~[classes/:na]
    at org.apache.activemq.ActiveMQMessageConsumer.dispatch(ActiveMQMessageConsumer.java:1401) [activemq-client-5.14.3.jar:5.14.3]
    at org.apache.activemq.ActiveMQSessionExecutor.dispatch(ActiveMQSessionExecutor.java:131) [activemq-client-5.14.3.jar:5.14.3]
    at org.apache.activemq.ActiveMQSessionExecutor.iterate(ActiveMQSessionExecutor.java:202) [activemq-client-5.14.3.jar:5.14.3]
    at org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133) [activemq-client-5.14.3.jar:5.14.3]
    at org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48) [activemq-client-5.14.3.jar:5.14.3]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [na:1.8.0_77]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [na:1.8.0_77]
    at java.lang.Thread.run(Unknown Source) [na:1.8.0_77]
Caused by: java.lang.IllegalStateException: EntityManagerFactory is closed
    at org.hibernate.jpa.internal.EntityManagerFactoryImpl.validateNotClosed(EntityManagerFactoryImpl.java:367) ~[hibernate-entitymanager-5.0.11.Final.jar:5.0.11.Final]
    at org.hibernate.jpa.internal.EntityManagerFactoryImpl.internalCreateEntityManager(EntityManagerFactoryImpl.java:316) ~[hibernate-entitymanager-5.0.11.Final.jar:5.0.11.Final]
    at org.hibernate.jpa.internal.EntityManagerFactoryImpl.createEntityManager(EntityManagerFactoryImpl.java:286) ~[hibernate-entitymanager-5.0.11.Final.jar:5.0.11.Final]
    at org.springframework.orm.jpa.JpaTransactionManager.createEntityManagerForTransaction(JpaTransactionManager.java:449) ~[spring-orm-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.orm.jpa.JpaTransactionManager.doBegin(JpaTransactionManager.java:369) ~[spring-orm-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    ... 17 common frames omitted

我的测试课：

    @RunWith(SpringRunner.class)
@SpringBootTest(classes = { TestGovernance.class })
@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD)
public class ActivitiIntegrationTest
{
    private static final String TEST_PROCESS_KEY = "oneTaskProcess";
    private static final String FIRST_TASK_KEY = "theTask";
    private static final String NEXT_TASK_KEY = "nextTask";

    @Autowired
    private JmsTemplate jms;

    @Autowired
    private WorkflowEventRepository eventRepository;

    @Autowired
    private TaskService taskService;

    @Test
    public void workFlowEventForRunningTaskMovesItToTheNextStage() throws InterruptedException
    {
        sendMessageToCreateNewInstanceOfProcess(TEST_PROCESS_KEY);

        Task activeTask = getActiveTask();        
        assertThat(activeTask.getTaskDefinitionKey(), is(FIRST_TASK_KEY));

        sendMessageToUpdateExistingTask(activeTask.getProcessInstanceId(), FIRST_TASK_KEY);

        Task nextTask = getActiveTask();        
        assertThat(nextTask.getTaskDefinitionKey(), is(NEXT_TASK_KEY));
    }

    @Test
    public void newWorkflowEventIsSavedToDatabaseAndKicksOffTask() throws InterruptedException
    {
        sendMessageToCreateNewInstanceOfProcess(TEST_PROCESS_KEY);

        assertThat(eventRepository.findAll(), hasSize(1));
    }

    @Test
    public void newWorkflowEventKicksOffTask() throws InterruptedException
    {
        sendMessageToCreateNewInstanceOfProcess(TEST_PROCESS_KEY);

        Task activeTask = getActiveTask();        
        assertThat(activeTask.getTaskDefinitionKey(), is(FIRST_TASK_KEY));
    }


    private void sendMessageToUpdateExistingTask(String processId, String event) throws InterruptedException
    {
        WorkflowEvent message = new WorkflowEvent();
        message.setRaisedDt(ZonedDateTime.now());
        message.setEvent(event);
        // Existing
        message.setIdWorkflowInstance(processId);
        jms.convertAndSend("workflow", message);
        Thread.sleep(5000);
    }

    private void sendMessageToCreateNewInstanceOfProcess(String event) throws InterruptedException
    {
        WorkflowEvent message = new WorkflowEvent();
        message.setRaisedDt(ZonedDateTime.now());
        message.setEvent(event);
        jms.convertAndSend("workflow", message);
        Thread.sleep(5000);
    }

    private Task getActiveTask()
    {
        // For some reason the tasks in the task service are hanging around even
        // though the context is being reloaded. This means we have to get the
        // ID of the only task in the database (since it has been cleaned
        // properly) and use it to look up the task.
        WorkflowEvent workflowEvent = eventRepository.findAll().get(0);
        Task activeTask = taskService.createTaskQuery().processInstanceId(workflowEvent.getIdWorkflowInstance().toString()).singleResult();
        return activeTask;
    }

}

在应用程序中抛出异常的方法（repository只是一个标准的Spring Data CrudRepository）：

    @Override
    @Transactional
    public void handleWorkflowEvent(WorkflowEvent event)
    {
        try
        {
            logger.info("Handling workflow event[{}]", event);

            // Exception is thrown here:
            repository.save(event);

            logger.info("Saved event to the database [{}]", event);
            if(event.getIdWorkflowInstance() == null)
            {
                String newWorkflow = engine.newWorkflow(event.getEvent(), event.getVariables());
                event.setIdWorkflowInstance(newWorkflow);
            }
            else 
            {
                engine.moveToNextStage(event.getIdWorkflowInstance(), event.getEvent(), event.getVariables());
            }
        }
        catch (Exception e)
        {
            logger.error("Error while handling workflow event:" , e);
        }
    }

我的测试配置类：

@SpringBootApplication
@EnableJms
@TestConfiguration
public class TestGovernance
{
    private static final String WORKFLOW_QUEUE_NAME = "workflow";

    @Bean
    public ConnectionFactory connectionFactory()
    {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
        return connectionFactory;
    }

    @Bean
    public EventListenerJmsConnection connection(ConnectionFactory connectionFactory) throws NamingException, JMSException
    {
        // Look up ConnectionFactory and Queue
        Destination destination = new ActiveMQQueue(WORKFLOW_QUEUE_NAME);

        // Create Connection
        Connection connection = connectionFactory.createConnection();

        Session listenerSession = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        MessageConsumer receiver = listenerSession.createConsumer(destination);

        EventListenerJmsConnection eventListenerConfig = new EventListenerJmsConnection(receiver, connection);
        return eventListenerConfig;
    }
}

JMS 消息监听器（不确定是否有帮助）：

/**
 * Provides an endpoint which will listen for new JMS messages carrying
 * {@link WorkflowEvent} objects.
 */
@Service
public class EventListener implements MessageListener
{
    Logger logger = LoggerFactory.getLogger(EventListener.class);

    private WorkflowEventHandler eventHandler;

    private MessageConverter messageConverter;

    private EventListenerJmsConnection listenerConnection;

    @Autowired
    public EventListener(EventListenerJmsConnection listenerConnection, WorkflowEventHandler eventHandler, MessageConverter messageConverter)
    {
        this.eventHandler = eventHandler;
        this.messageConverter = messageConverter;
        this.listenerConnection = listenerConnection;
    }

    @PostConstruct
    public void setUpConnection() throws NamingException, JMSException
    {
        listenerConnection.setMessageListener(this);
        listenerConnection.start();
    }

    private void onWorkflowEvent(WorkflowEvent event)
    {
        logger.info("Recieved new workflow event [{}]", event);
        eventHandler.handleWorkflowEvent(event);
    }

    @Override
    public void onMessage(Message message)
    {
        try
        {
            message.acknowledge();
            WorkflowEvent fromMessage = (WorkflowEvent) messageConverter.fromMessage(message);
            onWorkflowEvent((WorkflowEvent) fromMessage);
        }
        catch (Exception e)
        {
            logger.error("Error: ", e);
        }
    }
}

我已尝试添加 @Transactional' to the test methods and removing it from the code under test and various combinations with no success. I've also tried adding various test execution listeners and I still can't get it to work. If I remove the@DirtiesContext`，然后异常消失，所有测试均无异常运行（但正如我所料，它们确实因断言错误而失败）。

任何帮助将不胜感激。到目前为止，我的搜索还没有出现任何结果，一切都表明 @DirtiesContext 应该可以工作。

Answer 1

为此使用@DirtiesContext 是一个糟糕的主意（恕我直言），您应该做的是测试@Transactional。我还建议删除Thread.sleep 并改用awaitility 之类的东西。

理论上，当您执行查询时，应提交所有未决更改，因此您可以使用等待时间检查最多 6 秒，以查看数据库中是否已保留某些内容。如果这不起作用，您可以尝试在查询之前添加刷新。

@RunWith(SpringRunner.class)
@SpringBootTest(classes = { TestGovernance.class })
@Transactional
public class ActivitiIntegrationTest {

    private static final String TEST_PROCESS_KEY = "oneTaskProcess";
    private static final String FIRST_TASK_KEY = "theTask";
    private static final String NEXT_TASK_KEY = "nextTask";

    @Autowired
    private JmsTemplate jms;

    @Autowired
    private WorkflowEventRepository eventRepository;

    @Autowired
    private TaskService taskService;

    @Autowired
    private EntityManager em;

    @Test
    public void workFlowEventForRunningTaskMovesItToTheNextStage() throws InterruptedException
    {
        sendMessageToCreateNewInstanceOfProcess(TEST_PROCESS_KEY);

        await().atMost(6, SECONDS).until(getActiveTask() != null);

        Task activeTask = getActiveTask());
        assertThat(activeTask.getTaskDefinitionKey(), is(FIRST_TASK_KEY));

        sendMessageToUpdateExistingTask(activeTask.getProcessInstanceId(), FIRST_TASK_KEY);

        Task nextTask = getActiveTask();        
        assertThat(nextTask.getTaskDefinitionKey(), is(NEXT_TASK_KEY));
    }

    private Task getActiveTask()
    {
        em.flush(); // simulate a commit
        // For some reason the tasks in the task service are hanging around even
        // though the context is being reloaded. This means we have to get the
        // ID of the only task in the database (since it has been cleaned
        // properly) and use it to look up the task.
        WorkflowEvent workflowEvent = eventRepository.findAll().get(0);
        Task activeTask = taskService.createTaskQuery().processInstanceId(workflowEvent.getIdWorkflowInstance().toString()).singleResult();
        return activeTask;
    }

}

您可能需要/想要对您的 getActiveTask 进行一些改进，以便能够使用 return null，或者此更改甚至可能使其行为符合您的预期。

我只是做了一个方法，其他方法你可能自己想出来。您使用这种方法获得的收益可能是 2 倍，1 它不会再等待 5 秒，但会更少，并且您不必在测试之间重新加载整个应用程序。两者都应该使您的测试更快。