Producer consumer

main.go

package main

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

var total_produced uint64 = 0
var total_consumed uint64 = 0

func main() {
	c := make(chan int64, 1000)
	w := &sync.WaitGroup{}
	pw := &sync.WaitGroup{}

	num_producers := 3
	num_consumers := 10

	for i := 0; i < num_consumers; i++ {
		go channel_consumer(c, w, i)
	}

	for i := 0; i < num_producers; i++ {
		go channel_producer(c, pw, i, int64(i+1))
	}

	time.Sleep(time.Millisecond * 10)
	go wait_for_producers(c, pw)

	w.Wait()

	fmt.Println("=======================================")
	fmt.Printf("%d producers: %d, %d consumers: %d\n",
		num_producers, total_produced,
		num_consumers, total_consumed)
}

func wait_for_producers(c chan int64, w *sync.WaitGroup) {
	w.Wait()
	close(c)
}

func channel_producer(c chan int64, w *sync.WaitGroup, id int, runForSeconds int64) {
	defer w.Done()
	w.Add(1)
	i := time.Now().Add(time.Second * time.Duration(runForSeconds)).Unix()
	for time.Now().Unix() < i {
		c <- rand.Int63()
		atomic.AddUint64(&total_produced, 1)
	}
}

func channel_consumer(c chan int64, w *sync.WaitGroup, id int) {
	defer w.Done()
	w.Add(1)
	for {
		_, ok := <-c
		if !ok {
			break
		}
		atomic.AddUint64(&total_consumed, 1)
	}
}