2

Atomikos + Camel + ActiveMQ の組み合わせを使用しているときに、誰かがこの問題に遭遇したかどうか疑問に思っています。このコンボを使用して、トランザクション方式でキューからメッセージを剥がします。うまくいきそうです。問題は、ActiveMQ でアドバイザリ メッセージをオンにする必要がある状況にあることです。それを行った後、すべてのキューで常に接続が再作成されていることに気付きました。これは、ActiveMQ.Advisory.Consumer.Queue トピックのフラッディングによって証明されます。これは、接続の作成、トランザクションのオープン、コミット、および接続のクローズを継続的に行うため、DEBUG ロギングでも明らかです。これは、実際のアプリケーション生成メッセージなしで発生します。他のすべての非トランザクション キュー/トピックには、この問題はありません。他のいくつかの投稿で、接続プーリングとキャッシュがこの問題を軽減できることを読みました。キャッシュを使用するべきではなく、既に接続プールを使用しているようです。私はこの設定を使用しています：

<bean id="txq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="txJmsConfig" />
</bean>
<bean id="txJmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="atomikosConnectionFactory" />
    <property name="concurrentConsumers" value="1" />
    <property name="transacted" value="true" />
    <property name="maxConcurrentConsumers" value="${consumers.concurrent.max}" />
    <property name="transactionManager" ref="jtaTransactionManager" />
    <property name="cacheLevelName" value="CACHE_NONE" />
</bean>
<bean id="atomikosConnectionFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean"
    init-method="init" destroy-method="close">
    <property name="uniqueResourceName">
        <value>XA-JMS-ATOMIKOS</value>
    </property>
    <property name="localTransactionMode">
        <value>false</value>
    </property>
    <property name="poolSize">
        <value>4</value>
    </property>
    <property name="xaConnectionFactory">
        <ref bean="xaJmsConnectionFactory" />
    </property>
</bean>
<bean id="xaJmsConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
    <property name="brokerURL"
        value="${queue.address}?jms.watchTopicAdvisories=false&amp;jms.prefetchPolicy.all=0" />
</bean>
<bean id="userTransactionService" class="com.atomikos.icatch.config.UserTransactionServiceImp"
    init-method="init" destroy-method="shutdownForce">
    <constructor-arg>
        <props>
            <prop key="com.atomikos.icatch.service">
                com.atomikos.icatch.standalone.UserTransactionServiceFactory
            </prop>
            <prop key="com.atomikos.icatch.max_actives">${batch.transactions.concurrent.max}</prop>
        </props>
    </constructor-arg>
</bean>
<bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager"
    init-method="init" destroy-method="close" depends-on="userTransactionService">
    <property name="forceShutdown" value="false" />
</bean>
<bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
    <property name="transactionTimeout" value="300" />
</bean>
<bean id="jtaTransactionManager"
    class="org.springframework.transaction.jta.JtaTransactionManager">
    <property name="transactionManager" ref="atomikosTransactionManager" />
    <property name="userTransaction" ref="atomikosUserTransaction" />
</bean>

プーリングを実装したと思われる AtomikosConnectionFactoryBean を使用します。多分私はそれについて間違っていますか？他の誰かが私と一緒にこのボートに乗っているかどうか、そして彼らがそれを修正するために何をしたかを知りたい.

@PeterSmithが提案した実装

ペッター、あなたの提案に感謝します。XaPooledConnectionFactory を使用するように構成を変更しました。春はこれに満足していません。XaPooledConnectionFactory は XAConnectionFactory を実装していないと考えられます。

<bean id="atomikosConnectionFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean"
    init-method="init" destroy-method="close" depends-on="xaJmsPooledConnectionFactory">
    <property name="uniqueResourceName">
        <value>XA-JMS-ATOMIKOS</value>
    </property>
    <property name="localTransactionMode">
        <value>false</value>
    </property>
    <property name="maxPoolSize">
        <value>32</value>
    </property>
    <property name="xaConnectionFactory">
        <ref bean="xaJmsPooledConnectionFactory" />
    </property>
</bean>
<bean id="xaJmsPooledConnectionFactory" class="org.apache.activemq.pool.XaPooledConnectionFactory"
     init-method="start" destroy-method="stop" depends-on="xaJmsConnectionFactory">
    <property name="maxConnections" value="2" />
    <property name="connectionFactory" ref="xaJmsConnectionFactory" />
</bean> 
<bean id="xaJmsConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
    <property name="brokerURL"
        value="${queue.address}?jms.watchTopicAdvisories=false&amp;jms.prefetchPolicy.all=0" />
</bean>

java.lang.IllegalStateException: タイプ [org.apache.activemq.pool.XaPooledConnectionFactory] ​​の値をプロパティ 'xaConnectionFactory' の必要なタイプ [javax.jms.XAConnectionFactory] ​​に変換できません: 一致するエディターまたは変換戦略が見つかりません

ドキュメントには、XaPooledConnectionFactory クラスが javax.jms.XAConnectionFactory を実装すると記載されているため、この時点で少し迷っています。うまくいくはずです。

4

1 に答える 1

1

上記の組み合わせを使用して、これも見ました。

Spring DMLC (キャメル jms コンシューマーを実装している) は、consumer.receive() のループを使用して、XA トランザクションに読み取り操作を登録できるようです。

DMLC での XA キャッシングがなくても、ポーリングではない SMLC を使用することも可能です。

activemq CF を org.apache.activemq.pool.PooledConnectionFactory にラップしてみて、それが多少役立つかどうかを確認してください。

于 2013-01-23T08:11:27.573 に答える