如何让Spring Boot结合Atomikos实现@JmsListener的事务性？

你当前的问题核心是JMS事务与数据库事务处于两个独立的本地事务上下文：

• JMS监听器容器的setSessionTransacted(true)仅管理JMS会话的本地事务，方法抛异常时JMS事务回滚，消息重新入队；
• 数据库的@Transactional是Spring本地事务，当createRecord()执行完dao.save()且未抛异常时，本地事务已提交，后续监听器方法再抛异常无法回滚数据库记录。

要实现完全事务一致性，必须将JMS与DB2操作纳入同一个XA分布式事务，通过Atomikos统一协调管理。

1. 添加依赖 #

确保项目依赖中包含JTA Atomikos、ActiveMQ及DB2驱动：

<!-- Spring Boot JTA Atomikos 分布式事务支持 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>
<!-- ActiveMQ 消息队列依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<!-- DB2 JDBC驱动 -->
<dependency>
    <groupId>com.ibm.db2</groupId>
    <artifactId>db2jcc</artifactId>
    <version>10.5</version>
</dependency>

2. 配置Atomikos JMS XA连接工厂 #

替换原有JMS配置，使用Atomikos包装ActiveMQ的XA连接工厂，让JMS资源参与分布式事务：

@EnableJms
@Configuration
public class MQConfig {

    @Value("${activemq.broker-url}")
    private String brokerUrl;

    // 初始化ActiveMQ XA连接工厂
    @Bean
    public ActiveMQXAConnectionFactory activeMQXAConnectionFactory() {
        ActiveMQXAConnectionFactory xaConnFactory = new ActiveMQXAConnectionFactory();
        xaConnFactory.setBrokerURL(brokerUrl);
        
        // 配置重发策略
        RedeliveryPolicy rp = new RedeliveryPolicy();
        rp.setMaximumRedeliveries(3);
        rp.setRedeliveryDelay(1000L);
        xaConnFactory.setRedeliveryPolicy(rp);
        
        return xaConnFactory;
    }

    // 用Atomikos包装XA连接工厂，注册为全局事务资源
    @Bean
    public AtomikosConnectionFactoryBean jmsConnectionFactory() {
        AtomikosConnectionFactoryBean factoryBean = new AtomikosConnectionFactoryBean();
        factoryBean.setUniqueResourceName("activemq-xa-resource");
        factoryBean.setXaConnectionFactory(activeMQXAConnectionFactory());
        factoryBean.setMaxPoolSize(10);
        return factoryBean;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        JmsTemplate template = new JmsTemplate(jmsConnectionFactory());
        template.setSessionTransacted(false); // 禁用本地事务，由XA全局事务管理
        return template;
    }

    // 配置JMS监听器容器，关联JTA事务管理器
    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(PlatformTransactionManager transactionManager) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(jmsConnectionFactory());
        factory.setCacheLevelName("CACHE_CONSUMER");
        factory.setReceiveTimeout(1000L);
        factory.setTransactionManager(transactionManager); // 绑定全局事务管理器
        factory.setSessionTransacted(false); // 禁用JMS本地事务
        factory.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED); // 配合XA事务使用
        return factory;
    }
}

3. 配置Atomikos DB2 XA数据源 #

将DB2数据源包装为Atomikos XA数据源，让数据库操作参与分布式事务：

@Configuration
public class DataSourceConfig {

    @Value("${db2.url}")
    private String dbUrl;
    @Value("${db2.username}")
    private String dbUsername;
    @Value("${db2.password}")
    private String dbPassword;

    // 初始化DB2 XA数据源
    @Bean
    public DB2XADataSource db2XADataSource() {
        DB2XADataSource xaDataSource = new DB2XADataSource();
        xaDataSource.setURL(dbUrl);
        xaDataSource.setUser(dbUsername);
        xaDataSource.setPassword(dbPassword);
        return xaDataSource;
    }

    // 用Atomikos包装XA数据源，注册为全局事务资源
    @Bean
    public AtomikosDataSourceBean dataSource() {
        AtomikosDataSourceBean dataSourceBean = new AtomikosDataSourceBean();
        dataSourceBean.setUniqueResourceName("db2-xa-resource");
        dataSourceBean.setXaDataSource(db2XADataSource());
        dataSourceBean.setMinPoolSize(5);
        dataSourceBean.setMaxPoolSize(20);
        dataSourceBean.setTestQuery("SELECT 1 FROM SYSIBM.SYSDUMMY1");
        return dataSourceBean;
    }
}

4. 配置JTA全局事务管理器 #

将Atomikos事务管理器配置为Spring默认事务管理器：

@Configuration
@EnableTransactionManagement
public class TransactionConfig {

    @Bean
    public UserTransaction userTransaction() throws SystemException {
        UserTransactionImp userTransaction = new UserTransactionImp();
        userTransaction.setTransactionTimeout(300); // 全局事务超时时间（秒）
        return userTransaction;
    }

    @Bean
    public TransactionManager atomikosTransactionManager() {
        UserTransactionManager txManager = new UserTransactionManager();
        txManager.setForceShutdown(false);
        return txManager;
    }

    @Bean
    public PlatformTransactionManager transactionManager(UserTransaction userTransaction, TransactionManager atomikosTransactionManager) {
        return new JtaTransactionManager(userTransaction, atomikosTransactionManager);
    }
}

5. 调整服务类事务注解 #

确保服务层事务使用全局JTA事务管理器：

@Service
public class ImportRecordsService {

    @Autowired
    private ImportRecordsDAO dao;

    // 指定使用JTA全局事务管理器
    @Transactional(transactionManager = "transactionManager")
    public void createRecord() {
        ImportRecord record = new ImportRecord();
        record.setDateCreated(LocalDateTime.now());
        record.setName("test-001");
        dao.save(record);
        // 若此处抛出异常，XA事务会同时回滚DB插入与JMS消息确认
        // throw new RuntimeException("测试分布式事务回滚");
    }
}

6. 补充配置文件参数 #

在application.properties中添加基础配置：

# ActiveMQ配置
activemq.broker-url=tcp://localhost:61616

# DB2配置
db2.url=jdbc:db2://localhost:50000/your-db-name
db2.username=db-user
db2.password=db-password

# Atomikos日志配置
spring.jta.atomikos.properties.log-base-dir=./atomikos-transaction-logs

关键说明 #

• XA事务协调：Atomikos作为全局事务管理器，会同时跟踪JMS与DB2的XA资源，确保所有操作要么全部提交，要么全部回滚；
• 禁用本地事务：必须关闭JMS容器与JmsTemplate的本地事务，避免与全局XA事务冲突；
• 资源唯一标识：每个XA资源的uniqueResourceName必须唯一，Atomikos通过该标识区分不同的事务资源。

完成以上配置后，当监听器方法中调用createRecord()后抛出异常，XA事务会同时回滚数据库插入操作与JMS消息确认，实现完全的事务一致性。

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