Channels and a Worker Pool

Goal

By the end of this lesson, you can send values between goroutines with channels, wait for several things at once with select, and build a worker pool, the classic Go pattern for processing many jobs with a fixed number of goroutines. You use the pool in a new command, tasker import, which reads several JSON files at the same time.

What you will learn

• How to create channels, send values, and receive values
• Why sends and receives block, and how that synchronizes goroutines
• Buffered and unbuffered channels
• How to close a channel and loop over it with range
• How to wait for the first of several events with select, including timeouts
• How to build a worker pool with a jobs channel and a results channel
• How to keep results in the right order when work finishes in random order

Why this matters

In Lesson 6.1, goroutines shared memory and protected it with a mutex. Go has a second way for goroutines to work together, and Go programmers often prefer it. A famous Go saying describes it:

Do not communicate by sharing memory; instead, share memory by communicating.

A channel is a pipe between goroutines. One goroutine sends a value into it, and another receives it. Only one goroutine has the value at a time, so there is nothing to lock. Channels are the basis of many patterns you will use in backends: worker pools, pipelines, timeouts, and graceful shutdown.

Step 1: Send and receive

Create cmd/play/main.go:

package main

import (
	"fmt"
	"time"
)

func main() {
	messages := make(chan string)

	go func() {
		time.Sleep(100 * time.Millisecond)
		messages <- "hello from a goroutine"
	}()

	msg := <-messages
	fmt.Println(msg)
}

Output:

hello from a goroutine

• make(chan string) creates a channel that carries string values.
• messages <- "..." sends a value into the channel. The arrow points into the channel.
• msg := <-messages receives a value from the channel. The arrow points out of the channel.

Notice that main does not use a WaitGroup, but it still waits for the goroutine. A receive blocks: it waits until a value arrives. A send on this kind of channel also blocks until another goroutine receives. So a channel both moves data and synchronizes the two goroutines.

Step 2: Close a channel and loop over it

Add this to main:

	numbers := make(chan int)
	go func() {
		for i := 1; i <= 3; i++ {
			numbers <- i * 10
		}
		close(numbers)
	}()
	for n := range numbers {
		fmt.Println("received", n)
	}

Output:

received 10
received 20
received 30

close(numbers) says "no more values will come". for n := range numbers receives values until the channel is closed and empty, then the loop ends.

Rules for closing:

• Only the sender closes a channel, never the receiver. The sender knows when there is no more data.
• Sending on a closed channel causes a panic.
• Receiving from a closed channel returns the zero value at once. The comma ok form tells you if the channel is closed:

	v, ok := <-numbers
	fmt.Println(v, ok)

0 false

You do not have to close every channel. Close a channel only when the receiver needs to know that no more values will come, for example to end a range loop.

Step 3: Buffered channels

	buffered := make(chan string, 2)
	buffered <- "first"
	buffered <- "second"
	fmt.Println(len(buffered), cap(buffered))
	fmt.Println(<-buffered, <-buffered)

2 2
first second

The second argument of make is the buffer size. A buffered channel can hold that many values without a receiver. Sends block only when the buffer is full, and receives block only when it is empty. Values come out in the same order they went in.

An unbuffered channel (make(chan T)) has buffer size 0. Every send waits for a receive. Start with unbuffered channels. Add a buffer only when you have a clear reason.

Step 4: A deadlock with channels

Replace main for a moment with this:

func main() {
	ch := make(chan int)
	ch <- 1
}

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()

The send waits for a receiver, but there is no other goroutine. The program can never continue. When every goroutine is blocked, the Go runtime detects it and stops the program with this message. Put the previous code back.

Step 5: select and timeouts

select waits on several channel operations and runs the first one that is ready. Add this:

	slow := make(chan string)
	go func() {
		time.Sleep(2 * time.Second)
		slow <- "finally"
	}()
	select {
	case msg := <-slow:
		fmt.Println(msg)
	case <-time.After(500 * time.Millisecond):
		fmt.Println("timeout: gave up after 500ms")
	}

timeout: gave up after 500ms

time.After(d) returns a channel that receives a value after the duration d. The select waits for whichever comes first: the slow result or the timer. The timer wins, so the program stops waiting after 500 milliseconds.

This is how Go programs implement timeouts. In Lesson 6.3, you use context, which builds on the same idea and is the standard way to cancel work in backends.

Delete the cmd/play folder when you finish experimenting.

Step 6: The worker pool pattern

Imagine a user wants to import 100 task files. Reading them one by one is slow. Starting 100 goroutines at once could open too many files at the same time. A worker pool is the middle way:

