When using RabbitMQ as a java work queue, how should you handle concurrency and transient errors?

2
votes

I'm considering setting up a RabbitMQ broker to handle basic task processing for a java web application. The basic idea is that the producer is a web server that wants to be speedy about processing requests, but also needs to do some data processing. To accomplish this, it knows about some Job DTOs that can be serialized to AMQP messages, and somewhere out there is a consumer that knows how to process these jobs. Classic.

After reading the RabbitMQ documentation, I'm still left with a couple of questions.

  1. I want the consumer application to fully utilize its CPU for processing messages, and I'm wondering if RabbitMQ already provides the worker pool. Assuming I wanted 4 worker threads, is it enough to register 4 channels with 1 consumer each on the same connection? In that case, work would be done in handleDelivery() and an ack would be sent if it successfully completes. Or rather, should I use 1 consumer and manage a pool of workers at my own layer of the application?

  2. Consumers will be talking to the database, which means transient errors will occur; deadlocks, optimistic locking violations, database server restarts, and so on. What is the intended behavior if a consumer can't process a job? Issue a nack and wait for the broker to resend the job, or delay the ack until it's certain that the job can't be completed? Or does it not matter?

It's worth noting that jobs generally won't take longer than a few seconds unless the database is restarting or something. I appreciate any guidance!

1
javamultithreadingrabbitmqamqp

1 Answers

2
votes

1a. If you want a fixed number of working threads then you should open several channels and register one consumer per channel. RabbitMQ has a single dispatcher thread per channel so that one consumer will block other consumers registered on the same channel.

1b. If you need a dynamically growing pool of workers then you might consider using spring-rabbit (assuming you implement in Java). Specifically, SimpleMessageListenerContainer manages a pool of consumers growing and shrinking dynamically according to the work load. Refer to section "Listeners Concurrency" in Spring AMQP reference guide.

2a. If you know the worker has failed you can nack and the message will be delivered to another consumer. The message will not be delivered to another consumer until the consumer acknowledged it unless it's "auto ack" mode.