Avoid consuming messages before being able to process them

[indrekj]

Previously we were taking all messages using Queue#subscribe and then using a thread pool to process them. It is more useful to keep the messages that cannot be processed yet in the queue. This allows adding new consumers that can start processing them.

Prefetch value tells rabbitmq how many messages can consumers on a channel be given at a time (before they have to acknowledge or reject one of the earlier received messages).

[take-five]

Does it mean, that all messages will be processed sequentially? Don't you think it can reduce throughput?

My understanding of using prefetch is that it can limit concurrency level of message processing, but you still have to use thread pool or something similar. Otherwise you have to spawn X consumers to achieve X level of concurrency, and I'm not sure that it's more efficient than having a thread pool.

[willmore]

This also allowed us to remove our own thread pools as now rabbitmq will take it care for us.

Which thread pools do you mean exactly?

[indrekj]

@take-five you are correct, thank you. Additional testing reveals it makes it sequential. Still need to have our own thread pool. 2am programming mistakes, oh well. Updating soon.

[indrekj]

Updated. Did a manual test as well. Created a responder that sleeps 1 second and has max_concurrency set to 2. Sent 6 messages. It processed 2 messages in parallel and took 3 seconds in total.

[indrekj]

Updated

[willmore]

!squash