0

メッセージを転送するために春の統合（インバウンド/アウトバウンドチャネルアダプターを使用）を使用しています

  • JMS から AMQP
  • AMQP から JMS
  • AMQP から AMQP

これはうまくいきます。

宛先キューがいっぱいになったら、プロセスを停止し、メッセージを元のキューに送り返したいと思います (チャネルからのロールバック メッセージ)。

これは、JMS => AMQP および AMQP => JMSでは正常に機能しますが、AMQP => AMQP では KO になります。

次のコードは正常に動作します (JMS => AMQP)

<bean id="myListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer" >
    <property name="autoStartup" value="false" />
    <property name="connectionFactory" ref="connectionFactoryCaching" />
    <property name="destination" ref="jmsQueue" />
    <property name="maxMessagesPerTask" value="1" />
    <property name="receiveTimeout" value="1" />
    <property name="backOff" ref="fixedBackOff" />
    <property name="sessionTransacted" value="true"/>
</bean>
<int-jms:message-driven-channel-adapter id="jmsIn" container="myListener" channel="channelJMS_AMQP" error-channel="processChannel1"/>


<rabbit:template    id="rabbitTemplate" 
    connection-factory="rabbitConnectionFactory"  
    mandatory="true" 
    channel-transacted="true" 
    message-converter="simpleMessageConverter"/>

<int-amqp:outbound-channel-adapter  channel="channelJMS_AMQP" 
    routing-key="RK1" 
    exchange-name="EXC1" 
    amqp-template="rabbitTemplate" 
    default-delivery-mode="PERSISTENT"/>

ログ ファイルでは、メッセージはよくロールバックされます。

2020-11-19 10:08:57.860 [AMQP Connection XX.XX.XX.XX.130:5672] ERROR o.s.a.r.c.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - partial tx completion, class-id=90, method-id=20)
2020-11-19 10:08:57.861 [myListener-1] DEBUG o.s.a.r.c.CachingConnectionFactory - Detected closed channel on exception.  Re-initializing: AMQChannel(amqp://guest@XX.XX.XX.XX.130:5672/,1)
2020-11-19 10:08:57.950 [myListener-1] DEBUG o.s.a.r.c.RabbitResourceHolder - Rolling back messages to channel: Cached Rabbit Channel: AMQChannel(amqp://guest@XX.XX.XX.XX.130:5672/,1), conn: Proxy@68935e42 Dedicated Rabbit Connection: SimpleConnection@577ca937 [delegate=amqp://guest@XX.XX.XX.XX.130:5672/, localPort= 41314]

正常に動作する AMQP=> JMS の同様のコードがあります。

しかし、AMQP => AMQP の場合、問題があり、メッセージが失われ、プロセスが停止しません。

<bean id="myListener" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <property name="connectionFactory" ref="rabbitConnectionFactory" />
    <property name="queueNames" value="MY_QUEUE" />
    <property name="recoveryBackOff" ref="fixedBackOffRabbitMQ"/>
    <property name="channelTransacted" value="true"></property>
</bean>
    
<int-amqp:inbound-channel-adapter   channel="channelAMQP_AMQP" 
    id="inboundChannelAdapter" 
    auto-startup="true" listener-container="myListener" error-channel="processChannel1" />  
    
<rabbit:template    id="rabbitTemplate" 
    connection-factory="rabbitConnectionFactory2"  
    mandatory="true" 
    channel-transacted="true" 
    message-converter="simpleMessageConverter"/>

<int-amqp:outbound-channel-adapter  channel="channelAMQP_AMQP" 
    routing-key="RK1" 
    exchange-name="EXC1" 
    amqp-template="rabbitTemplate" 
    default-delivery-mode="PERSISTENT"/>

ログファイル:

2020-11-19 10:15:24.260 [AMQP Connection XX.XX.XX.XX.130:5672] ERROR o.s.a.r.c.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - partial tx completion, class-id=90, method-id=20)
2020-11-19 10:15:24.351 [myListener-1] DEBUG o.s.a.r.c.CachingConnectionFactory - Detected closed channel on exception.  Re-initializing: AMQChannel(amqp://guest@XX.XX.XX.XX.130:5672/,1)
2020-11-19 10:15:24.356 [pool-4-thread-3] DEBUG o.s.a.r.l.BlockingQueueConsumer - Received shutdown signal for consumer tag=amq.ctag--o5d3WkrF0N2aiCHul-qiA

そして、メッセージを停止してロールバックする代わりに、コンシューマーを再起動します。

その後、私のメッセージは失われます:- (

この特定のケースで期待される操作を行うための推奨事項はありますか? 特定の治療法を追加する必要がありますか?

4

1 に答える 1