Я пытаюсь интегрировать Bitronix Transaction Manager в мой загрузочный проект Spring для совместного управления транзакциями jdbc и jms.У меня есть две базы данных и один брокер ActiveMQ для JMS.Я подключил базы данных в одной транзакции, но когда я попытался включить JMS, похоже, это не сработало.
Это моя конфигурация Bitronix Transaction Manager:
@Configuration
@EnableTransactionManagement
public class BitronixJtaConfiguration {
private static final Logger log = LoggerFactory.getLogger(BitronixJtaConfiguration.class);
@Value("${bitronix.tm.serverId}")
private String serverId;
@Value("${bitronix.tm.journal.disk.logPart1Filename:}")
private String logPart1Filename;
@Value("${bitronix.tm.journal.disk.logPart2Filename:}")
private String logPart2Filename;
@Bean
public bitronix.tm.Configuration transactionManagerServices() {
bitronix.tm.Configuration configuration = TransactionManagerServices.getConfiguration();
configuration.setServerId(serverId);
if ("".equals(logPart1Filename) && "".equals(logPart2Filename)) {
configuration.setJournal(null);
log.info("Disable journal for testing.");
} else {
configuration.setLogPart1Filename(logPart1Filename);
configuration.setLogPart2Filename(logPart2Filename);
}
return configuration;
}
@Bean
public TransactionManager transactionManager() {
return TransactionManagerServices.getTransactionManager();
}
@Bean
public UserTransaction userTransaction() {
return TransactionManagerServices.getTransactionManager();
}
@Bean
public PlatformTransactionManager platformTransactionManager() {
UserTransaction userTransaction = userTransaction();
TransactionManager transactionManager = transactionManager();
return new JtaTransactionManager(userTransaction, transactionManager);
}
}
Это одна из моих конфигураций класса базы данных:
@Configuration
@EnableTransactionManagement
public class TransportationPlanDBConfig {
private static final Logger LOGGER = LoggerFactory.getLogger("ppalfile");
@Value("${tp.jdbc.driverClassName}")
private String driverClassName;
@Value("${tp.jdbc.username}")
private String username;
@Value("${tp.jdbc.url}")
private String url;
@Value("${tp.jdbc.password}")
private String password;
@Value("${tp.c3p0.max_size}")
private int c3p0MaxSize;
@Value("${tp.c3p0.min_size}")
private int c3p0MinSize;
@Value("${tp.c3p0.unreturned_connection_timeout}")
private int c3p0UnreturnedConnectionTimeout;
@Value("${tp.c3p0.acquire_increment}")
private int c3p0AcquireIncrement;
@Value("${tp.c3p0.max_idle_time}")
private int c3p0MaxIdleTime;
public TransportationPlanDBConfig() {
// Empty constructor
}
@Bean(name = "tpds", destroyMethod = "close")
public DataSource dataSource() {
LOGGER.debug("Creating Transportation plan DS");
PoolingDataSource poolingDataSource = new PoolingDataSource();
poolingDataSource.setClassName(driverClassName);
poolingDataSource.setUniqueName("tpds");
Properties props = new Properties();
props.put("url", url);
props.put("user", username);
props.put("password", password);
poolingDataSource.setDriverProperties(props);
poolingDataSource.setAllowLocalTransactions(true);
poolingDataSource.setMaxPoolSize(c3p0MaxSize);
poolingDataSource.init();
return poolingDataSource;
}
@Bean(name = "tpJdbcTemplate")
JdbcTemplate jdbcTemplate(@Qualifier("tpds") DataSource dataSource) {
LOGGER.debug("Creating JdbcTemplate transport plan");
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
LOGGER.debug(" JdbcTemplate Transport Plan created ");
return jdbcTemplate;
}
}
Мой класс конфигурации ActiveMQ:
@Configuration
@EnableTransactionManagement
public class ActivesMQsConfiguration {
@Bean
public ConnectionFactory jmsConnectionFactoryLocal() {
PoolingConnectionFactory btmPoolingConnectionFactory = new PoolingConnectionFactory();
btmPoolingConnectionFactory.setClassName("org.apache.activemq.ActiveMQXAConnectionFactory");
btmPoolingConnectionFactory.setUniqueName("AMQLocal");
btmPoolingConnectionFactory.setMinPoolSize(1);
btmPoolingConnectionFactory.setMaxPoolSize(5);
btmPoolingConnectionFactory.setAllowLocalTransactions(true);
btmPoolingConnectionFactory.setUser("admin");
btmPoolingConnectionFactory.setPassword("admin");
btmPoolingConnectionFactory.getDriverProperties().setProperty("brokerURL", "tcp://localhost:61616");
btmPoolingConnectionFactory.init();
return btmPoolingConnectionFactory;
}
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerFactoryLocal(
@Qualifier("jmsConnectionFactoryLocal") ConnectionFactory connectionFactory,
DefaultJmsListenerContainerFactoryConfigurer configurer) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setSessionTransacted(true);
configurer.configure(factory, connectionFactory);
return factory;
}
}
Моя реализация JMS Listener:
@Component
@Transactional
public class ContactTransactionReceiver {
private int mensajesConsumer2 = 0;
@Autowired
@Qualifier("versionJdbcTemplate")
private JdbcTemplate versionJdbcTemplate;
@Autowired
@Qualifier("tpJdbcTemplate")
private JdbcTemplate tpjdbcTemplate;
@Autowired
private VersionsConfDao versionsConfDao;
@Autowired
private TrainDao trainDao;
@Transactional(rollbackFor=Exception.class)
@JmsListener(destination = "Consumer.consumer2.VirtualTopic.TopicPrueba")
public void receiveMessageFromContacts2(Message message) throws Exception {
mensajesConsumer2++;
TextMessage txtMessage = (TextMessage) message;
System.out.println("Segundo consumer:" + txtMessage.getText() + " recibidos:" + mensajesConsumer2);
VersionsConf versionsconf = new VersionsConf("V" + mensajesConsumer2, "V" + mensajesConsumer2, false,new Timestamp(1L), 1);
VersionsConf versionsResult = versionsConfDao.insertUpdate(versionJdbcTemplate, versionsconf);
if (mensajesConsumer2 == 2) {
throw new Exception();
}
Train train = new Train("101"+mensajesConsumer2, 1L, 2L, false, true, "atp");
Train trainResult = trainDao.insertUpdate(tpjdbcTemplate, train);
if (mensajesConsumer2 == 3) {
throw new Exception();
}
}
}
На основе моей реализации слушателя, как я понял, функциональность Bitronix:
При первом входящем сообщении: необходимо вставитьодна строка в каждой базе данных и исключить сообщение из очереди.-> Это работает нормально.
На втором и третьем входящем сообщении: необходимо вставить 0 строк из-за исключения и сохранить сообщение в очереди.-> Строки не вставлены, но сообщение удалено.
Кроме того, я хотел бы добавить, что во время выполнения регистрируется следующее: [main] bitronix.tm.recovery.Recoverer: восстановление совершило 0 оборванных транзакций и откатило 0 прерванных транзакций на 4 ресурсах [AMQLocal, vds, AMQRemote, tpds]
Итак, я понял, что оба брокера и обе базы данныхзарегистрированы.Но когда слушатель обрабатывает второе сообщение (выдает исключение) и записывает в журнал:
WARN 5740 [Session Task-1] bitronix.tm.twopc.Preparer: выполнение транзакции с 0 зачисленным ресурсом
Любая идея о проблеме ??
Вы можете найти полный код на: https://github.com/PedroRamirezTOR/spring-jms-jdbc-integration.git
Спасибо!