Ignore Backpressure #

In asynchronous message system architectures, data flows are very dynamic. Sometimes producer applications experience drastic traffic surges, publishing thousands of messages per second to the broker. On the other side, consumer applications may be slowing down because of degraded downstream database query performance or network latency. The balance between producer send speed and consumer process speed is the key to system stability.

When imbalances occur, the RabbitMQ broker has an automatic defense mechanism to align those data flows, known as Backpressure. However, one of the most common architectural errors ending in production infrastructure collapse is ignoring this backpressure mechanism. Developers often set consumer configurations without message pull limits (prefetch count = 0), letting the broker flood consumer application RAM until Out-Of-Memory (OOM) crashes occur. On the producer side, they often don’t monitor blocked connection warnings from the broker, aggressively pushing messages until the RabbitMQ cluster collapses completely. This article deeply dissects the operational dangers of ignoring backpressure, how the internal AMQP QoS protocol works, and how to build safe backpressure mitigations on producer and consumer sides.

The Backpressure Concept in Asynchronous Systems #

Simply put, backpressure is a way for data-receiving systems to give feedback signals to data senders that current workloads have exceeded their processing capacity, and to ask senders to slow down send rates.

In the RabbitMQ ecosystem, backpressure occurs at two main intersection points:

1. Consumer-Side Backpressure (Broker to Consumer): RabbitMQ must know how many messages can be delivered in parallel to one consumer application before that consumer sends completion confirmations (ACKs).
2. Producer-Side Backpressure (Producer to Broker): The broker must have a way to stop producers from continuously sending data if RAM capacity or disk storage on the broker machine approaches critical limits (watermarks).

Ignoring either of these two defense points triggers cascading failures spreading across our entire service infrastructure.

The Danger of Prefetch Count = 0 (Unlimited) #

The most common mistake developers make when registering consumers in RabbitMQ is not configuring the Prefetch Count limit (defaulting to 0, meaning unlimited).

When we register a consumer with prefetch = 0 using the basic.consume instruction:

1. RabbitMQ assumes the consumer has unlimited processing capacity and infinite RAM.
2. The broker immediately sprays all messages queued in the queue to the consumer’s TCP connection as fast as possible without waiting for ACKs.
3. If there are 50,000 transaction messages in the queue, all of them are delivered at once and stacked in the consumer application’s internal memory buffer.

flowchart TD
    Broker["RabbitMQ Broker<br/>'(Queue: 100.000 Messages)'"] -->|"Unlimited message spray (Prefetch = 0)"| Consumer["Consumer Application<br/>'(TCP Buffer: 100.000 Messages, <br/>RAM: Memory balloons to OS limit)'"]
    Consumer --> Crash["ERROR: OUT OF MEMORY (OOM) CRASH!"]

The direct impact of this prefetch = 0 anomaly is an Out-Of-Memory (OOM) Crash on the consumer application. The operating system (or container manager like Kubernetes) detects drastic RAM memory spikes on the consumer pod and immediately kills (SIGKILL) our application process.

Worse:

• Once consumers die, TCP connections drop, and un-ACKed messages are requeued by RabbitMQ to the original queue.
• If we have an orchestrator process (e.g., Kubernetes) automatically restarting dead consumer pods, that new pod immediately registers a new consumer with prefetch = 0 again.
• The broker again sprays 50,000 messages to that new pod, triggering another OOM crash.
• Our system gets stuck in an endless OOM Crash Loop cycle draining all monitoring logs and not processing a single business transaction.

How the Basic.Qos Mechanism and Prefetch Limits Work #

To prevent the OOM disaster above, the AMQP protocol provides the QoS (Quality of Service) command through the basic.qos instruction. This mechanism lets us set the Prefetch Count value.

Prefetch Count defines the maximum number of unacknowledged messages the broker may deliver to one consumer channel simultaneously.

How the QoS flow control works:

• If we set prefetch = 10, the RabbitMQ broker sends a maximum of 10 messages to our consumer channel.
• The broker stops sending the 11th message and holds it safely in the queue, even if that queue has millions of ready messages.
• Once our consumer finishes processing one message and sends the basic.ack confirmation, the unacked message count on that channel drops to 9.
• The broker is then allowed to send the next 1 message from the main queue to the consumer.

