10 Message Queue Patterns for Spring Boot: Decoupling, Async, Traffic Shaping & More

Introduction

In distributed and microservice architectures, direct service-to-service calls often lead to tight coupling and performance bottlenecks. Message queues (MQ) act as a buffering layer that enables asynchronous decoupling, handles high-concurrency traffic spikes, and ensures eventual data consistency. Mastering core MQ patterns is essential for building highly available, high-performance systems.

2.1 System Decoupling

Problem: An order creation workflow that calls multiple downstream services (inventory, points, email, analytics) creates tight coupling. Adding a new consumer requires modifying the order service.

Before MQ (tight coupling):

public class OrderService {

private final InventoryService inventoryService;

private final PointsService pointsService;

private final EmailService emailService;

private final AnalyticsService analyticsService;

public void createOrder(Order order) {

orderDao.save(order);

inventoryService.updateInventory(order);

pointsService.addPoints(order.getUserId(), order.getAmount());

emailService.sendOrderConfirmation(order);

analyticsService.trackOrderCreated(order);

}

}

Solution: Publish an OrderCreatedEvent to RabbitMQ. Consumers subscribe independently.

@Service

public class OrderService {

private final RabbitTemplate rabbitTemplate;

public void createOrder(Order order) {

orderDao.save(order);

// 2. Send message to MQ

rabbitTemplate.convertAndSend(

"order.exchange",

"order.created",

new OrderCreatedEvent(

order.getId(),

order.getUserId(),

order.getAmount()

)

);

}

}

@Component

@RabbitListener(queues = "inventory.queue")

public class InventoryConsumer {

private final InventoryService inventoryService;

@RabbitHandler

public void handleOrderCreated(OrderCreatedEvent event) {

inventoryService.updateInventory(event.getOrderId());

}

}

Now the Order Service is decoupled from consumer implementation details.

2.2 Asynchronous Processing

Operations like video transcoding, thumbnail generation, content moderation, and user notification should not block the user's synchronous request.

Example: Video upload system. Synchronous processing forces the user to wait for all steps. With MQ, the API returns immediately while background consumers handle heavy tasks.

@Service

public class VideoService {

@Autowired

private KafkaTemplate<String, Object> kafkaTemplate;

public UploadResponse uploadVideo(MultipartFile file, String userId) {

// 1. Save data

String videoId = saveOriginalVideo(file);

// 2. Send message

kafkaTemplate.send(

"video-processing",

new VideoProcessingEvent(videoId, userId)

);

// 3. Return immediately

return new UploadResponse(videoId, "upload_success");

}

}

@Service

public class VideoProcessingConsumer {

@KafkaListener(topics = "video-processing")

public void processVideo(VideoProcessingEvent event) {

videoProcessor.transcode(event.getVideoId());

videoProcessor.generateThumbnails(event.getVideoId());

contentModerationService.checkContent(event.getVideoId());

notificationService.notifyUser(

event.getUserId(),

event.getVideoId()

);

}

}

2.3 Traffic Shaping (Peak Shaving)

One of the most critical production use cases. Example: Flash sale traffic spikes from 100 req/sec to 50,000 req/sec. Databases cannot scale 500x instantly. MQ acts as a buffer.

@Service

public class SecKillService {

private final RedisTemplate<String, Object> redisTemplate;

private final RabbitTemplate rabbitTemplate;

public SecKillResponse secKill(SecKillRequest request) {

// 1. Check user eligibility

if (!checkUserQualification(request.getUserId())) {

return SecKillResponse.failed("User is not eligible");

}

// 2. Pre-decrement stock in Redis (atomic)

Long remaining = redisTemplate.opsForValue().decrement(

"sec_kill_stock:" + request.getItemId()

);

if (remaining == null || remaining < 0) {

redisTemplate.opsForValue().increment(

"sec_kill_stock:" + request.getItemId()

);

return SecKillResponse.failed("Insufficient stock");

}

// 3. Send flash-sale success message

rabbitTemplate.convertAndSend(

"sec_kill.exchange",

"sec_kill.success",

new SecKillSuccessEvent(

request.getUserId(),

request.getItemId()

)

);

return SecKillResponse.success("Flash sale successful");

}

}

@Component

@RabbitListener(queues = "sec_kill.order.queue")

