Spring Boot、原子和ActiveMQ的Jms队列重新传递问题

Question

我正在尝试设置一个jms队列，并在事务失败时进行重新交付。出现的情况是，一条消息被多次重新传递（和处理），因为在消息处理代码之后，会话会关闭两次。第二次关闭尝试抛出错误，因为它已经关闭，所以事务被回滚，消息被重新传递。

以下是我的配置：

@Configuration
public class MyJtaConfiguration {

private static final Logger LOGGER = Logger.getLogger(MyJtaConfiguration.class);

@Bean(name = "atomikosUserTransaction")
public UserTransaction atomikosUserTransaction() throws Throwable {
    UserTransactionImp userTransactionImp = new UserTransactionImp();
    userTransactionImp.setTransactionTimeout(10000);

    AtomikosJtaPlatform.transaction = userTransactionImp;

    return userTransactionImp;
}

@Bean(name = "atomikosTransactionManager", initMethod = "init", destroyMethod = "close")
public TransactionManager atomikosTransactionManager() throws Throwable {
    UserTransactionManager userTransactionManager = new UserTransactionManager();
    userTransactionManager.setForceShutdown(false);

    AtomikosJtaPlatform.transactionManager = userTransactionManager;

    return userTransactionManager;
}

@Bean(name = "transactionManager")
@DependsOn({ "atomikosUserTransaction", "atomikosTransactionManager" })
public PlatformTransactionManager transactionManager() throws Throwable {
    UserTransaction userTransaction = atomikosUserTransaction();
    TransactionManager atomikosTransactionManager = atomikosTransactionManager();
    return new JtaTransactionManager(userTransaction, atomikosTransactionManager);
}

@Bean
public QueueConnectionFactory connectionFactory() {     
    Map<String, Object> parameters = new HashMap<>();
    parameters.put("dataDirectory", "activeMqDataDirectory");
    String brokerUrl = Interpolator.getString("vm://MyBroker?broker.persistent=true&broker.dataDirectory=${dataDirectory}&broker.useJmx=true", parameters );


    ActiveMQXAConnectionFactory activeMQXAConnectionFactory = new ActiveMQXAConnectionFactory(brokerUrl);
    activeMQXAConnectionFactory.setTrustedPackages(
            new ArrayList<>(Arrays.asList("java.lang,javax.security,java.util,org.apache.activemq,org.fusesource.hawtbuf,com.thoughtworks.xstream.mapper,com.brabo".split(","))));

    RedeliveryPolicy redeliveryPolicy = activeMQXAConnectionFactory.getRedeliveryPolicy();
    redeliveryPolicy.setInitialRedeliveryDelay(10 * 1000);
    redeliveryPolicy.setMaximumRedeliveryDelay(10 * 1000);
    redeliveryPolicy.setMaximumRedeliveries(2);

    AtomikosConnectionFactoryBean atomikosConnectionFactoryBean = new AtomikosConnectionFactoryBean();
    atomikosConnectionFactoryBean.setUniqueResourceName("xamq");
    atomikosConnectionFactoryBean.setLocalTransactionMode(false);
    atomikosConnectionFactoryBean.setXaConnectionFactory(activeMQXAConnectionFactory);

    try {
        atomikosConnectionFactoryBean.init();
    } catch (JMSException e) {
        throw new RuntimeException(e);
    }
    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
    cachingConnectionFactory.setTargetConnectionFactory(atomikosConnectionFactoryBean);
    cachingConnectionFactory.setSessionCacheSize(50);
    cachingConnectionFactory.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            LOGGER.error(exception);
        }
    });
    return cachingConnectionFactory;
}


@Bean(name = "myTestQueue")
public Queue backgroundTaskQueue() {
    ActiveMQQueue queue = new ActiveMQQueue("myTestQueue");
    return queue;
}


@Bean
public DefaultMessageListenerContainer backgroundTaskQueueListenerContainer(@Autowired ConnectionFactory connectionFactory,
        @Autowired @Qualifier("myServiceProcessorBean") MessageListener messageListener, @Autowired PlatformTransactionManager txManager) {
    return createListenerContainer(connectionFactory, messageListener, txManager, QueueManager.BACKGROUND_TASK_QUEUE_JNDI);
}

private DefaultMessageListenerContainer createListenerContainer(ConnectionFactory connectionFactory, MessageListener messageListener, PlatformTransactionManager txManager,
        String destinationName) {       
    DefaultMessageListenerContainer listenerContainer = new DefaultMessageListenerContainer();
    listenerContainer.setConnectionFactory(connectionFactory);
    listenerContainer.setDestinationName(destinationName);
    listenerContainer.setMessageListener(messageListener);
    listenerContainer.setTransactionManager(txManager);
    listenerContainer.setSessionTransacted(true); 

    listenerContainer.setConcurrentConsumers(1);
    listenerContainer.setReceiveTimeout(3000);
    return listenerContainer;
}

}

和

@Configuration
@DependsOn("transactionManager")
@EnableJpaRepositories(basePackages = {"com.myapp.common"}, 
    entityManagerFactoryRef = "myEntityManager", transactionManagerRef = "transactionManager")
@ConfigurationProperties("myapp.ds")
@Validated
public class MyDataSourceConfiguration {

@Bean
public DataSource dataSource() throws SQLException {
    OracleXADataSource mysqlXaDataSource = new OracleXADataSource();
    mysqlXaDataSource.setURL(url);
    mysqlXaDataSource.setPassword(password);
    mysqlXaDataSource.setUser(username);

    AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
    xaDataSource.setXaDataSource(mysqlXaDataSource);
    xaDataSource.setUniqueResourceName("xads");
    xaDataSource.setPoolSize(10);
    xaDataSource.setMaxPoolSize(70);

    xaDataSource.init();
    return xaDataSource;
}

@Bean(name = "myEntityManager")
@DependsOn("transactionManager")
public LocalContainerEntityManagerFactoryBean customerEntityManager() throws Throwable {

    HashMap<String, Object> properties = new HashMap<String, Object>();     
    properties.put("javax.persistence.transactionType",     "JTA");
    properties.put("hibernate.archive.autodetection",       "hbm");
    properties.put("hibernate.dialect",                     "org.hibernate.dialect.Oracle10gDialect");
    properties.put("hibernate.transaction.jta.platform",    AtomikosJtaPlatform.class.getName());
    properties.put("hibernate.generate_statistics",         "false");
    properties.put("hibernate.jdbc_fetch_size",             "2");
    properties.put("hibernate.jdbc.batch_size",             "20");
    properties.put("hibernate.show_sql",                    "false");
    properties.put("hibernate.format_sql",                  "false");
    properties.put("hibernate.hbm2ddl.auto",                "none");
    properties.put("hibernate.id.new_generator_mappings",   "false"); 

    LocalContainerEntityManagerFactoryBean entityManager = new LocalContainerEntityManagerFactoryBean();
    entityManager.setJtaDataSource(dataSource());
    entityManager.setPackagesToScan("com.myapp.common");
    entityManager.setPersistenceUnitName("myPersistenceUnit");
    entityManager.setJpaPropertyMap(properties);
    entityManager.setPersistenceProvider(new HibernatePersistenceProvider());
    return entityManager;
}

}

怎么了？上面提到的错误是：ExceptionApperstandardImpl:hh000346:托管刷新期间出错[会话/实体管理器关闭]

Answer 1

我会尽量确保datasource connectionFactory在事务管理器之前初始化，然后关闭。这本身就是一种进步，不管它是否解决了问题。