Spring Boot — Rabbitmq 101
5 min readJul 22, 2024
Installing Rabbitmq in our local DEV setup
You can use below mentioned docker command,
> docker run --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management
# --name => name of the container
# -p => port mapping b/w host and containerYou can now access rabbitmq admin console on http://localhost:15672 with these creds guest/guest.
Types of Exchanges
In RabbitMQ, exchanges route messages from producers to queues based on specific routing rules.
- Direct Exchange — Routes messages with a specific routing key to queues bound with the same key. For instance, For precise routing where the consumer must receive messages based on an exact match of the routing key.
Producer -> [Exchange] -> Queue 1 (Key: "info")
-> Queue 2 (Key: "error")- Fanout Exchange — Routes messages to all queues bound to it, ignoring the routing key. The routing key is irrelevant. For example, Broadcasting system-wide notifications or updates to all consumers.
Producer -> [Exchange] -> Queue 1
-> Queue 2
-> Queue 3- Topic Exchange — Routes messages based on wildcard patterns in the routing key. The routing key is compared against binding patterns with * (exactly single word) and # (0 or more words). Used for complex routing where messages need to be sent to queues based on pattern matching.
Producer -> [Exchange] -> Queue 1 (Key: "logs.info")
-> Queue 2 (Key: "logs.#")
-> Queue 3 (Key: "*.error")
Message 1: Routing key logs.info → Delivered to Queue 1 and Queue 2.
Message 2: Routing key logs.error → Delivered to Queue 2 and Queue 3.
Message 3: Routing key logs.info.server → Delivered to Queue 2.- Default Exchange (Implicit Direct Exchange) — A built-in direct exchange that routes messages to queues with names matching the routing key. No configuration is needed. Queue names act as routing keys.
Let’s now begin to write some code
Below mentioned is code snippet for sample Spring Boot application (transaction-service) that produces events and consumes it to send email notifications for successful transaction using rabbitmq.
1) Common code
Below mentioned code is common code needed on the consumer and producer micro-services.
Get Saksham Handu’s stories in your inbox
Join Medium for free to get updates from this writer.
1.1) Adding dependency in build.grade
dependencies {
implementation("org.springframework.boot.spring-boot-starter-amqp:3.1.6")
}1.2) Adding rabbitmq configuration in application.yml. Also, events related details needs to be mentioned here.
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
events:
workflow1:
exchange: ts_email_notification_tx
routing-key: email_notification.command.v1.send
queue: ts_email_notification_q
retry-exchange: ts_email_notification_tx
retry-queue: ts_email_notification_q
dlq: ts_email_notification_dlq
retry-count: 3
retry-delay-ms: 60000
retry-geometric-constant: 2
consumer-executor-pool-size: 11.3) Create EmailNotificationEventConfig.java which read an event workflow configuration from application.yml
@Component
@ConfigurationProperties("events.workflow1")
@Getter
@Setter
public class EmailNotificationEventConfig {
private String exchange;
private String routingKey;
private String queue;
private String retryExchange;
private String retryQueue;
private String dlq;
}1.4) Create AmqpConfiguration.java where rabbitmq bindings are mentioned.
@RequiredArgsConstructor
@Configuration
public class AmqpConfiguration {
private final static DEAD_LETTER_EXCHANGE_ARG_NAME = "x-dead-letter-exchange";
private final static RETRY_ROUTING_KEY_SUFFIX = "#";
private final EmailNotificationEventConfig eventConfig;
@Bean
public Declarables emailNotificationEventWorkflowBinding() {
val exchange = new TopicExchange(eventConfig.getExchange());
val queue = QueueBuilder.durable(eventConfig.getQueue()).build();
val retryExchange = new TopicExchange(eventConfig.getRetryExchange());
val retryQueue = QueueBuilder.durable(eventConfig.getRetryQueue())
.withArgument(DEAD_LETTER_EXCHANGE_ARG_NAME, eventConfig.getExchange())
.build();
val dlq = QueueBuilder.durable(eventConfig.getDlq()).build();
return new Declarables(exchange, queue, retryExchange, retryQueue, dlq,
bind(queue).to(exchange).with(eventConfig.getRoutingKey()),
bind(queue).to(exchange).with(eventConfig.getRoutingKey() + RETRY_ROUTING_KEY_SUFFIX))
bind(retryQueue).to(retryExchange).with(RETRY_ROUTING_KEY_SUFFIX);
}
}1.5) Create EmailPojo.java which will hold the data that needs to sent to the listener.
public class EmailPojo {
private String to;
private String[] bcc;
private String[] cc;
}Sample code snippet for emitting events
2.1) Below mentioned is a sample code used to emit an event to the rabbitmq,
@Slf4j
@Component
@RequiredArgsConstructor
public class AmqpSender {
private final ObjectMapper objectMapper;
private final AmqpTemplate amqpTemplate;
private final EmailNotificationEventConfig eventConfig;
public void sendEmail(EmailPojo emailPojo) {
try {
publish(UUID.randomUUID()), objectMapper.writeValueAsBytes(emailPojo), eventConfig.getExchange(), eventConfig.getRoutingKey());
} catch (final JsonProcessingException ex) {
log.error(ex.getMessage());
}
}
private void publish(final UUID id, final byte[] payload, final String exchange, final String routingKey) {
val message = MessageBuilder.withBody(payload)
.setMessageId(id.toString())
.setContentType(MediaType.APPLICATION_JSON_VALUE)
.build();
amqpTemplate.send(exchange, routingKey, message);
}
}Sample code snippet for consuming events
Below mentioned is a sample code used to listen to a queue (with few retries when listener logic fails). When the listener logic fails above threshold count, that message is discarded and sent to DLQ.
3.1) Create AmqpListener.java which will listen to the events and execute the logic to trigger email notification for the transaction in our scenario.
@Slf4j
@Service
@RequiredArgsConstructor
public class AmqpListener {
private final static EMAIL_NOTIFICATION_LISTENER_ID = "emailNotificationListenerId";
private final ObjectMapper objectMapper;
private final Validator validator;
@RabbitListener(
id = EMAIL_NOTIFICATION_LISTENER_ID,
queue = "${events.workflow1.queue}"
errorHandler = EmailNotificationEventListenerErrorHandler.BEAN_NAME,
concurrency = "${events.workflow1.consumer-executor-pool-size}")
public void processEmailNotificationListenerId(final Message message) {
// creating message payload
val emailPojo = parse(message, EmailPojo.class);
// add listener logic here
}
private <T> T parse(final Message message, final Class<T> clazz) {
val messageId = EventUtility.messageId(message);
try {
val value = objectMapper.readValue(message.getBody(), clazz);
val violations = validator.validate(value);
if (!CollectionUtils.isEmpty(violations)) {
violations.forEach((violation -> log.error(toPrettyMessage(violation))));
throw new MalformedEventMessageException("Validation of event message failed");
}
return value;
} catch (final Exception ex) {
throw new MalformedEventMessageException("Failed to parse event message");
}
}
private String toPrettyMessage(final ConstraintViolations<?> constraintViolation) {
val className = constraintViolation.getRootBeanClass().getSimpleName();
val target = constraintViolation.getPropertyPath().toString();
val message = constraintViolation.getMessage();
return String.format("%s.%s: %s", className, target, message);
}
}3.2) Create BaseEventListenerErrorHandler.java
@Slf4j
@RequiredArgsContructor
abstract class BaseEventListenerErrorHandler implements RabbitListenerErrorHandler {
protected final AmqpTemplate amqpTemplate;
protected final EmailNotificationEventConfig eventConfig;
protected abstract String dlq();
protected abstract String retryExchange();
@Override
public Object handle(final Message amqpMessage,
final org.springframework.messaging.Message<?> message,
final ListenerExecutionFailedException exceptionn exception) {
val messageId = EventUtility.messageId(amqpMessage);
val cause = exception.getCause();
val retryNumber = EventUtility.retryNumber(amqpMessage);
if (retryNumber >= eventConfig.getRetryCount()) {
sendToDLQ(amqpMessage);
return null;
}
val delayMs = calcDelay(amqpMessage);
sendForRetry(amqpMessage, delayMs);
return null;
}
private void sendToDlq(final Message message) {
amqpTemplate.send(dlq(), message);
}
private long calcDelay(final Message message) {
val retryNumber = EventUtility.retryNumber(message);
val waitTime = eventConfig.getRetryDelayMs();
val geometricConstant = eventConfig.getRetryGeometricConstant();
return waitTime * Math.round(Math.pow(geometricConstant, retryNumber));
}
private void sendForRetry(final Message message, final long delayMs) {
message.getMessageProperties().setExpiration(Long.toString(delayMs));
EventUtility.increaseRetryNumber(message);
amqpTemplate.send(retryExchange(), delayedRoutingKey(message), message);
}
private String delayRoutingKey(final Message message) {
val receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
return receivedRoutingKey.endsWith(AmqpConfiguration.RETRY_ROUTING_KEY_SUFFIX)
? receivedRoutingKey : receivedRoutingKey ? recievedRoutingKey + AmqpConfiguration.RETRY_ROUTING_KEY_SUFFIX;
}
}3.3) Create EmailNotificationEventListenerErrorHandler.java
@Component()
public class EmailNotificationEventListenerErrorHandler extends BaseEventListenerHandler {
public static final String BEAN_NAME = "EmailNotificationEventListenerErrorHandler";
public EmailNotificationEventListenerErrorHandler(final AmqpTemplate amqpTemplate, final EmailNotificationEventConfig eventConfig) {
super(amqpTemplate, eventConfig);
}
@Override
protected String dlq() {
return eventConfig.getDlq();
}
@Override
protected String retryExchange() {
return eventConfig.getRetryExchange();
}
}3.4) Create EventUtility.java which is common utility
public final class EventUtility {
public static final String RETRY_NUMBER_HEADER_NAME = "x-retry-number";
public static String messageId(final Message message) {
return message.getMessageProperties().getMessageId();
}
public static int retryNumber(final Message message) {
val messageProperties = message.getMessageProperties();
val retryNumber = messageProperties.getHeaders().getOrDefault(RETRY_NUMBER_HEADER_NAME, 0);
return Integer.valueOf(retryNumber.toString());
}
public static void increaseRetryNumber(final Message message) {
val messageHeaders = message.getMessageProperties().getHeaders();
messageHeaders.put(RETRY_NUMBER_HEADER_NAME, retryNumber(message) + 1);
}
}Note
- We usually create 2 exchanges in micro-service, first exchange acts as a Gateway to the application and the seconds one is used for retries. These 2 exchanges can be shared across different workflows in micro-service.
- We can configure ListenerContainerFactory as shown below,
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
final CachingConnectionFactory connectionFactory) {
val factory = new SimpleRabbitListenerContainerFactory();
// configs related to message consumption
factory.setConnectionFactory(connectionFactory);
factory.setConcurrentConsumers(1);
factory.setMaxConcurrentConsumers(5);
factory.setTaskExecutor(Executors.newFixedThreadPool(10));
factory.setPrefetchCount(15);
// reconnect logic to rabbitmq instance
factory.setAutoStartup(false);
factory.setRecoveryInterval(1000); // in milliseconds
factory.setMissingQueuesFatal(false);
// setup MDC interceptor
factory.setAdviceChain(new MdcAdvice());
}