public class SecKillOrderConsumer {

@RabbitHandler

public void handleSecKillSuccess(SecKillSuccessEvent event) {

orderService.createSecKillOrder(

event.getUserId(),

event.getItemId()

);

}

}

The queue absorbs traffic bursts, allowing consumers to process at a controlled rate.

2.4 Data Synchronization

Microservices maintain separate databases (e.g., User Service → User DB, Order Service → Order DB). When user data changes, Order Service may need a local copy/cache. To avoid tight coupling, User Service publishes an event.

@Service

public class UserService {

@Transactional

public User updateUser(User user) {

// 1. Update database

userDao.update(user);

// 2. Send message within transaction

rocketMQTemplate.sendMessageInTransaction(

"user-update-topic",

MessageBuilder.withPayload(

new UserUpdateEvent(user.getId(), user.getStatus())

).build(),

null

);

return user;

}

}

@Service

@RocketMQMessageListener(

topic = "user-update-topic",

consumerGroup = "order-group"

)

public class UserUpdateConsumer implements RocketMQListener<UserUpdateEvent> {

@Override

public void onMessage(UserUpdateEvent event) {

orderService.updateUserCache(

event.getUserId(),

event.getStatus()

);

}

}

2.5 Centralized Log Collection

In distributed systems, logs are scattered across many machines/services. Searching each server individually is impractical. MQ serves as a transport layer.

Service A --> logs

Service B --> logs

Service C --> logs

Service D --> logs

Applications publish log events to Kafka. Consumers store them in Elasticsearch or perform real-time monitoring. The same log stream can feed multiple consumers.

@Component

public class LogCollector {

private final KafkaTemplate<String, String> kafkaTemplate;

public void collectLog(

String appId,

String level,

String message,

Map<String, Object> context

) {

LogEvent logEvent = new LogEvent(

appId,

level,

message,

context,

System.currentTimeMillis()

);

kafkaTemplate.send(

"app-logs",

appId,

JsonUtils.toJson(logEvent)

);

}

}

@Service

public class LogConsumer {

@KafkaListener(topics = "app-logs", groupId = "log-es")

public void consumeLog(String message) {

LogEvent logEvent = JsonUtils.fromJson(message, LogEvent.class);

elasticsearchService.indexLog(logEvent);

if ("ERROR".equals(logEvent.getLevel())) {

alertService.checkAndAlert(logEvent);

}

}

}

2.6 Broadcast Messages

When every service instance must receive the same message (e.g., configuration changes across 100 instances), broadcast messaging distributes the update.

@Service

public class ConfigService {

private final RedisTemplate<String, Object> redisTemplate;

public void updateConfig(String configKey, String configValue) {

// 1. Update database config

configDao.updateConfig(configKey, configValue);

// 2. Broadcast config update

redisTemplate.convertAndSend(

"config-update-channel",

new ConfigUpdateEvent(configKey, configValue)

);

}

}

// Subscriber

@Component

public class ConfigUpdateListener {

private final LocalConfigCache localConfigCache;

@RedisListener(channel = "config-update-channel")

public void handleConfigUpdate(ConfigUpdateEvent event) {

localConfigCache.updateConfig(

event.getKey(),

event.getValue()

);

}

}

2.7 Ordered Messages

Some business workflows require strict ordering. Example: Order state transitions (CREATED → PAID → SHIPPED → DELIVERED). If SHIPPED arrives before PAID, the system enters an invalid state.

Solution: Use order ID as sharding/partition key to ensure messages for the same order go to the same partition and are processed sequentially.

Order ID = 12345

|

v

Same partition

|

v

Messages processed sequentially

@Service

public class OrderStateService {

private final RocketMQTemplate rocketMQTemplate;

public void changeOrderState(String orderId, String oldState, String newState) {

OrderStateEvent event = new OrderStateEvent(orderId, oldState, newState);

// Send ordered message using orderId as sharding key

rocketMQTemplate.syncSendOrderly(

"order-state-topic",

event,

orderId  // Ensures same-order messages processed in order

);

}

}

@Service

@RocketMQMessageListener(

topic = "order-state-topic",

consumerGroup = "order-state-group",

consumeMode = ConsumeMode.ORDERLY

)

public class OrderStateConsumer implements RocketMQListener<OrderStateEvent> {

@Override

public void onMessage(OrderStateEvent event) {

orderService.processStateChange(event);

}

}

2.8 Delayed Messages

