Producer-Consumer Pattern #

When a system needs to process data continuously — whether it is an email queue, a stream of events from IoT sensors, or nightly batch jobs — there are almost always two responsibilities involved: producing data and processing data. The problem arises when both responsibilities are done in one tightly coupled flow: the producer must wait for the consumer to finish, or the consumer keeps polling even though there is no new data. The result is low throughput, high latency, and hard-to-test code. The Producer-Consumer Pattern breaks those two responsibilities into standalone units that communicate through a shared buffer, so both can run concurrently without directly blocking each other.

What Is the Producer-Consumer Pattern? #

The Producer-Consumer Pattern is a concurrency pattern that separates the process of producing data from the process of consuming it through a shared queue or buffer.

The three core components of this pattern are:

The communication flow is always one-way:

flowchart LR
    P1([Producer 1]) --> B[(Buffer / Channel)]
    P2([Producer 2]) --> B
    B --> C1([Consumer 1])
    B --> C2([Consumer 2])
    B --> C3([Consumer 3])

Producers and consumers run independently. The producer does not know who will consume its data, and the consumer does not know how the data is generated. The only thing they know is: there is a buffer between them.

In Go, this buffer is natively represented using a channel — a thread-safe concurrency primitive already in the runtime and idiomatic for goroutine communication.

Why Channels, Not Mutexes? #

Before diving into the implementation, it is important to understand why Go encourages channels as the communication mechanism instead of mutexes or plain shared memory.

Go’s famous principle says: “Do not communicate by sharing memory; instead, share memory by communicating.” In other words, rather than several goroutines accessing one shared variable with locks scattered around, it is better for the goroutines to send and receive data through a channel.

Consider the difference between the two approaches:

// ANTI-PATTERN: shared memory with a mutex — prone to data races if a lock is forgotten
var queue []int
var mu sync.Mutex

func produce(val int) {
    mu.Lock()
    queue = append(queue, val)
    mu.Unlock()
}

func consume() int {
    mu.Lock()
    defer mu.Unlock()
    if len(queue) == 0 {
        return -1 // no idiomatic way to block here
    }
    val := queue[0]
    queue = queue[1:]
    return val
}

// CORRECT: channel as buffer — blocking happens automatically, no manual mutex needed
ch := make(chan int, 10)

go func() { ch <- produceValue() }()  // producer
go func() { process(<-ch) }()         // consumer

Channels give you blocking semantics for free: the producer automatically waits when the buffer is full, and the consumer automatically waits when the buffer is empty. No sync.Cond, no polling loops.

Single Producer, Single Consumer #

The simplest implementation is one producer goroutine and one consumer goroutine. This is a good starting point for understanding the basic mechanics.

package main

import (
	"fmt"
	"time"
)

// producer sends data to the channel and closes it when done
func producer(ch chan<- int) {
	for i := 1; i <= 5; i++ {
		fmt.Printf("[Producer] Producing item: %d\n", i)
		ch <- i
		time.Sleep(300 * time.Millisecond)
	}
	close(ch) // CORRECT: channel closed from the producer side
}

// consumer reads from the channel until it is closed
func consumer(ch <-chan int) {
	for item := range ch { // range stops automatically when the channel is closed
		fmt.Printf("[Consumer] Processing item: %d\n", item)
		time.Sleep(700 * time.Millisecond)
	}
	fmt.Println("[Consumer] Done")
}

func main() {
	ch := make(chan int, 3) // buffered channel, capacity 3

	go producer(ch)
	consumer(ch) // consumer runs on the main goroutine
}

There are several important details here:

First, the channel is declared with capacity 3. This means the producer can push up to 3 items before blocking, giving the producer “room to breathe” while the consumer is busy processing previous items.

Second, close(ch) is called from the producer, not the consumer. This is the standard rule in Go: only the sender may close a channel.

Third, for item := range ch in the consumer stops automatically when the channel is closed and all items have been read — no manual termination condition needed.