1. Put all jobs into a channel.
2. Start a fixed number of workers. Each worker takes a job from the channel, does the work, and sends the result to a second channel.
3. Collect all results.

The number of workers limits how much work happens at the same time.

Step 7: Build the importer

An import file contains a list of tasks:

[{"title": "Buy milk", "tags": ["shop"]}, {"title": "Pay rent"}]

Create internal/importer/importer.go:

// Package importer reads task lists from several sources at the same time.
package importer

import (
	"encoding/json"
	"fmt"
	"os"
	"sync"
)

// Item is one task in an import file.
type Item struct {
	Title string   `json:"title"`
	Tags  []string `json:"tags"`
}

// Result is the outcome of reading one source.
type Result struct {
	Source string
	Items  []Item
	Err    error
}

// ReadFiles reads all files with at most workers files open at the same time.
// The results are in the same order as paths.
func ReadFiles(paths []string, workers int) []Result {
	type job struct {
		index int
		path  string
	}
	type indexedResult struct {
		index  int
		result Result
	}

	jobs := make(chan job)
	results := make(chan indexedResult)

	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs {
				results <- indexedResult{index: j.index, result: readFile(j.path)}
			}
		}()
	}

	go func() {
		for i, p := range paths {
			jobs <- job{index: i, path: p}
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	out := make([]Result, len(paths))
	for r := range results {
		out[r.index] = r.result
	}
	return out
}

// readFile reads one JSON file with a list of items.
func readFile(path string) Result {
	res := Result{Source: path}

	data, err := os.ReadFile(path)
	if err != nil {
		res.Err = err
		return res
	}
	if err := json.Unmarshal(data, &res.Items); err != nil {
		res.Err = fmt.Errorf("parse %s: %w", path, err)
	}
	return res
}

ReadFiles has four parts, and each one has one job:

1. The workers. workers goroutines each run for j := range jobs. They take jobs until the jobs channel is closed, then they stop.
2. The producer. One goroutine sends every path into jobs, then closes it. It runs in its own goroutine because jobs is unbuffered: each send waits for a free worker.
3. The closer. One goroutine waits until all workers are done, then closes results. Without it, the range results loop below would wait forever.
4. The collector. main's own goroutine receives every result.

Errors are values, here too. A file that fails does not stop the others. Its error travels inside its Result, and the caller decides what to do.

Keeping the order. Workers finish in random order, as you saw in Lesson 6.1. Each job carries its index in the input, and the collector puts each result at that index in out. The output order always matches the input, so the command prints a predictable report.

type job struct inside the function is a local type. Only ReadFiles needs it, so it lives there.

Who closes what? Each channel is closed by the goroutine that sends on it, after its last send: the producer closes jobs, and the closer closes results after all workers, the senders, have finished. Follow this rule, and you never send on a closed channel.

Step 8: Test the importer

Create internal/importer/importer_test.go:

package importer

import (
	"os"
	"path/filepath"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func writeFile(t *testing.T, dir, name, content string) string {
	t.Helper()
	path := filepath.Join(dir, name)
	require.NoError(t, os.WriteFile(path, []byte(content), 0o644))
	return path
}

func TestReadFiles(t *testing.T) {
	dir := t.TempDir()
	a := writeFile(t, dir, "a.json", `[{"title": "Buy milk", "tags": ["shop"]}]`)
	b := writeFile(t, dir, "b.json", `[{"title": "Pay rent"}, {"title": "Call Ana"}]`)
	broken := writeFile(t, dir, "broken.json", `[{"title":`)
	missing := filepath.Join(dir, "missing.json")

	results := ReadFiles([]string{a, b, broken, missing}, 2)

	require.Len(t, results, 4)

	assert.Equal(t, a, results[0].Source)
	assert.NoError(t, results[0].Err)
	assert.Equal(t, []Item{{Title: "Buy milk", Tags: []string{"shop"}}}, results[0].Items)

	assert.NoError(t, results[1].Err)
	assert.Len(t, results[1].Items, 2)

	assert.ErrorContains(t, results[2].Err, "parse")
	assert.ErrorIs(t, results[3].Err, os.ErrNotExist)
}

func TestReadFilesMoreWorkersThanFiles(t *testing.T) {
	path := writeFile(t, t.TempDir(), "a.json", `[{"title": "Only one"}]`)

	results := ReadFiles([]string{path}, 10)

	require.Len(t, results, 1)
	assert.NoError(t, results[0].Err)
}

func TestReadFilesNoFiles(t *testing.T) {
	results := ReadFiles(nil, 4)

	assert.Empty(t, results)
}

The tests cover the normal case, a broken file, a missing file, more workers than files, and no files at all. The last two cases are where pool code often hangs or crashes, so they are worth a test. Run them with the race detector:

go test -race ./internal/importer

Step 9: The import command

Create internal/cli/import.go:

package cli

import (
	"errors"
	"fmt"

	"github.com/spf13/cobra"

	"github.com/yourname/tasker/internal/importer"
)

func newImportCmd(opts *options) *cobra.Command {
	var workers int

	cmd := &cobra.Command{
		Use:   "import <file>...",
		Short: "Import tasks from JSON files",
		Long: `Import tasks from one or more JSON files.

Each file contains a list of tasks, for example:
  [{"title": "Buy milk", "tags": ["shop"]}, {"title": "Pay rent"}]

Tasker reads several files at the same time.`,
		Args: cobra.MinimumNArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			if workers < 1 {
				return errors.New("--workers must be at least 1")
			}

			store, err := opts.openStore()
			if err != nil {
				return err
			}

			out := cmd.OutOrStdout()
			failed := 0
			for _, res := range importer.ReadFiles(args, workers) {
				if res.Err != nil {
					fmt.Fprintf(cmd.ErrOrStderr(), "error: %v\n", res.Err)
					failed++
					continue
				}
				for _, item := range res.Items {
					if _, err := store.Add(item.Title, item.Tags); err != nil {
						return fmt.Errorf("%s: %q: %w", res.Source, item.Title, err)
					}
				}
				fmt.Fprintf(out, "Imported %d tasks from %s\n", len(res.Items), res.Source)
			}

			if failed > 0 {
				return fmt.Errorf("%d of %d files could not be imported", failed, len(args))
			}
			return nil
		},
	}

	cmd.Flags().IntVarP(&workers, "workers", "w", 4, "number of files to read at the same time")
	return cmd
}

Register it in root.go by adding newImportCmd(opts), to the AddCommand list.

Files are read concurrently, but tasks are added one by one in the main goroutine, in input order. The store is now safe for concurrent use, so adding from the workers would work too. But adding in order gives every user the same IDs for the same input, which is easier to understand and to test.

The command imports every good file and reports every bad one. Then it fails with a summary, so scripts see exit code 1 and know that something went wrong.

Step 10: Try it

Create some files:

echo '[{"title": "Buy milk", "tags": ["shop"]}, {"title": "Pay rent"}]' > home.json
echo '[{"title": "Write report", "tags": ["work"]}]' > work.json
echo '[{"title":' > broken.json

Build and import:

go build ./cmd/tasker
./tasker import home.json work.json broken.json missing.json

Imported 2 tasks from home.json
Imported 1 tasks from work.json
error: parse broken.json: unexpected end of JSON input
error: open missing.json: no such file or directory
error: 2 of 4 files could not be imported

./tasker list

1. [ ] Buy milk #shop
2. [ ] Pay rent
3. [ ] Write report #work

The output follows the order of the arguments, even though the workers finished in any order.

Why it works: each goroutine owns its job

In the worker pool, no two goroutines ever touch the same value at the same time:

• A job belongs to the producer until it sends it, then to exactly one worker.
• A result belongs to its worker until it sends it, then to the collector.
• out is only written by the collector.

There is no shared memory, so there is no mutex. The channels transfer ownership from one goroutine to the next. This is what "share memory by communicating" means.

Use a mutex when several goroutines need the same data, like the store. Use channels when data flows from one goroutine to another, like jobs and results.

Exercises

1. Grammar. "Imported 1 tasks" is wrong English. Print "task" for 1 and "tasks" for every other number.
2. Count in parallel. Write a function CountItems(paths []string, workers int) int that uses the same pool pattern but sends only a number for each file through the results channel, and returns the total.
3. A buffered jobs channel. Change jobs to make(chan job, len(paths)). Can you now remove the producer goroutine and send the jobs directly in ReadFiles? Why?
4. Progress. Add a progress channel to the pool that receives the source name each time a file finishes. Print done: a.json lines from the command as files finish. Do they come in input order?
5. Find the hang. Remove the goroutine that closes results, and run go test ./internal/importer. What happens, and why? Use -timeout 5s to stop it. Put the code back.

Checkpoint

You are ready for the next lesson if you can:

• Create a channel, send to it, and receive from it
• Explain why a send on an unbuffered channel blocks
• Close a channel and range over it, and say who should close it
• Use select with time.After to stop waiting after a timeout
• Build a worker pool with jobs and results channels
• Keep results in input order when workers finish in random order