2

ActiveMqをJMSサーバーとして使用し、Atomikosをトランザクションマネージャーとして使用しています。

ActiveMq Admin Webインターフェイスで、1つのメッセージがキューに入れられたのに、2つ（！）のメッセージがデキューされたことがわかります。

ただし、jmsコンシューマーはメッセージを1回だけ処理するため、処理に重複はありません。単純なSpringJmsTransactionManagerを使用すると、1つのキューに入れられたメッセージと1つのデキューされたメッセージがあります。この問題は、AtomikosJTAトランザクションマネージャーでのみ発生します。

何が問題ですか？2回デキューされたメッセージが表示されないようにAtomikosを構成するにはどうすればよいですか？

私の設定は以下の通りです。チュートリアルとほぼ同じです。ところで、AtomikosのSpring統合の例はうまく機能しますが、Spring 2.0を使用しているのに対し、私はSpring3.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」でした。チュートリアル通りにやったのですが、QueueConnectionFactoryBean ではなく、AtomikosConnectionFactoryBean クラスのはずです。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 に答える