The execution flow can be visualized like this:

sequenceDiagram
    participant P as Producer
    participant CH as Channel (buf=3)
    participant C as Consumer

    P->>CH: send(1)
    P->>CH: send(2)
    P->>CH: send(3)
    Note over P,CH: Buffer full, producer blocking
    C->>CH: receive() → 1
    Note over CH,C: Consumer starts processing item 1
    P->>CH: send(4)
    C->>CH: receive() → 2
    P->>CH: send(5)
    P->>CH: close()
    C->>CH: receive() → 3
    C->>CH: receive() → 4
    C->>CH: receive() → 5
    Note over C: Channel closed, range done

Multiple Producers & Multiple Consumers #

The real power of the Producer-Consumer Pattern shows when you scale to many producers and consumers. Go makes this very easy because channels are already thread-safe — many goroutines can write to and read from the same channel without race conditions.

package main

import (
	"fmt"
	"sync"
	"time"
)

func producer(id int, ch chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 1; i <= 4; i++ {
		item := fmt.Sprintf("P%d-item%d", id, i)
		fmt.Printf("[Producer %d] Producing: %s\n", id, item)
		ch <- item
		time.Sleep(200 * time.Millisecond)
	}
}

func consumer(id int, ch <-chan string, wg *sync.WaitGroup) {
	defer wg.Done()
	for item := range ch {
		fmt.Printf("[Consumer %d] Processing: %s\n", id, item)
		time.Sleep(500 * time.Millisecond)
	}
}

func main() {
	const numProducers = 3
	const numConsumers = 2
	const bufferSize = 10

	ch := make(chan string, bufferSize)

	var prodWG sync.WaitGroup
	var consWG sync.WaitGroup

	// Start consumers first so they are ready to receive data
	for i := 1; i <= numConsumers; i++ {
		consWG.Add(1)
		go consumer(i, ch, &consWG)
	}

	// Start producers
	for i := 1; i <= numProducers; i++ {
		prodWG.Add(1)
		go producer(i, ch, &prodWG)
	}

	// Wait for all producers to finish, then close the channel
	prodWG.Wait()
	close(ch)

	// Wait for all consumers to finish processing remaining data
	consWG.Wait()

	fmt.Println("All items processed successfully")
}

The operation order here is important and is often a source of bugs when done wrong:

flowchart TD
    A[Start consumers] --> B[Start producers]
    B --> C[prodWG.Wait\\nwait for all producers to finish]
    C --> D[close ch\\nclose the channel]
    D --> E[consWG.Wait\\nwait for consumers to drain remaining items]
    E --> F[Program finished]

Do not close the channel before all producers are done. If a producer is still running when close(ch) is called, the program panics with send on closed channel. Always use a sync.WaitGroup to ensure all producers have finished before closing the channel.

Understanding Backpressure #

Backpressure is a mechanism where a slow consumer automatically “pushes back” on the producer so it does not produce data faster than can be processed. This is one of the biggest advantages of a channel-based Producer-Consumer Pattern.

When the buffer is full, the ch <- data operation in the producer blocks until a consumer picks up an item. This is not a bug — it is the right design for keeping the system stable under high load.

// ANTI-PATTERN: unbuffered channel with unlimited goroutines
// every item gets a new goroutine — can exhaust memory
func processAll(items []int) {
	ch := make(chan int) // unbuffered
	for _, item := range items {
		go func(v int) { ch <- v }() // ✗ goroutine leak if no consumer
	}
}

// CORRECT: bounded buffer with a fixed number of consumers
func processAll(items []int) {
	ch := make(chan int, 50) // bounded buffer
	var wg sync.WaitGroup

	// Fixed consumer pool
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for item := range ch {
				process(item) // ✓ fixed consumer count, natural backpressure
			}
		}()
	}

	for _, item := range items {
		ch <- item // blocks when the buffer is full — backpressure at work
	}

	close(ch)
	wg.Wait()
}

Guidelines for choosing the buffer size:

