inanzzz | A simple worker and work queue example with Golang

In this example we are going to create a "buffered" channel that has the capacity of maximum 11 resource slot. If more than 11 jobs given, it refuses them. We will have only one worker/goroutine running in order to process the jobs. The application will block until all jobs are completed so no job is lost.

Improvements

Although this is a "basic" example, it is important to consider some additional features such as:

Terminating long running jobs.

Running multiple workers.

Preventing channels to accept more jobs while gracefully stopping workers.

Killing all workers.

Example

There is also a good example here.

package main

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

var (
	// wg is used to force the application wait for all goroutines to finish before exiting.
	wg sync.WaitGroup
	// jobChan is a buffered channel that has the capacity of maximum 11 resource slot.
	jobChan = make(chan int, 11)
	// waiters is used to make goroutines sleep in order to simulate time it took to process the job.
	waiters = []int{0, 1, 2, 3, 4}
)

func main() {
	rand.Seed(time.Now().UnixNano())

	fmt.Println("BEGIN")

	// Tell how many worker you will be running.
	wg.Add(1)

	// Run 1 worker to handle jobs.
	go worker(jobChan, &wg)

	// Send 10 jobs to job channel.
	for i := 1; i <= 10; i++ {
		if !queueJob(i, jobChan) {
			fmt.Println("Channel is full... Service unavailable...")
		}
	}

	// Close the job channel.
	close(jobChan)

	// Block exiting until all the goroutines are finished.
	wg.Wait()

	fmt.Println("END")
}

// queueJob puts job into channel. If channel buffer is full, return false.
func queueJob(job int, jobChan chan<- int) bool {
	select {
	case jobChan <- job:
		return true
	default:
		return false
	}
}

// worker processes jobs.
func worker(jobChan <-chan int, wg *sync.WaitGroup) {
	// As soon as the current goroutine finishes (job done!), notify back WaitGroup.
	defer wg.Done()

	fmt.Println("Worker is waiting for jobs")

	for job := range jobChan {
		fmt.Println("Worker picked job", job)

		wait := time.Duration(rand.Intn(len(waiters)))
		time.Sleep(wait * time.Second)

		// Process the job ...

		fmt.Println("Worker completed job", job, "in", int(wait), "second(s)")
	}
}

Test

BEGIN
Worker is waiting for jobs
Worker picked job 1
Worker completed job 1 in 4 second(s)
Worker picked job 2
Worker completed job 2 in 1 second(s)
Worker picked job 3
Worker completed job 3 in 4 second(s)
Worker picked job 4
Worker completed job 4 in 3 second(s)
Worker picked job 5
Worker completed job 5 in 2 second(s)
Worker picked job 6
Worker completed job 6 in 1 second(s)
Worker picked job 7
Worker completed job 7 in 1 second(s)
Worker picked job 8
Worker completed job 8 in 2 second(s)
Worker picked job 9
Worker completed job 9 in 1 second(s)
Worker picked job 10
Worker completed job 10 in 2 second(s)
END

If the job amount is above the channel limit, you will see some jobs are being refused. For example if you send 15 jobs to channel limited to 11 jobs, you should see something like below where 4 jobs are refused.

BEGIN
Channel is full... Service unavailable...
Channel is full... Service unavailable...
Channel is full... Service unavailable...
Channel is full... Service unavailable...
...
...