Site icon Adron's Composite Code

Go Concurrency Patterns(Producer-Consumer Pattern)

Adron

Overview

The Producer-Consumer pattern decouples the production of data from its consumption using a shared buffer (channel). Multiple producers can generate data and send it to the buffer, while multiple consumers read from the buffer and process the data. This pattern is essential for buffering bursts of work, decoupling work rates between producers and consumers, and load leveling.

NOTE: For other posts on concurrency patterns, check out the index post to this series of concurrency patterns.

Implementation Details

Structure

The producer-consumer implementation in examples/producer_consumer.go consists of three main components:

  1. Producers – Multiple goroutines that generate data
  2. Buffer – A shared channel that holds the data
  3. Consumers – Multiple goroutines that process the data

Code Analysis

Let’s break down the main function and understand how each component works:

func RunProducerConsumer() {
    bufferSize := 5
    numProducers := 2
    numConsumers := 3
    numItems := 10

    buffer := make(chan int, bufferSize)
    var wg sync.WaitGroup

    // Start producers
    for p := 1; p <= numProducers; p++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for i := 0; i < numItems; i++ {
                item := rand.Intn(100)
                buffer <- item
                fmt.Printf("Producer %d produced: %d\n", id, item)
                time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond)
            }
        }(p)
    }

    // Start consumers
    var consumerWg sync.WaitGroup
    for c := 1; c <= numConsumers; c++ {
        consumerWg.Add(1)
        go func(id int) {
            defer consumerWg.Done()
            for item := range buffer {
                fmt.Printf("Consumer %d consumed: %d\n", id, item)
                time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond)
            }
        }(c)
    }

    // Wait for all producers to finish, then close the buffer
    wg.Wait()
    close(buffer)

    // Wait for all consumers to finish
    consumerWg.Wait()
}

Step-by-step breakdown:

  1. Configuration Setup:
    • bufferSize := 5 defines the buffer capacity – can hold up to 5 items
    • numProducers := 2 launches 2 producer goroutines
    • numConsumers := 3 launches 3 consumer goroutines
    • numItems := 10 specifies how many items each producer will create
    • These values create a scenario where producers generate more data than consumers can immediately process
  2. Buffer and Synchronization Setup:
    • buffer := make(chan int, bufferSize) creates a buffered channel with capacity 5
    • Buffering allows producers to continue working even when consumers are slow
    • var wg sync.WaitGroup tracks when all producers complete
    • var consumerWg sync.WaitGroup tracks when all consumers complete
  3. Producer Launch Loop:
    • Launches numProducers goroutines (2 in this case)
    • Each producer gets a unique ID (1, 2) for tracking
    • Uses closure func(id int) { ... }(p) to capture the producer ID
    • wg.Add(1) increments the producer wait group before each goroutine
  4. Producer Goroutine Implementation:
    • defer wg.Done() ensures the producer signals completion when it exits
    • for i := 0; i < numItems; i++ loop creates exactly 10 items per producer
    • item := rand.Intn(100) generates random data (0-99) to simulate real data
    • buffer <- item sends the item to the shared buffer (may block if buffer is full)
    • time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond) simulates variable production time (100-300ms)
  5. Consumer Launch Loop:
    • Launches numConsumers goroutines (3 in this case)
    • Each consumer gets a unique ID (1, 2, 3) for tracking
    • Uses closure to capture the consumer ID
    • consumerWg.Add(1) increments the consumer wait group before each goroutine
  6. Consumer Goroutine Implementation:
    • defer consumerWg.Done() ensures the consumer signals completion when it exits
    • for item := range buffer continuously reads from the buffer until it closes
    • time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond) simulates variable processing time (100-400ms)
    • Consumers compete for items from the shared buffer (automatic load balancing)
  7. Graceful Shutdown Process:
    • wg.Wait() waits for all producers to finish generating their items
    • close(buffer) signals to consumers that no more data is coming
    • consumerWg.Wait() waits for all consumers to finish processing remaining items
    • This ensures clean termination without goroutine leaks

Key Design Patterns:

  1. Closure Pattern: func(id int) { ... }(p) captures the loop variable p in each goroutine’s closure, ensuring each goroutine gets a unique ID.
  2. Dual WaitGroup Strategy: Separate wait groups for producers and consumers allow for coordinated shutdown – producers finish first, then buffer closes, then consumers finish.
  3. Buffered Channel as Queue: The buffer decouples production and consumption rates, allowing the system to handle temporary mismatches in speed.
  4. Range Loop for Consumers: for item := range buffer automatically exits when the buffer closes, providing natural termination.
  5. Simulated Work: Random delays make concurrency visible and demonstrate how the pattern handles varying processing speeds.

How It Works

  1. Producer Initialization: Multiple producer goroutines start generating data
  2. Data Production: Producers create data and send it to the shared buffer channel
  3. Buffer Management: The buffered channel holds data until consumers are ready
  4. Consumer Processing: Multiple consumer goroutines read from the buffer and process data
  5. Graceful Shutdown: When all producers finish, the buffer is closed, signaling consumers to stop

The pattern enables asynchronous processing where producers and consumers can operate at different rates.

Why This Implementation?

Buffered Channel as Buffer

Multiple Producers

Multiple Consumers

Separate WaitGroups

Channel Closing Strategy

Key Design Decisions

  1. Buffered Channel Size: The buffer size (5) determines how much data can be queued
  2. Producer Count: Multiple producers (2) demonstrate parallel data generation
  3. Consumer Count: Multiple consumers (3) demonstrate parallel data processing
  4. Simulated Work: Random delays simulate real processing time and make concurrency visible
  5. Structured Shutdown: Proper coordination ensures clean program termination

Performance Characteristics

Throughput

Latency

Resource Usage

Common Use Cases

Data Processing Pipelines

Message Queuing Systems

Batch Processing

Real-time Systems

API Rate Limiting

Content Processing

Database Operations

The producer-consumer pattern is particularly effective when you have:

This pattern provides a robust foundation for building scalable, responsive systems that can handle varying loads efficiently.

The final example implementation looks like this:

package examples

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// RunProducerConsumer demonstrates the producer-consumer pattern with multiple producers and consumers.
func RunProducerConsumer() {
	fmt.Println("=== Producer-Consumer Pattern Example ===")

	bufferSize := 5
	numProducers := 2
	numConsumers := 3
	numItems := 10

	buffer := make(chan int, bufferSize)
	var wg sync.WaitGroup

	// Start producers
	for p := 1; p <= numProducers; p++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for i := 0; i < numItems; i++ {
				item := rand.Intn(100)
				buffer <- item
				fmt.Printf("Producer %d produced: %d\n", id, item)
				time.Sleep(time.Duration(rand.Intn(200)+100) * time.Millisecond)
			}
		}(p)
	}

	// Start consumers
	var consumerWg sync.WaitGroup
	for c := 1; c <= numConsumers; c++ {
		consumerWg.Add(1)
		go func(id int) {
			defer consumerWg.Done()
			for item := range buffer {
				fmt.Printf("Consumer %d consumed: %d\n", id, item)
				time.Sleep(time.Duration(rand.Intn(300)+100) * time.Millisecond)
			}
		}(c)
	}

	// Wait for all producers to finish, then close the buffer
	wg.Wait()
	close(buffer)

	// Wait for all consumers to finish
	consumerWg.Wait()

	fmt.Println("Producer-Consumer example completed!")
}

To run this example, and build the code yourself, check out this and other examples in the go-fluency-concurrency-model-patterns repo. That’s it for this topic, tomorrow I’ll post on the supervisor-restart pattern.

Exit mobile version