0

我正在使用 spring 集成（带有入站/出站通道适配器）从

  • JMS 到 AMQP
  • AMQP 到 JMS
  • AMQP 到 AMQP

这工作正常。

当目标队列已满时，我想停止该过程，并将消息发送回原始队列（从通道回滚消息）。

这适用于 JMS => AMQP 和 AMQP => JMS，但适用于 AMQP => AMQP。

以下代码工作正常（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

0

我的 bean ID 有错误。

它适用于所有情况:-)

于 2020-11-20T16:27:58.123 回答