Reliable publishing to RabbitMQ with Spring AMQP
RabbitMQ is surely a popular way for message communication between microservices. From what I’ve found, there are already many articles on reliability and how to implement retry mechanism with RabbitMQ consumers. But apart from the consumer’s side, there are still the problems of publishing and how to ensure all the messages arrive safely at the queue in the first place. So for today, I’m going to talk about the publisher’s side, how to ensure 99.99% of the time your messages will be published successfully.
First, I should note that RabbitMQ also has a detailed guide on this topic with some plain Java code examples. And this article is inspired by the concept there, but is aimed specifically at Spring AMQP users. I will show the example of how to implement the concepts with RabbitTemplate.
For the code example, we will start with a simple service class that use RabbitTemplate to send message to a queue.
@Service
public class RabbitPublisherService {
private final RabbitTemplate rabbitTemplate;
public void publishMessage(MyData myData) {
rabbitTemplate.convertAndSend(Constant.MY_EXCHANGE, Constant.MY_ROUTING_KEY,
myData);
}
}This code will publish the message fine most of the time, but it still does not account for failures at all. Generally there are 3 failure cases that can happen at MQ publisher that we should handle:
- Connection error when trying to publish message (like when RabbitMQ is down)
- Get negative ack in confirm callback (when queue/exchange is missing)
- Does not get any response from confirm callback (some packet loss from network problem)
So let’s tackle them one by one.
#1 Connection error
For the first case, RabbitTemplate already provide a way to set retryTemplate with retry policy (using Spring Retry) to help with handling problems with broker connectivity:
@Bean
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
final var rabbitTemplate = new RabbitTemplate(connectionFactory);
RetryTemplate retryTemplate = new RetryTemplate();
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
backOffPolicy.setInitialInterval(500);
backOffPolicy.setMultiplier(3.0);
backOffPolicy.setMaxInterval(3000_000);
retryTemplate.setBackOffPolicy(backOffPolicy);
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
simpleRetryPolicy.setMaxAttempts(15);
retryTemplate.setRetryPolicy(simpleRetryPolicy);
rabbitTemplate.setRetryTemplate(retryTemplate);
return rabbitTemplate;
}You can set the retry backoff policy as per your requirement. In the example above, it is set with exponential backoff policy and 15 max attempts.
#2 Negative Ack
You handle negative ack by setting publisher confirm type to correlated in application.yml :
rabbitmq:
host: localhost
port: 5672
publisher-confirm-type: correlatedPublisher confirms are a RabbitMQ extension to implement reliable publishing. When publisher confirms are enabled on a channel, messages the client publishes are confirmed asynchronously by the broker, meaning they have been taken care of on the server side.
publisher-confirm-type: correlated is used with CorrelationData object to correlate confirmations with sent messages. With these correlated informations, we can write a code to handle which message will need to be re-published with ease.
Get Radit P.’s stories in your inbox
Join Medium for free to get updates from this writer.
Then we add the codes to ourRabbitPublisherService class to handle ack and negative ack asynchronously in confirm callback:
@Service
public class RabbitPublisherService {
private final RabbitTemplate rabbitTemplate;
protected ConcurrentHashMap<String, MyData> outstandingConfirms = new ConcurrentHashMap<>();
protected ConcurrentLinkedQueue<RetryCorrelationData> negativeAckedMessages = new ConcurrentLinkedQueue<>();
@PostConstruct
public void postConstruct() {
rabbitTemplate.setConfirmCallback((correlation, ack, reason) -> {
if (correlation == null) {
return;
}
RetryCorrelationData retryCorrelationData = (RetryCorrelationData) correlation;
handleRabbitAcknowledgement(ack, retryCorrelationData);
});
}
void handleRabbitAcknowledgement(boolean isAck, RetryCorrelationData retryCorrelationData) {
final String id = retryCorrelationData.getId();
if (!isAck) {
log.info("id {} got nack-ed, save message to nack-ed message queue for retry", id);
negativeAckedMessages.add(new RetryCorrelationData(id,
retryCorrelationData.getRetryCount() + 1));
} else {
// acked, message reached the broker successsfully
cleanOutstandingConfirm(id);
log.info("id {} got acked", id);
}
}
public void publishMessage(MyData myData, int retryCount) {
final String id = myData.getId();
RetryCorrelationData correlationData = new RetryCorrelationData(id, retryCount);
outstandingConfirms.putIfAbsent(id, myData);
rabbitTemplate.convertAndSend(Constant.MY_EXCHANGE, Constant.MY_ROUTING_KEY,
myData, correlationData);
}
public MyData cleanOutstandingConfirm(String id) {
// remove the data from outstandingConfirms
// created as public method for usage in other retry schedulers as well.
return outstandingConfirms.remove(id);
}In the code above, we modified publishMessage method to send correlationData, which is the data to correlate publisher confirms. The outstandingConfirms is a hash map to keep the pending messages in temporary cache for retry mechanism in the case that we don’t get ack from MQ. Each of our message have a unique id that can help correlate back to the content in outstandingConfirms.
rabbitTemplate.setConfirmCallback is the part that will handle the confirm callback.
handleRabbitAcknowledgement will check if it is negative acked message, if yes then the message will be added into negativeAckedMessages which is a another temporary data collection (specially for nack-ed case) that keeps them in a queue for retrieving at interval. If it is positive acked (confirmed), we will just clear that data from outstandingConfirms hash map.
Some readers might have a question here: why write another data collection to keep the messages in memory? why don’t just re-publish the nack-ed messages directly from the callback? RabbitMQ’s guide already provide a good answer:
It can be tempting to re-publish a nack-ed message from the corresponding callback but this should be avoided, as confirm callbacks are dispatched in an I/O thread where channels are not supposed to do operations. A better solution consists in enqueuing the message in an in-memory queue which is polled by a publishing thread. A class like ConcurrentLinkedQueue would be a good candidate to transmit messages between the confirm callbacks and a publishing thread.
We extends CorrelationData class to keep the retry count as well. We can check the retryCount to ensure the message stop retrying when reach some defined max attempts:
public class RetryCorrelationData extends CorrelationData {
private final int retryCount;
public RetryCorrelationData(String id, int retryCount) {
super(id);
this.retryCount = retryCount;
}
public int getRetryCount() {
return retryCount;
}
}Then we have a Spring Scheduled task scheduleNackedRepublishTask that will read the nacked messages from negativeAckedMessages queue at interval and try to republish them again.
@Scheduled(fixedDelay = 10000)
public void scheduleNackedRepublishTask() {
Queue<RetryCorrelationData> queue = rabbitPublisherService.getNegativeAckedMessages();
for (int i = 0; i < appConfig.getNackedRepublishBatchSize(); i++) {
if (queue.isEmpty()) {
break;
}
RetryCorrelationData retryCorrelationData = queue.remove();
log.info("Retry nack-ed message : {}", retryCorrelationData.getId());
checkRetryCountAndRepublish(retryCorrelationData.getRetryCount(), retryCorrelationData.getId());
}
}
void checkRetryCountAndRepublish(int retryCount, String id) {
MyData data = rabbitPublisherService.cleanOutstandingConfirm(id);
if (data == null) {
log.warn("Fail to retrieve data from outstandingConfirms queue with id : {}", id);
return;
}
if (retryCount < appConfig.getMaxPublisherRetryLimit()) {
rabbitPublisherService.publishMessage(data, retryCount + 1);
} else {
log.warn("Retry counter reach limit, will stop retry for this message");
}
}Notice that we check the retry count as well, to make sure there’s a limit to the retry mechanism and we’re not stuck with infinite retry.
#3 Unconfirmed messages
For messages that never receive any confirm callback at all, we handle it with a timeout (like 10 seconds) in another Scheduled task:
@Scheduled(fixedDelay = 5000)
public void scheduleUnconfirmedRepublishTask() {
Collection<CorrelationData> unconfirmed = rabbitTemplate.getUnconfirmed(appConfig.getUnconfirmedTimeout());
if (unconfirmed != null) {
for (CorrelationData correlationData : unconfirmed) {
RetryCorrelationData retryCorrelationData = (RetryCorrelationData) correlationData;
log.info("Retry unconfirmed message id : {}", retryCorrelationData.getId());
checkRetryCountAndRepublish(retryCorrelationData.getRetryCount(), correlationData.getId());
}
}
}Luckily, rabbitTemplate provides getUnconfirmed method that returned all the unconfirmed data that surpass some defined age (timeout). So we use it to republish the messages at interval as well.
And please beware that since we cannot ensure that the message reached the broker on the first time or not, we need to ensure that the consumer is able to support handling duplicate messages in the case that we publish the same message again by our retry strategy.
Conclusion
The complete source code of this article’s examples are available at https://github.com/raditpan/rabbitmq-publisher-demo.
Please note that these are just examples of implementing reliable publishing with Spring AMQP. I hope you get an idea on how to do it in the way that suits your requirements. The codes here are simplified to focus on the topic of this article. For example, you can add more codes for log tracing on confirm callback, Micrometer metrics for monitoring and alerts in these cases of failure or set the publisher service to asynchronously publish the message.
Thanks for reading. Wish everyone publish the messages safely and reliably
Need more info?
Check out my article on RabbitMQ reliability below (in Thai):
