用JPA和JMS进行春季集成测试的交易传播

Question

请帮我理解以下内容。

我有一个Spring集成测试，我想测试ProcessCommentsDao类的方法：）

@RunWith(SpringJUnit4ClassRunner.class) 
@ContextConfiguration(locations = {"classpath:testContext.xml"}) 
@Transactional() 
public class ParseCommentsTest { 

    @Resource 
    private ProcessCommentsDao processCommentsDao; 

    @Test 
    public void testJMS() throws Exception { 

    // Test data creation 
    ......................... 

    processCommentsDao.parseComments(); 
    } 
} 

在该方法中parseComments（，我得到实体的名单，然后每个实体处理通过Spring的JMS消息监听实现：

@Service 
public class ProcessCommentsDaoImpl extends BaseDao implements IProcessCommentsDao { 

    private static final int PARSE_COMMENTS_COUNT_LIMIT = 100; 
    @Autowired 
    private JmsTemplate jmsTemplate; 
    @Autowired 
    private Queue parseCommentsDestination; 

    @Override 
    public void parseComments() { 

     List<Comment> comments = attributeDao.findTopUnparsedComments(PARSE_COMMENTS_COUNT_LIMIT); 

     for (Comment comment : comments) { 
     jmsTemplate.convertAndSend(parseCommentsDestination, comment); 
     } 
    } 
} 

了MessageListener的执行标准如下：

@Component 
public class QueueListener implements MessageListener { 

    @PersistenceContext 
    private EntityManager em; 

    @Transactional() 
    public void onMessage(final Message message) { 
     try { 
     if (message instanceof ObjectMessage) { 
      final ObjectMessage objMessage = (ObjectMessage) message; 
      Comment comment = (Comment) objMessage.getObject(); 

      //...Some logic ... 

      comment = em.find(Comment.class, comment.getId()); 
      comment.setPosStatus(ParsingType.PROCESSED); 
      em.merge(comment); 

       //...Some logic ... 

    } catch (final JMSException e) { 
     e.printStackTrace(); 
    } 
} 

}

其结果是，该方法em.find（Comment.class，comment.getId（））返回null，因为数据是在另一个线程创建和当前线程不知道这个交易什么。有没有办法设置事务传播，以便MessageListener方法可以在主线程中创建，其中测试方法在哪里运行？

Answer 1

其结果是，该方法em.find（Comment.class，comment.getId（））返回null，因为数据是在另一个线程创建和当前线程不知道这个交易什么。

更准确地说，消息侦听器在单独的事务中运行，并且它看不到由ParseCommentsTest.testJMS创建的数据，因为该方法没有提交。

更重要的是，您的测试写得不正确。它有一个竞争条件：对jmsTemplate.convertAndSend()的调用是异步的，所以QueueListener.messageListener()中的逻辑可以在测试方法完成后调用（并回滚它所做的更改）。每次运行时，此测试可能会产生不同的结果。

您的代码也不容易测试。考虑将处理逻辑从onMessage()方法提取到POJO并单独进行测试。

Answer 2

我找到了我的问题的下一个解决方案。测试数据在一个单独的事务，这创造manualy产生：

@RunWith(SpringJUnit4ClassRunner.class) 
@ContextConfiguration(locations = {"classpath:testContext.xml"}) 
@Transactional() 
public class ParseCommentsTest { 

    @Resource 
    private ProcessCommentsDao processCommentsDao; 
    @Autowired 
    private PlatformTransactionManager transactionManager; 

    @Before 
    public void tearUp() { 
    createTestData(); 
    } 

    @Test 
    @Rollback(false) 
    public void testJMS() throws Exception { 
    processCommentsDao.parseComments(); 
    } 

    @After 
    public void tearDown() { 
    removeTestData(); 
} 

private void createTestData() { 
    TransactionTemplate txTemplate = new TransactionTemplate(transactionManager); 
    txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); 
    txTemplate.execute(new TransactionCallback<Object>() { 

     @Override 
     public Object doInTransaction(TransactionStatus status) { 
      try { 
      // Test data creation 
      ........................... 
     } 
    }); 
    } 
} 

在方法ProcessCommentsDaoImpl.parseComments（）被实现等待所有异步JMS请求的完成。主线程完成其工作，直到所有的实体处理：

@Service 
public class ProcessCommentsDaoImpl extends BaseDao implements IProcessCommentsDao { 

    private static final int PARSE_COMMENTS_COUNT_LIMIT = 100; 
    @Autowired 
    private JmsTemplate jmsTemplate; 
    @Autowired 
    private Queue parseCommentsDestination; 

    @Override 
    public void parseComments() { 

    List<Comment> comments = attributeDao.findTopUnparsedComments(PARSE_COMMENTS_COUNT_LIMIT); 

    for (Comment comment : comments) { 
    jmsTemplate.convertAndSend(parseCommentsDestination, comment); 
    } 
    // Wait all request procesed 
    waitParseCommentsProcessed(comments); 
    } 

    @Override 
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW) 
    public void parseComment(Long commentId) { 
    ...................... 
    //Some logic 
    .................... 
    } 
} 

而且MessageLisener的一个重构如下：ParseCommentQueueListener的

public class ParseCommentQueueListener { 

    private static Logger log = Logger.getLogger(ParseCommentQueueListener.class); 

    @Resource(name = SpringContext.DAO_PROCESS_COMMENTS) 
    private IProcessCommentsDao processCommentsDao; 

    public Object receive(ObjectMessage message) { 
    try { 
     Long id = (Long) message.getObject(); 
     processCommentsDao.parseComment(id); 
    } catch (JMSException ex) { 
     log.error(ex.toString()); 
    } 
    return message; 
    } 
} 

xml配置是如下：

<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter"> 
    <constructor-arg> 
     <bean class="com.provoxlabs.wordminer.parsing.ParseCommentQueueListener"/> 
    </constructor-arg> 
    <property name="defaultListenerMethod" value="receive"/> 
    <property name="defaultResponseDestination" ref="parseCommentsStatusDestination"/> 
    <!-- we don't want automatic message context extraction --> 
    <property name="messageConverter"> 
     <null/> 
    </property> 
</bean>