Cancellation with Context #

Long-running systems — like background job processors or daemons — need to be stoppable cleanly. The idiomatic pattern in Go is using context.Context as the cancellation signal.

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func producer(ctx context.Context, ch chan<- int) {
	defer close(ch)
	i := 0
	for {
		select {
		case <-ctx.Done():
			fmt.Println("[Producer] Context cancelled, stopping production")
			return
		case ch <- i:
			fmt.Printf("[Producer] Producing: %d\n", i)
			i++
			time.Sleep(200 * time.Millisecond)
		}
	}
}

func consumer(ctx context.Context, id int, ch <-chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			fmt.Printf("[Consumer %d] Context cancelled, stopping consumption\n", id)
			return
		case item, ok := <-ch:
			if !ok {
				fmt.Printf("[Consumer %d] Channel closed, done\n", id)
				return
			}
			fmt.Printf("[Consumer %d] Processing: %d\n", id, item)
			time.Sleep(500 * time.Millisecond)
		}
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	ch := make(chan int, 5)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go consumer(ctx, i, ch, &wg)
	}

	go producer(ctx, ch)

	wg.Wait()
	fmt.Println("All consumers finished")
}

Using select with two cases — ctx.Done() and the channel operation — lets producers and consumers respond to cancellation signals without polling or shared boolean variables.

stateDiagram-v2
    [*] --> Running : Start goroutine
    Running --> Processing : Receive item from channel
    Processing --> Running : Item processed
    Running --> Cancelled : ctx.Done() receives a signal
    Running --> Done : Channel closed, ok == false
    Cancelled --> [*]
    Done --> [*]

Pipeline: Chaining Producer-Consumer #

The Producer-Consumer Pattern can evolve into a pipeline — a series of stages where the output of one stage becomes the input of the next. This is very useful for data processing with several transformation steps.

package main

import (
	"fmt"
	"strings"
)

// Stage 1: generate words
func generate(words ...string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for _, w := range words {
			ch <- w
		}
	}()
	return ch
}

// Stage 2: convert to uppercase
func toUpper(in <-chan string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for w := range in {
			ch <- strings.ToUpper(w)
		}
	}()
	return ch
}

// Stage 3: add a prefix
func addPrefix(prefix string, in <-chan string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for w := range in {
			ch <- fmt.Sprintf("[%s] %s", prefix, w)
		}
	}()
	return ch
}

func main() {
	// Chain the pipeline: generate → toUpper → addPrefix
	words := generate("golang", "concurrency", "channel", "pipeline")
	upper := toUpper(words)
	result := addPrefix("OUTPUT", upper)

	for item := range result {
		fmt.Println(item)
	}
}

The result:

[OUTPUT] GOLANG
[OUTPUT] CONCURRENCY
[OUTPUT] CHANNEL
[OUTPUT] PIPELINE

Each stage runs as a separate goroutine and communicates through channels. This is a concrete example of how Producer-Consumer becomes the foundation of more complex patterns.

flowchart LR
    A[generate\\nwords] -->|chan string| B[toUpper]
    B -->|chan string| C[addPrefix]
    C -->|chan string| D[main\\nconsumer]

Producer-Consumer vs Worker Pool #

These two patterns are often mentioned together and are indeed closely related, but they have different focuses.

In practice, a Worker Pool is a specialization of Producer-Consumer: you have a producer filling the queue, and a worker pool (consumers) with a fixed count taking from that queue. Understanding Producer-Consumer well is a prerequisite to understanding Worker Pool.

Anti-Patterns to Avoid #

// ✗ Closing the channel from the consumer — panics if the producer is still sending
func badConsumer(ch chan int) {
    val := <-ch
    close(ch) // DON'T: only the producer may close the channel
    process(val)
}

// ✓ Close the channel only from the producer; the consumer only reads
func goodConsumer(ch <-chan int) { // <-chan: read-only, cannot close
    for val := range ch {
        process(val)
    }
}

