Apache ActiveMQ Camel 事务回滚

Question

为了更好地理解ActiveMQ和Camel，我正在编写一个事务回滚的单元测试。但是似乎测试没有成功，这意味着我可能做错了什么！以下是代码:

public class MyTest extends CamelTestSupport {
@Override
protected JndiRegistry createRegistry() throws Exception {
    JndiRegistry reg = super.createRegistry();

    DataSourceTransactionManager txMgr = new DataSourceTransactionManager();

    SpringTransactionPolicy txPolicy = new SpringTransactionPolicy();
    txPolicy.setTransactionManager(txMgr);
    txPolicy.setPropagationBehaviorName("PROPAGATION_REQUIRED");
    reg.bind("required", txPolicy);

    return reg;
}

@Before
public void setUp() throws Exception {
    super.setUp();
}

@After
public void tearDown() throws Exception {
    super.tearDown();
}

@Test
public void testTransaction() throws Exception {
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
    context.addComponent("jms", JmsComponent.jmsComponentTransacted(connectionFactory));
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("jms:queue:in")
                    .transacted("required")
                    .process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            System.out.println("Expected failure");
                            throw new RuntimeException("Expected failure");
                        }
                    })
                    .to("mock:result");
        }
    });
    context.start();

    MockEndpoint result = context.getEndpoint("mock:result", MockEndpoint.class);
    result.expectedMessageCount(0);

    NotifyBuilder notifyBuilder = new NotifyBuilder(context).whenDone(1).create();
    context.createProducerTemplate().sendBody("jms:queue:in", "Test");

    boolean matches = notifyBuilder.matches(5, TimeUnit.SECONDS);
    assertTrue(matches);

    Thread.sleep(1000);
    assertMockEndpointsSatisfied();

    BrowsableEndpoint in = context.getEndpoint("jms:queue:in", BrowsableEndpoint.class);
    List<Exchange> list = in.getExchanges();
    assertEquals(1, list.size());
    String body = list.get(0).getIn().getBody(String.class);
    assertEquals("Test", body);

    context.stop();
}

这个断言失败是因为列表的大小不是1，如果回滚成功应该通过。我做错了什么？谢谢！

- Jake

Answer 1

在您的情况下，您正在使用DataSourceTransactionManager，而您需要的是一个引用您的ConnectionFactory的org.springframework.jms.connection.JmsTransactionManager。还可以在JMS URI中添加?transacted=true而不引用事务管理器，这将使用本地JMS事务。

在这两种情况下，默认情况下消息将最终进入ActiveMQ中的死信队列，而不是返回原始队列。可以对此行为进行配置。请参考配置方法。

- Jakub Korab