Spring AMQP Integration - Consumer Manual Acknowledgement

Question

I am testing `Spring-AMQP` with `Spring-Integration` support, I've following configuration and test: ```

public static void main(String[] args) { System.out.println("Starting consumer with integration.."); AbstractApplicationContext context = new ClassPathXmlApplicationContext( "classpath:META-INF/spring/integration/spring-integration-context-consumer.xml"); PollableChannel consumingChannel = context.getBean("consumingChannel", PollableChannel.class); int count = 0; while (true) { Message msg = consumingChannel.receive(1000); System.out.println((count++) + " -> " + msg); try { //sleep to check numbe…

Accepted Answer

The problem is because you are using async handoff using a `QueueChannel`. It is generally better to control the concurrency in the container (`concurrent-consumers="2"`) and don't do any async handoffs in your flow (use `DirectChannel`s). That way, AUTO ack will work just fine. Instead of receiving from the `PollableChannel` subscribe a `new MessageHandler()` to a `SubscribableChannel`.

Update:

You normally don't need to deal with Messages in an SI application, but the equivalent of your test with a DirectChannel would be...

```
    SubscribableChannel channel = context.getBean("fromRabbit", SubscribableChannel.class);

channel.subscribe(new MessageHandler() {

@Override
        public void handleMessage(Message<?> message) throws MessagingException {
            System.out.println("Got " + message);
        }
    });
```

Spring AMQP Integration - Consumer Manual Acknowledgement

Solution

Problem