// ✗ Goroutine leak: the consumer has no exit path
func leakySetup() {
    ch := make(chan int)
    go func() {
        for val := range ch { // this goroutine never finishes
            process(val)
        }
    }()
    // ch is never closed → goroutine leak forever
}

// ✓ Always make sure the channel will be closed
func safeSetup() chan int {
    ch := make(chan int, 10)
    go func() {
        defer close(ch) // ✓ deferred close ensures the channel is always closed
        for _, item := range fetchItems() {
            ch <- item
        }
    }()
    return ch
}

// ✗ Unbounded buffer via a new goroutine per item
func unboundedProducer(items []int) {
    ch := make(chan int)
    for _, item := range items { // 1 million items = 1 million goroutines
        go func(v int) { ch <- v }() // ✗ can exhaust memory
    }
}

// ✓ Use a buffered channel with a reasonable size
func boundedProducer(items []int) <-chan int {
    ch := make(chan int, 100) // ✓ bounded buffer, natural backpressure
    go func() {
        defer close(ch)
        for _, item := range items {
            ch <- item
        }
    }()
    return ch
}

// ✗ Not handling panics in the consumer — can kill the entire program
func unsafeConsumer(ch <-chan int) {
    for val := range ch {
        riskyOperation(val) // if this panics, the program crashes
    }
}

// ✓ Recover panics in the consumer so the goroutine pool stays alive
func safeConsumerWithRecover(ch <-chan int) {
    for val := range ch {
        func() {
            defer func() {
                if r := recover(); r != nil {
                    log.Printf("Recovered panic: %v", r) // ✓ log and continue
                }
            }()
            riskyOperation(val)
        }()
    }
}

Producer-Consumer Review Checklist #

CHANNEL DESIGN:
  □ Channels are directional (chan<- for producers, <-chan for consumers)
  □ The buffer size is chosen based on the producer/consumer speed ratio
  □ Channels are only closed from the producer side

GOROUTINE LIFECYCLE:
  □ Every goroutine has a clear exit path (context, channel close, or done signal)
  □ No goroutine can leak (channels that are never closed)
  □ A WaitGroup is used to synchronize goroutine completion

CANCELLATION:
  □ A context is used for cancellation signals in long-running systems
  □ select is used to be responsive to both ctx.Done() and the channel
  □ Timeouts are considered for operations that could hang

ERROR HANDLING:
  □ Panics in consumers are recovered so they do not kill the whole program
  □ Consumer errors are communicated back (e.g. via a separate error channel)
  □ Partial failures in multi-producer/consumer setups are handled correctly

OBSERVABILITY:
  □ Logging exists at every stage for debugging
  □ Buffer length and throughput metrics are monitored
  □ Goroutine count is monitored to detect leaks

Summary #

• Producer-Consumer separates two responsibilities — producing data and processing data — so both can run concurrently without direct coupling.
• A channel is the idiomatic buffer in Go — thread-safe, supports blocking semantics, and eliminates the need for manual mutexes in goroutine communication.
• Only the producer may close the channel — closing it from the consumer causes a panic if the producer is still sending.
• for item := range ch is the idiomatic way to read a channel — it stops automatically when the channel is closed and all items are read.
• sync.WaitGroup is a must for ensuring all producers finish before the channel closes, and all consumers finish before the program exits.
• Backpressure works automatically — a full channel blocks the producer, keeping the system stable without manual throttling logic.
• Use context.Context for cancellation in long-running systems — select with ctx.Done() makes goroutines responsive to stop signals.
• A pipeline is a natural evolution of Producer-Consumer — one stage’s output becomes the next stage’s input, each running as a separate goroutine.
• A Worker Pool is a specialization of Producer-Consumer with a fixed (bounded) consumer count — understanding Producer-Consumer is a prerequisite for understanding Worker Pool.
• Recover panics in consumers so one failed item does not kill the whole goroutine pool.

← Previous: Worker Pool   Next: Future-Promise →