Example code to buffer concurrent long running task

job_buffering.go

package main

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

type job struct {
	Name string
}

func Process(j job) {
  // randomize the time to process the job
	sleep := 1000 + rand.Intn(1000)
	time.Sleep(time.Duration(sleep) * time.Millisecond)
  
	fmt.Printf("processing %s\n", j.Name)
}

func worker(j job) {
	Process(j)
}

func main() {
	var total = 1000
	var maxRunning = int64(50)
	var running int64
	var wg sync.WaitGroup

	var (
		goCh   = make(chan interface{})
		stopCh = make(chan interface{})
	)

	for i := 0; i < total; i++ {
		var idx = i
		wg.Add(1)
		go func() {
			var currentRunning = atomic.AddInt64(&running, 1)
			if currentRunning == maxRunning {
				stopCh <- ""
			} else {
				goCh <- ""
			}
			worker(job{
				Name: fmt.Sprintf("process-%s", strconv.Itoa(idx)),
			})
			if currentRunning == maxRunning {
				goCh <- ""
			}
			currentRunning = atomic.AddInt64(&running, -1)
			wg.Done()
		}()

		select {
		case <-goCh:
		case <-stopCh:
			<-goCh
		}

	}

	wg.Wait()

}