Advanced Queue Types — Cheat Sheet

Priority Queues

@Configuration
public class PriorityQueueConfig {

    // Declare priority queue with max 10 priority levels (0–10)
    // Keep x-max-priority small (≤10) — each level needs its own internal queue
    @Bean
    public Queue orderPriorityQueue() {
        return QueueBuilder.durable("orders.priority")
            .withArgument("x-max-priority", 10)
            .build();
    }
}

// Publisher: set priority per message
@Service
public class OrderPublisher {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void publish(OrderEvent event, int priority) {
        rabbitTemplate.convertAndSend(
            "orders.exchange", "order.priority", event,
            message -> {
                message.getMessageProperties().setPriority(priority);
                return message;
            }
        );
    }

    public void publishUrgent(OrderEvent event) {
        publish(event, 10);   // highest priority — jumps the queue
    }

    public void publishNormal(OrderEvent event) {
        publish(event, 5);    // medium
    }

    public void publishBulk(OrderEvent event) {
        publish(event, 1);    // lowest priority
    }
}

Quorum Queues

// Low-level AMQP client
Channel channel = connection.createChannel();

Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum");
args.put("x-quorum-initial-group-size", 3);  // default: use 3 nodes

// Quorum queues are always durable=true — non-durable is not supported
channel.queueDeclare("orders-quorum", true, false, false, args);

// Spring AMQP
@Bean
public Queue ordersQuorumQueue() {
    return QueueBuilder.durable("orders-quorum")
        .quorum()               // sets x-queue-type=quorum
        .quorumInitialGroupSize(3)
        .build();
}

# OR via rabbitmq.conf — make quorum the default policy
# Match all queues with a policy:
rabbitmqctl set_policy quorum-queues ".*" \
    '{"queue-mode":"default","x-queue-type":"quorum"}' \
    --apply-to queues

Lazy Queues

// Spring AMQP — declare a lazy queue
@Bean
public Queue lazyOrderQueue() {
    return QueueBuilder.durable("orders.processing")
        .lazy()    // x-queue-mode=lazy
        .build();
}

// Or manually via arguments
@Bean
public Queue lazyQueue() {
    return QueueBuilder.durable("bulk-exports")
        .withArgument("x-queue-mode", "lazy")
        .build();
}

# Apply lazy mode to existing queues via management API or CLI
# (no restart required — takes effect for new messages)
rabbitmqctl set_policy lazy-queues ".*" \
  '{"queue-mode":"lazy"}' \
  --apply-to queues

# RabbitMQ 3.12+ — classic queues are lazy by default
# Explicitly set to default (in-memory) mode:
@Bean
public Queue defaultModeQueue() {
    return QueueBuilder.durable("hot-path")
        .withArgument("x-queue-mode", "default")  // in-memory (legacy fast mode)
        .build();
}

RabbitMQ Streams

<!-- pom.xml -->
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>stream-client</artifactId>
  <version>0.15.0</version>
</dependency>

// Declare stream via AMQP (standard RabbitTemplate)
@Bean
public Queue orderEventStream() {
    return QueueBuilder.durable("order-events-stream")
        .withArgument("x-queue-type", "stream")
        .withArgument("x-max-length-bytes", 10_000_000_000L) // 10 GB max
        .withArgument("x-max-age", "7D")                     // 7-day retention
        .build();
}

// Publishing via standard AMQP (same as classic queue)
rabbitTemplate.convertAndSend("", "order-events-stream", event);

// OR use dedicated stream Environment for high throughput
Environment env = Environment.builder()
    .host("localhost")
    .port(5552)
    .build();

Producer producer = env.producerBuilder()
    .stream("order-events-stream")
    .build();

producer.send(
    producer.messageBuilder()
        .addData(serialize(event))
        .properties().messageId(UUID.randomUUID().toString())
        .messageBuilder().build(),
    confirmationStatus -> {
        if (!confirmationStatus.isConfirmed()) {
            log.warn("Message not confirmed");
        }
    }
);

Delayed Message Exchange

# Install the plugin on the broker
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# Spring AMQP bean declaration
@Bean
public CustomExchange delayedExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");   // underlying routing type
    return new CustomExchange(
        "orders.delayed",   // exchange name
        "x-delayed-message",// exchange type
        true,               // durable
        false,              // auto-delete
        args
    );
}

@Bean
public Queue scheduledOrderQueue() {
    return QueueBuilder.durable("orders.scheduled").build();
}

@Bean
public Binding delayedBinding(Queue scheduledOrderQueue, CustomExchange delayedExchange) {
    return BindingBuilder.bind(scheduledOrderQueue)
        .to(delayedExchange).with("order.scheduled").noargs();
}