2

我使用 ActiveMq 作为 JMS 服务器，使用 Atomikos 作为事务管理器。

在 ActiveMq Admin Web 界面上，我看到一条消息已入队，但有 2 条（！）消息已出队。

但是jms消费者只处理一次消息，没有重复处理。当我使用简单的 Spring JmsTransactionManager 时，有一个入队消息和一个出队消息。该问题仅出现在 Atomikos JTA 事务管理器中。

问题是什么？如何配置 Atomikos 看不到两次出队的消息？

我的配置如下。它几乎与教程中的相同。顺便说一句，Atomikos 的 spring 集成示例运行良好，但它使用 Spring 2.0，而我使用 Spring 3.1。

<bean id="activeMqXaConnectionFactory" class="org.apache.activemq.spring.ActiveMQXAConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616"/>
</bean>

<bean id="atomikosQueueConnectionFactory" class="com.atomikos.jms.QueueConnectionFactoryBean" init-method="init">
    <property name="resourceName" value="xaMq"/>
    <property name="xaQueueConnectionFactory" ref="activeMqXaConnectionFactory"/>
</bean>

<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <property name="targetConnectionFactory" ref="atomikosQueueConnectionFactory"/>
</bean>

<bean id="jmsDefaultContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="singleConnectionFactory"/>
    <property name="messageListener" ref="consumer"/>
    <property name="concurrentConsumers" value="1"/>
    <property name="destinationName" value="q.jtaxa"/>
    <property name="receiveTimeout" value="3000"/>
    <property name="recoveryInterval" value="5000"/>
    <property name="transactionManager" ref="transactionManager"/>
    <property name="sessionTransacted" value="true"/>
</bean>

<!-- Transactions -->
<!--Construct Atomikos UserTransactionManager, needed to configure Spring-->
<bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init"
      destroy-method="close">
    <!--  when close is called, should we force transactions to terminate or not? -->
    <property name="forceShutdown" value="true"/>
</bean>

<!-- Also use Atomikos UserTransactionImp, needed to configure Spring  -->
<bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
    <property name="transactionTimeout" value="300"/>
</bean>

<!-- Configure the Spring framework to use JTA transactions from Atomikos -->
<bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
    <property name="transactionManager" ref="atomikosTransactionManager"/>
    <property name="userTransaction" ref="atomikosUserTransaction"/>
    <property name="nestedTransactionAllowed" value="true"/>
</bean>

消费类是：

@Component
public class Consumer implements MessageListener {
  @Override
  public void onMessage(Message message) {
    if (message instanceof TextMessage) {
        try {
            TextMessage textMsg = (TextMessage) message;
            System.out.println("         " + textMsg.getText());
        } catch (JMSException ex) {
            ex.printStackTrace();
        }
    }
  }
}
4

1 回答 1

1

我发现什么配置错误。它是“atomikosQueueConnectionFactory”。我确实按照教程进行了操作，但它应该只是 AtomikosConnectionFactoryBean 类而不是 QueueConnectionFactoryBean。我删除了 atomikosQueueConnectionFactory 并添加了 atomikosConnectionFactory

<bean id="atomikosConnectionFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean" init-method="init">
    <property name="uniqueResourceName" value="amq1"/>
    <property name="xaConnectionFactory" ref="mqXaConnectionFactory"/>
</bean>

之后它工作正常。我在这里找到了正确的配置。

于 2012-03-01T12:32:29.133 回答