如何避免Mule4中从AMQ队列消费的消息回流至原队列？

我使用以下Mule 4 XML流程从backout队列获取消息，流程执行显示成功，但消息始终回流并停留在队列中。是不是缺少了消费确认相关的设置？

<flow name="message-reprocessing-reports-subflow" doc:id="f3ffe923-ee3e-4e0d-8a29-ec874b9755d7" >
    <scheduler doc:name="Scheduler" doc:id="ff715c36-be3e-4f79-bdc9-db7934c75985" >
        <scheduling-strategy >
            <fixed-frequency timeUnit="HOURS" frequency="1"/>
        </scheduling-strategy>
    </scheduler>
    <logger level="INFO" doc:name="Logger" doc:id="66ef28f0-c4a6-4bec-b824-a51ce7e8ea79" message="@@@@ ---> scheduler triggered"/>
    <jms:consume doc:name="Consume" doc:id="e63155ec-7563-4409-bd83-c66e6e1c792a" config-ref="AMQ-DocuSign-Connector" destination="#[Mule::p('amq.reports.bqueue')]" maximumWaitUnit="SECONDS" maximumWait="10">
        <jms:consumer-type>
            <jms:queue-consumer />
        </jms:consumer-type>
    </jms:consume>
    <logger level="INFO" doc:name="Logger" doc:id="44cb4e95-bbe7-4d00-beef-4ba36340fda2" message="#[payload]" />
    <ee:transform doc:name="Transform Message" doc:id="5a51e211-93b4-4408-a5c2-a0157e25c26b">
        <ee:message>
            <ee:set-payload><![CDATA[%dw 2.0
output application/java
---
payload]]></ee:set-payload>
        </ee:message>
    </ee:transform>
    <choice doc:name="Choice" doc:id="ea7cbccb-6b47-456b-a2cb-c67561b53917">
        <when expression="#[!isEmpty(payload)]">
            <logger level="INFO" doc:name="Logger" doc:id="b8c49215-383f-45e3-aaf3-2b9511a53828" message="#[&quot;Moving messages from DLQ&quot; ++ Mule::p('amq.docusign.reports.bqueue')  as String ++   &quot;to&quot; ++ Mule::p('amq.docusign.reports.topic') as String]" />
            <async doc:name="Async" doc:id="5ffa9543-8171-4a3e-ad9e-5a4480feaa55" >
                <jms:publish doc:name="Publish" doc:id="44425d09-4ced-4a9d-a1c8-9c6259a2b6c9" destination="${amq.reports.topic}" config-ref="AMQ-Connector" destinationType="TOPIC" />
            </async>            
        </when>
        <otherwise>
            <logger level="INFO" doc:name="Logger" doc:id="eb5f2d82-f0a3-4a8d-93f0-25ac4784e0fc" message="#[&quot;No message found in DLQ&quot; ++  Mule::p('amq.docusign.reports.bqueue')  as String ++  &quot;!&quot;]" />
        </otherwise>
    </choice>
    <error-handler >
        <on-error-continue enableNotifications="true" logException="true" doc:name="On Error Continue" doc:id="6443cb12-3b4e-4217-b559-f4ed22d2d27f" >
            <logger level="INFO" doc:name="Logger" doc:id="db46b10a-2203-4acb-b65d-07cd674b3373" message="no message found in Dead letter Queue"/>
        </on-error-continue>
    </error-handler>
</flow>

核心问题出在JMS消息确认时机和异步处理的冲突上：

1. 默认情况下，Mule的JMS消费组件会在同步流程执行完成后自动确认消息，但你把消息发布放在async作用域中，同步流程会先结束并尝试确认，此时异步的发布操作可能还未完成，MQ broker会认为消息未被正确处理，导致消息回流。
2. 未显式指定消息确认模式，部分MQ broker会要求显式确认才能彻底删除队列中的消息。

修复步骤 #

• 给jms:consume组件显式设置ackMode="MANUAL"，切换为手动确认模式
• 在异步发布操作成功完成后，调用jms:acknowledge组件手动确认消息
• 确保所有流程分支（包括无消息、异常场景）都有对应的确认逻辑，避免消息长期处于未确认状态

