我需要将消息保存在rabbit mq中，并在作业成功完成后将其删除

Question

我需要在rabbit mq中保存m条消息。我在SimpleMessageListenerContainer中将acknowledgeMode用作手册。这有助于我将值存储在unacked in rabbit mq中。但即使在作业完成后，这些消息仍然未经确认。我需要的邮件被删除后，作业成功完成。请帮我找到解决办法

<beans:bean id="PartitionHandler" class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler" init-method="afterPropertiesSet" scope="job">
        <beans:property name="messagingOperations" ref="messagingTemplate"></beans:property>
        <beans:property name="stepName" value="slave" />
        <beans:property name="gridSize" value="${spring.gridsize}" />
        <beans:property name="pollInterval" value="5000"></beans:property>
        <beans:property name="jobExplorer" ref="jobExplorer"></beans:property>
        <beans:property name="replyChannel" ref="outboundReplies"></beans:property>
    </beans:bean>

    <beans:bean id="PeriodicTrigger" class="org.springframework.scheduling.support.PeriodicTrigger">
        <beans:constructor-arg value="5000"></beans:constructor-arg>
    </beans:bean> 


<beans:bean id="requestQueue" class="org.springframework.amqp.core.Queue">
    <beans:constructor-arg name="name" value="testQueue">
    </beans:constructor-arg>
    <beans:constructor-arg name="durable" value="true">
    </beans:constructor-arg> 
</beans:bean>

<int:poller id="PollerMetadata" default="true" trigger="PeriodicTrigger" task-executor="taskExecutor"></int:poller>  

<beans:bean id="amqptemplate" 
    class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <beans:property name="connectionFactory" ref="rabbitConnFactory" />
    <beans:property name="routingKey" value="testQueue"/>
    <beans:property name="queue" value="testQueue"/>
</beans:bean>

<beans:bean id="amqpOutboundEndpoint" class="org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint">
    <beans:constructor-arg ref="amqptemplate"/>
    <beans:property name="expectReply" value="false"></beans:property>
    <beans:property name="routingKey" value="testQueue"></beans:property>
    <beans:property name="outputChannel" ref="inboundRequests"></beans:property>
</beans:bean>

<int:service-activator ref="amqpOutboundEndpoint" input-channel="outboundRequests"/>

<beans:bean id="SimpleMessageListenerContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <beans:constructor-arg ref="rabbitConnFactory"/>
    <beans:property name="queueNames" value="testQueue"></beans:property>
    <beans:property name="autoStartup" value="false"></beans:property> 
    <beans:property name="acknowledgeMode" value="MANUAL"></beans:property>
    <beans:property name="concurrentConsumers" value="5"></beans:property>
</beans:bean>



<beans:bean id="AmqpInboundChannelAdapter" class="org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter"  init-method="afterPropertiesSet">
    <beans:constructor-arg ref="SimpleMessageListenerContainer"/>
    <beans:property name="outputChannel" ref="inboundRequests"></beans:property>
</beans:bean>

<beans:bean id="StepExecutionRequestHandler" class="org.springframework.batch.integration.partition.StepExecutionRequestHandler">
    <beans:property name="jobExplorer" ref="jobExplorer"/>
    <beans:property name="stepLocator" ref="stepLocator"/>
</beans:bean>

<int:service-activator ref="StepExecutionRequestHandler" input-channel="inboundRequests" output-channel="outboundStaging"/>


<bean id="rabbitConnFactory" 
    class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
    <constructor-arg><value>localhost</value></constructor-arg>
    <property name="username" value="guest" />
    <property name="password" value="guest" />
    <property name="virtualHost" value="/" />
    <property name="port" value="5672" />
</bean>


<bean id="admin" class="org.springframework.amqp.rabbit.core.RabbitAdmin">
    <constructor-arg ref="rabbitConnFactory" />
</bean>

<bean id="messagingTemplate"
    class="org.springframework.integration.core.MessagingTemplate">
 <constructor-arg ref="outboundRequests" />
  <property name="receiveTimeout" value="60000000"/>
</bean>

<bean id="outboundRequests" class="org.springframework.integration.channel.DirectChannel" >
<property name="maxSubscribers" value="5"></property>
</bean>


<int:channel id="outboundReplies" scope="job"><int:queue/></int:channel>

<bean id="outboundStaging" class="org.springframework.integration.channel.NullChannel"></bean>

<bean id="inboundRequests" class="org.springframework.integration.channel.QueueChannel"></bean>

<bean id="stepLocator" class="org.springframework.batch.integration.partition.BeanFactoryStepLocator"/>

Answer 1

使用手动确认时，您负责确认。

看我对这个问题的回答。