Producer-Consumer Pattern in Go

In this post, Goroutines and channels are introduced. These are 2 of the most useful constructs in go. Together, when used properly, they give developers great flexibility in dealing with concurrency. They are one of the most common topics in an interview.

Implement a simple producer consumer pattern in go.

var buffer = make(chan int, 5)

func produce(wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 10; i++ {
        buffer <- i
        time.Sleep(time.Millisecond * time.Duration(rand.Intn(100)))
    }
    fmt.Println("producer done")
}

func consume(wg *sync.WaitGroup) {
    defer wg.Done()
    for data := range buffer {
        fmt.Println(data)
        time.Sleep(time.Millisecond * time.Duration(rand.Intn(400)))
    }
    fmt.Println("consumer done")
}

func main() {
    var producerWg sync.WaitGroup
    var consumerWg sync.WaitGroup
    producerWg.Add(1)
    go produce(&producerWg)
    go func() {
        producerWg.Wait()
        close(buffer)
        fmt.Println("closed channel")
    }()
    consumerWg.Add(1)
    go consume(&consumerWg)
    consumerWg.Wait()
    fmt.Println("done")
}

This is one of the simplest implementations; but the pattern is quite common. We have a thread 'producing' values and a thread that must 'consume' them. In golang, the way to pass these values between threads is a channel.

We start by creating a channel for the integers. Then create the routines that implement the producer and consumer functions.

In any multithread situation, synchronization is a problem. Golang created WaitGroup's as one means to implement synchronization. They work simply as counters and a thread that needs to synchronize will wait until the count is 0. A controlling thread uses the Done() function to decrement the counter.

In this problem, we create a WaitGroup for both the producer and the consumer, initializing both to count 1 (using the Add() function).

The main thread launches the producer, consumer, and an inline thread that waits on the producer then it waits for the consumer to complete.

The producer thread starts sending data normally. When done, it uses the WaitGroup to signal that it is done sending to the channel. The inline goroutine waits on the producer WaitGroup which closes the channel. If the channel is never closed, the consumer would sleep forever waiting for more data and the process would never terminate.

When the consumer has no more data (because the channel has been closed), it notifies the second WaitGroup that it is done.

The main thread which launched the producer and consumer threads waits until the consumer WaitGroup allows it to complete. This prevents the main thread from terminating prematurely which would kill all the threads in the process.

This isn't the only way to implement the producer-consumer pattern.

There are also some issues such as external termination from signals such as SIGTERM and SIGINT that would need to be addressed for production code. This is a simple demonstration that shows the basics.

How else would you implement it? What is missing in the above implementation? Post your comments or links to other implementations below.

Thanks!

The code for this post and all posts in this series can be found here