修改后的示例代码 #

<flow name="message-reprocessing-reports-subflow" doc:id="f3ffe923-ee3e-4e0d-8a29-ec874b9755d7" >
    <scheduler doc:name="Scheduler" doc:id="ff715c36-be3e-4f79-bdc9-db7934c75985" >
        <scheduling-strategy >
            <fixed-frequency timeUnit="HOURS" frequency="1"/>
        </scheduling-strategy>
    </scheduler>
    <logger level="INFO" doc:name="Logger" doc:id="66ef28f0-c4a6-4bec-b824-a51ce7e8ea79" message="@@@@ ---> scheduler triggered"/>
    <!-- 开启手动确认模式 -->
    <jms:consume doc:name="Consume" doc:id="e63155ec-7563-4409-bd83-c66e6e1c792a" 
        config-ref="AMQ-DocuSign-Connector" 
        destination="#[Mule::p('amq.reports.bqueue')]" 
        maximumWaitUnit="SECONDS" 
        maximumWait="10"
        ackMode="MANUAL">
        <jms:consumer-type>
            <jms:queue-consumer />
        </jms:consumer-type>
    </jms:consume>
    <logger level="INFO" doc:name="Logger" doc:id="44cb4e95-bbe7-4d00-beef-4ba36340fda2" message="#[payload]" />
    <ee:transform doc:name="Transform Message" doc:id="5a51e211-93b4-4408-a5c2-a0157e25c26b">
        <ee:message>
            <ee:set-payload><![CDATA[%dw 2.0
output application/java
---
payload]]></ee:set-payload>
        </ee:message>
    </ee:transform>
    <choice doc:name="Choice" doc:id="ea7cbccb-6b47-456b-a2cb-c67561b53917">
        <when expression="#[!isEmpty(payload)]">
            <logger level="INFO" doc:name="Logger" doc:id="b8c49215-383f-45e3-aaf3-2b9511a53828" 
                message="#[&quot;Moving messages from DLQ&quot; ++ Mule::p('amq.docusign.reports.bqueue')  as String ++   &quot;to&quot; ++ Mule::p('amq.docusign.reports.topic') as String]" />
            <async doc:name="Async" doc:id="5ffa9543-8171-4a3e-ad9e-5a4480feaa55" >
                <jms:publish doc:name="Publish" doc:id="44425d09-4ced-4a9d-a1c8-9c6259a2b6c9" 
                    destination="${amq.reports.topic}" 
                    config-ref="AMQ-Connector" 
                    destinationType="TOPIC" />
                <!-- 发布成功后手动确认消息 -->
                <jms:acknowledge doc:name="Acknowledge" doc:id="abc123"/>
            </async>            
        </when>
        <otherwise>
            <logger level="INFO" doc:name="Logger" doc:id="eb5f2d82-f0a3-4a8d-93f0-25ac4784e0fc" 
                message="#[&quot;No message found in DLQ&quot; ++  Mule::p('amq.docusign.reports.bqueue')  as String ++  &quot;!&quot;]" />
            <!-- 无消息场景下确认 -->
            <jms:acknowledge doc:name="Acknowledge Empty" doc:id="def456"/>
        </otherwise>
    </choice>
    <error-handler >
        <on-error-continue enableNotifications="true" logException="true" doc:name="On Error Continue" doc:id="6443cb12-3b4e-4217-b559-f4ed22d2d27f" >
            <logger level="INFO" doc:name="Logger" doc:id="db46b10a-2203-4acb-b65d-07cd674b3373" message="no message found in Dead letter Queue"/>
            <!-- 异常场景可根据需求选择是否确认，比如不确认让消息回流 -->
            <!-- <jms:acknowledge doc:name="Acknowledge on Error" doc:id="ghi789"/> -->
        </on-error-continue>
    </error-handler>
</flow>

额外说明 #

如果不需要异步执行消息发布，可以直接移除async作用域，同步流程完成后Mule会自动确认消息（默认AUTO模式），也能解决消息回流问题。

内容的提问来源于stack exchange，提问作者kushma gonna