How to execute an action after a specific delay? Example: Unpaid order cancellation after 30 minutes.

// Order Service - Send Delayed Message

@Service

public class OrderService {

@Autowired

private RabbitTemplate rabbitTemplate;

public void createOrder(Order order) {

// 1. Save order

orderDao.save(order);

// 2. Send delayed message to check payment after 30 minutes

rabbitTemplate.convertAndSend(

"order.delay.exchange",

"order.create",

new OrderCreateEvent(order.getId()),

message -> {

message.getMessageProperties().setDelay(30 * 60 * 1000); // 30 minutes

return message;

}

);

}

}

@Component

@RabbitListener(queues = "order.delay.queue")

public class OrderTimeoutConsumer {

@RabbitHandler

public void checkOrderPayment(OrderCreateEvent event) {

Order order = orderDao.findById(event.getOrderId());

if ("UNPAID".equals(order.getStatus())) {

orderService.cancelOrder(

order.getId(),

"Payment timeout"

);

}

}

}

2.9 Message Retry

Distributed systems inevitably face transient failures: database unavailable, downstream API timeout, network failure, dependent service restart. Messages must not be lost. A robust architecture implements retry with backoff and dead-letter queues (DLQ).

@Service

@Slf4j

public class RetryableConsumer {

private final RabbitTemplate rabbitTemplate;

@RabbitListener(queues = "business.queue")

public void processMessage(Message message, Channel channel) {

try {

// Business processing

businessService.process(message);

// ACK

channel.basicAck(

message.getMessageProperties().getDeliveryTag(),

false

);

} catch (TemporaryException e) {

log.warn("Processing failed, retrying", e);

// Reject and requeue

channel.basicNack(

message.getMessageProperties().getDeliveryTag(),

false,

true // requeue

);

} catch (PermanentException e) {

// Permanent failure, send to DLQ

log.error("Processing failed, sending to dead-letter queue", e);

channel.basicNack(

message.getMessageProperties().getDeliveryTag(),

false,

false // do not requeue

);

}

}

}

Retry immediately on transient failure, then wait with increasing intervals. Messages that cannot be processed eventually move to a DLQ for investigation or later recovery. Critical production principle: never retry indefinitely, or a "poison message" can consume resources and trigger a retry storm.

2.10 Transactional Messages

Advanced scenario: Create an order in the database AND publish an OrderCreated event atomically. If DB commit succeeds but message publish fails, downstream services never receive the event. If message publishes but DB rolls back, consumers act on a non-existent order. This is a classic distributed consistency problem.

Transactional messaging coordinates local transactions with message publishing. RocketMQ provides a two-phase commit protocol.

@Service

public class TransactionalMessageService {

private final RocketMQTemplate rocketMQTemplate;

@Transactional

public void createOrderWithTransaction(Order order) {

// 1. Save order

orderRepository.save(order);

// 2. Send transactional message

TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(

"order-tx-topic",

MessageBuilder.withPayload(

new OrderCreatedEvent(order.getId())

).build(),

order // Transaction argument

);

if (!result.getLocalTransactionState()

.equals(LocalTransactionState.COMMIT_MESSAGE)) {

throw new RuntimeException("Transactional message sending failed");

}

}

}

@Component

@RocketMQTransactionListener

public class OrderTransactionListener implements RocketMQLocalTransactionListener {

private final OrderRepository orderRepository;

@Override

public RocketMQLocalTransactionState executeLocalTransaction(

Message msg,

Object arg

) {

try {

// Check local transaction state

Order order = (Order) arg;

Order existingOrder = orderRepository.findById(order.getId());

if (existingOrder != null

&& "CREATED".equals(existingOrder.getStatus())) {

return RocketMQLocalTransactionState.COMMIT_MESSAGE;

} else {

return RocketMQLocalTransactionState.ROLLBACK_MESSAGE;

}

}

}

@Override

public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {

// Check local transaction state

String orderId = (String) msg.getHeaders().get("order_id");

Order order = orderDao.findById(orderId);

if (order != null && "CREATED".equals(order.getStatus())) {

return RocketMQLocalTransactionState.COMMIT_MESSAGE;

} else {

return RocketMQLocalTransactionState.ROLLBACK_MESSAGE;

}

}

}

The transaction listener can also verify local transaction state when needed. This is especially useful when business data and events must remain consistent.