This QoS mechanism guarantees consumer RAM memory stays stable because the number of actively processed messages in memory is always controlled below the prefetch count threshold.

Determining Optimal Prefetch Values #

Determining Prefetch Count values must not be done speculatively. We must analyze the workload profile characteristics handled by our consumer applications:

Scenario A: I/O Bound & Slow Workloads (e.g., Heavy Database Queries, External API Calls) #

If every message takes 100ms to 2000ms to process because of slow I/O operations:

• Prefetch Recommendation: Small values (e.g., between 1 to 10 per channel).
• Reason: Setting prefetch too large only piles messages in consumer memory buffers queuing database I/O. Better to let messages stay queued at the broker so if other idle consumers exist, they can take messages evenly (fair dispatch).

Scenario B: CPU Bound & Fast Workloads (e.g., Memory Validation, Light JSON Parsing) #

If message processing runs very fast (under 5ms):

• Prefetch Recommendation: Medium to high values (e.g., between 50 to 100).
• Reason: Prefetch that’s too small (like 1) makes consumers often idle waiting for the next message to arrive through network round-trip time latency. Larger prefetch values act as buffers minimizing network transmission latency impacts between broker and consumers.

Handling Blocked Connection Alarms on Producers #

Producer-side backpressure occurs when the RabbitMQ broker detects its internal resources are pressured past safety thresholds. RabbitMQ has two critical automatic alarms:

1. Memory Alarm (High Watermark): Activated when broker RAM consumption exceeds the configured percentage limit (default 40% of total host RAM).
2. Disk Space Alarm: Activated when free disk storage space on the broker machine drops below the safe limit (default 50MB, but recommended to be set equal to host RAM size).

Once one of these alarms triggers:

• RabbitMQ sends a protocol-level control signal to block all producer connections (block connection).
• From the TCP network perspective, the broker stops reading data from producer connection sockets (zero window TCP).
• Producer applications trying to publish new messages experience execution blocking or timeouts.

Anti-Pattern: Ignoring Blocked Connections #

If our producer application code doesn’t detect this blocked status and keeps aggressively producing events:

• Connections time out.
• Badly designed producer applications do blind retry storms without pauses.
• This triggers thread/goroutine pile-ups stuck in the producer application, until eventually our producer application itself crashes or runs out of RAM memory.

Practical Solution: Notification Listeners & Circuit Breakers #

Reliable producers must listen to connection blocking notifications sent by RabbitMQ. In the AMQP protocol, RabbitMQ sends the connection.blocked and connection.unblocked methods.

Once the producer application receives the connection.blocked signal:

1. The application must activate an internal Circuit Breaker.
2. Temporarily stop all message publication activities to RabbitMQ.
3. Divert new data to a temporary local memory queue (with limit bounds), store it in a local database, or directly reject new client queries with HTTP 429 (Too Many Requests) codes.
4. When the connection.unblocked signal arrives, restore the connection and gradually resend pending data.

Data Flow Comparison with Prefetch Limits vs Without Prefetch #

The diagram below illustrates the dramatic stability difference of consumer application memory when using prefetch limits vs without restrictions (prefetch = 0).

flowchart TD
    subgraph Tanpa_Prefetch["Scenario Without Prefetch Limit (Anti-Pattern)"]
        direction TB
        B1[RabbitMQ Broker] -->|Send 10,000 Messages at Once| C1[Consumer App]
        C1 -->|RAM Balloons| O1("OOM SIGKILL")
        O1 -->|Connection Drops| B1
    end

    subgraph Dengan_Prefetch["Scenario With Prefetch Limit (Best Practice)"]
        direction TB
        B2[RabbitMQ Broker] -->|Send Maximum 5 Messages| C2[Consumer App]
        C2 -->|Process & ACK| B2
        B2 -->|Send 1 Next Message| C2
        C2 -->|RAM Stable| S2("System Safe & Smooth")
    end

    style O1 stroke:#f44336,stroke-width:2px
    style S2 stroke:#4caf50,stroke-width:2px

