java - AMQP 防止自动消息读取

Question

我正在使用 Spring AMQP 来监听消息(配置有监听器容器、服务激活器、链、桥接器和聚合器)。在应用程序启动时，AMQP 开始读取我们不想要的消息。我尝试了 auto-startup=false 但它不起作用。我错过了什么吗？

此外，如果它确实有效，那么我如何以编程方式再次启动它们？我尝试了listenerContainer.start();。聚合商和其他人呢？

编辑

以下是我的配置:

<rabbit:queue name="my_queue1" declared-by="consumerAdmin"/>
<rabbit:queue name="my_queue2" declared-by="consumerAdmin"/>
<rabbit:queue name="my_batch1" declared-by="consumerAdmin"/>

<int-amqp:channel id="myPollableChannel" message-driven="false" connection-factory="consumerConnFactory" queue-name="my_queue2"/>
<int-event:inbound-channel-adapter channel="myPollableChannel" auto-startup="false"/>

<int-amqp:channel id="myAggregateChannel" connection-factory="consumerConnFactory"/>
<int-event:inbound-channel-adapter channel="myAggregateChannel" auto-startup="false"/>

<int-amqp:channel id="myChannel" connection-factory="consumerConnFactory"/>
<int-event:inbound-channel-adapter channel="myChannel" auto-startup="false"/>

<int-amqp:channel id="myFailedChannel" connection-factory="consumerConnFactory"/>
<int-event:inbound-channel-adapter channel="myFailedChannel" auto-startup="false"/>

<rabbit:template id="genericTopicTemplateWithRetry" connection-factory="connectionFactory" exchange="my_exchange" retry-template="retryTemplate"/>

<rabbit:topic-exchange name="my_exchange" declared-by="consumerAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="my_queue1" pattern="pattern1"/>
            <rabbit:binding queue="my_queue2" pattern="pattern1"/>
        </rabbit:bindings>
</rabbit:topic-exchange>

<int:handler-retry-advice id="retryAdvice" max-attempts="5" recovery-channel="myFailedChannel">
    <int:exponential-back-off initial="3000" multiplier="5.0" maximum="300000"/>
</int:handler-retry-advice>

<int:bridge input-channel="myPollableChannel" output-channel="myAggregateChannel">
    <int:poller max-messages-per-poll="100" fixed-rate="5000"/>
</int:bridge>

<int:aggregator id="myBatchAggregator"
    ref="myAggregator" 
    correlation-strategy="myCorrelationStrategy" 
    release-strategy="myReleaseStrategy" 
    input-channel="myAggregateChannel" 
    output-channel="myChannel" 
    expire-groups-upon-completion="true"
    send-partial-result-on-expiry="true"
    group-timeout="1000" />

<int:chain input-channel="myFailedChannel">
    <int:transformer expression="'Failed to publish messages to my channel:' + payload.failedMessage.payload" />
    <int-stream:stderr-channel-adapter append-newline="true"/>
</int:chain>

<int:service-activator input-channel="myChannel" output-channel="nullChannel" ref="myWorker" method="myMethod">
    <int:request-handler-advice-chain><ref bean="retryAdvice" /></int:request-handler-advice-chain>
</int:service-activator>

<rabbit:listener-container connection-factory="consumerConnFactory" requeue-rejected="false" concurrency="1">
    <rabbit:listener ref="myListener" method="listen" queue-names="queues1" admin="consumerAdmin" />
</rabbit:listener-container>

Answer 1

好的。感谢您的配置!

不确定为什么您需要基于 AMQP 的 channel ，但您的主要问题正是在那里。

但请注意，<int-amqp:channel>有auto-startup="false"选项也是如此。

您将准备好从 AMQP 接收数据，您只需要 start()这些 channel 的id .