worker_goroutines.go

package main

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

var rng = rand.New(rand.NewSource(time.Now().UnixNano()))

func main() {
	in, out := Start[int](10)

	go func() {
		// Produce random numbers.
		for i := 0; i < 1000; i++ {
			in <- func() (int, error) {
				return rng.Intn(100), nil
			}
		}

		close(in)
	}()

	// Consume the produced numbers and sum them up.
	sum := 0
	for res := range out {
		if res.Err != nil {
			log.Fatalf("Error: %v", res.Err)
		}

		sum += res.Value
	}

	fmt.Printf("Total: %d\n", sum)
}

type Result[T any] struct {
	Value T
	Err   error
}

type In[T any] chan<- func() (T, error)
type Out[T any] <-chan Result[T]

// Start creates a worker pool consisting of the given number of concurrent workers that will call
// functions from the returned input channel.
// The result of the function call will be sent to the returned output channel.
func Start[T any](concurrency int) (In[T], Out[T]) {
	in := make(chan func() (T, error))
	out := make(chan Result[T])

	var wg sync.WaitGroup
	wg.Add(concurrency)
	for i := 0; i < concurrency; i++ {
		// Start a worker in a goroutine.
		go func() {
			for f := range in {
				v, err := f()

				out <- Result[T]{Value: v, Err: err}
			}
			wg.Done()
		}()
	}

	// Start an extra goroutine that is responsible to close the output channel
	// when the processing has finished.
	go func() {
		wg.Wait()
		close(out)
	}()

	return in, out
}