Go Code Implementation: QoS Prefetch Limit Settings #

Below is complete Go code demonstrating how we disciplinarily configure Qos on the consumer side to control backpressure, and how the producer side registers a listener to handle blocked connection warnings from RabbitMQ.

package main

import (
	"context"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

// ConsumerHelper manages message reading lifecycles with QoS.
type ConsumerHelper struct {
	conn *amqp.Connection
}

// StartConsumer activates the consumer loop with a limited Prefetch Count.
func (c *ConsumerHelper) StartConsumer(queueName string, prefetchCount int) {
	ch, err := c.conn.Channel()
	if err != nil {
		log.Fatalf("Gagal membuka channel consumer: %v", err)
	}
	defer ch.Close()

	// 1. SET THE QoS PREFETCH LIMIT (OOM Crash Loop Prevention)
	// prefetchSize = 0 (byte size not limited)
	// global = false (the limit applies per active consumer on this channel only)
	err = ch.Qos(
		prefetchCount, // Prefetch Count limit
		0,             // Prefetch Size
		false,         // Global
	)
	if err != nil {
		log.Fatalf("Gagal mengonfigurasi QoS Prefetch: %v", err)
	}

	msgs, err := ch.Consume(
		queueName,
		"",    // consumer tag
		false, // auto-ack set to FALSE so the broker waits for manual confirmation
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan consumer: %v", err)
	}

	log.Printf("[Consumer] Mendengarkan %s dengan Prefetch Limit: %d", queueName, prefetchCount)

	// Start message processing
	for msg := range msgs {
		log.Printf("[Consumer] Menerima pesan Tag: %d. Mulai memproses...", msg.DeliveryTag)
		
		// Simulate an I/O bound database query process (e.g., 500ms)
		time.Sleep(500 * time.Millisecond)

		// Send the ACK manually to tell the broker we're ready for the next message
		err := msg.Ack(false)
		if err != nil {
			log.Printf("[Consumer] Gagal mengirim ACK: %v", err)
		} else {
			log.Printf("[Consumer] Sukses memproses & ACK pesan Tag: %d", msg.DeliveryTag)
		}
	}
}

// ProducerHelper manages message delivery and broker alarm detection.
type ProducerHelper struct {
	conn *amqp.Connection
}

// MonitorConnectionAlerts listens for blocked/unblocked signals from the broker.
func (p *ProducerHelper) MonitorConnectionAlerts(ctx context.Context) {
	// 2. REGISTER THE NOTIFY BLOCKED LISTENER
	// NotifyBlocked sends blocked/unblocked statuses to our Go channel
	blockedChan := make(chan amqp.ConnectionBlocked)
	p.conn.NotifyBlocked(blockedChan)

	go func() {
		isBlocked := false
		for {
			select {
			case <-ctx.Done():
				return
			case blockReason, ok := <-blockedChan:
				if !ok {
					return
				}
				if blockReason.Active {
					isBlocked = true
					// 3. ACTIVATE THE INTERNAL CIRCUIT BREAKER
					log.Printf("[Producer ALERT] KONEKSI DIBLOKIR OLEH BROKER! Alasan: %s", blockReason.Reason)
					log.Println("[Producer ALERT] Memicu Circuit Breaker: Publikasi ditunda sementara.")
				} else {
					isBlocked = false
					// 4. DEACTIVATE THE CIRCUIT BREAKER
					log.Println("[Producer ALERT] KONEKSI DIPULIHKAN OLEH BROKER.")
					log.Println("[Producer ALERT] Memulihkan publikasi data secara bertahap.")
				}
			}
			// Use the isBlocked variable at the application level before calling ch.Publish
			_ = isBlocked
		}
	}()
}

func main() {
	// Connect to RabbitMQ
	conn, err := amqp.Dial("amqp://guest:***@localhost:5672/")
	if err != nil {
		log.Fatalf("Gagal koneksi ke RabbitMQ: %v", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Gagal membuka channel setup: %v", err)
	}
	defer ch.Close()

	// Declare a static queue
	queueName := "queue.backpressure.demo"
	_, err = ch.QueueDeclare(queueName, true, false, false, false, nil)
	if err != nil {
		log.Fatalf("Gagal deklarasi queue: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Initialize the helpers
	producer := &ProducerHelper{conn: conn}
	consumer := &ConsumerHelper{conn: conn}

	// Start broker alarm monitoring
	producer.MonitorConnectionAlerts(ctx)

	// Run the consumer in a separate goroutine with a Prefetch Limit = 5
	// This guarantees consumer memory safety even if the broker loads millions of messages
	go consumer.StartConsumer(queueName, 5)

	// Simulate moderate-load message publication
	go func() {
		for i := 1; i <= 20; i++ {
			err = ch.PublishWithContext(ctx,
				"",        // default exchange
				queueName, // routing key
				false,
				false,
				amqp.Publishing{
					ContentType: "text/plain",
					Body:        []byte("Contoh data beban kerja"),
				},
			)
			if err != nil {
				log.Printf("[Producer] Gagal kirim data: %v", err)
			}
			time.Sleep(100 * time.Millisecond) // Fast publication rate
		}
	}()

	// Wait for the process to finish
	time.Sleep(15 * time.Second)
}

Prefetch Limit Impact Comparison #

The following table analyzes the consequences of prefetch value choices on application system and RabbitMQ broker stability.

System Backpressure Audit Checklist #

Use the checklist below to verify whether our RabbitMQ integration architecture already has strong backpressure defenses before deploying to production:

CONSUMER & PRODUCER FLOW CONTROL AUDIT:
  □ Does every consumer application disable the Auto-ACK feature (auto-ack = false) when calling basic.consume?
  □ Do we call the 'ch.Qos(prefetchLimit, ...)' method with a limited value before activating consumer loops?
  □ Is the Prefetch Limit value logically adjusted based on processing duration (small prefetch for slow DB queries, medium prefetch for fast processes)?
  □ Does the producer application register a 'NotifyBlocked' listener to detect 'Memory Watermark' or 'Disk Alarm' alarms from the broker?
  □ Does the producer have a 'Circuit Breaker' mechanism (e.g., diverting publications to a local DB or rejecting client queries with HTTP 429) when connections are blocked?
  □ Are 'Unacked Messages' metrics and 'Blocked Connections' counts monitored in real-time on our cluster's Grafana Dashboard?

IMPROVEMENT STEPS IF NOT:
  □ Immediately add QoS/Prefetch function calls in our consumer AMQP driver initialization code.

Summary #

• Prefetch = 0 Dangers — Setting unlimited prefetch (default 0) forces RabbitMQ to instantly spray all queue backlog messages into consumer RAM memory, triggering OOM Crash Loop anomalies that cripple applications.
• basic.qos Functions — The basic.qos command limits the number of unacknowledged messages flowing in channels, keeping consumer memory usage stable.
• Blocked Connection Alarms — The RabbitMQ broker uses High Memory Watermark alarms and Disk Space Alarms to force-block producer connections if broker internal resources approach critical limits.
• Producer Circuit Breakers — Producers must listen to connection.blocked warnings through NotifyBlocked listeners to dynamically stop event publication rates, avoiding cascading system failures.
• Precise Prefetch Tuning — Use small prefetch counts (1-10) for slow task work (database I/O), and medium-high prefetches (50-100) for fast processing so optimal throughput is free from network latency.

Closing #

Designing message broker-based systems isn’t just about how fast we can deliver data from producers to consumers. True distributed system stability lies in each component’s ability to adapt to dynamic load changes. Ignoring backpressure is a reckless architecture decision, like driving a car at full speed without caring about the brakes and fuel indicators.

Remember this golden rule: In RabbitMQ architecture, backpressure isn’t a performance barrier — it’s an automatic safety system keeping our infrastructure standing when loads exceed limits.

By disciplinarily applying prefetch limits on the consumer side, listening to blocked connection warnings on the producer side, and proactively configuring monitoring metrics, we ensure our asynchronous systems have high durability, are free from costly crash loops, and always reliably serve users under various data traffic conditions.

← Previous: Not Using DLQ   Next: Design First →