Многопоточность в Go: sync.Cond

Недавно я столкнулся с кейсом, где только одна горутина асинхронно получала определённые данные, а множество других должны были подождать, пока данные придут и затем их вы́читать.

Как синхронизировать доступ к общим данным?

Просто каналы использовать нельзя — тогда только одна горутина могла бы получить данные. Можно было бы создать глобальный лок и прокинуть его в каждую горутину. Это обеспечивает синхронизацию чтения, но понадобился бы ещё какой-то механизм оповещения, что данные получены и их можно использовать.

Конкретно в этом случае, решение нашлось в стандартной библиотеке. sync.Cond как раз и создан, чтобы заставить поток (или потоки) подождать события, после чего продолжить свою работу.

Для примера использования, создадим программу, где две горутины ждут (cond.Wait), пока третья запишет данные в общий словарь и сообщит об этом с помощью cond.Broadcast.

package main

import (
	"fmt"
	"os"
	"os/signal"
	"sync"
	"time"
)

func listen(name string, data map[string]string, c *sync.Cond) {
	c.L.Lock()
	c.Wait()

	fmt.Printf("[%s] %s\n", name, data["key"])

	c.L.Unlock()
}

func broadcast(name string, data map[string]string, c *sync.Cond) {
	time.Sleep(time.Second)

	c.L.Lock()

	data["key"] = "value"

	fmt.Printf("[%s] данные получены\n", name)

	c.Broadcast()
	c.L.Unlock()
}

func main() {
	data := map[string]string{}

	cond := sync.NewCond(&sync.Mutex{})

	go listen("слушатель 1", data, cond)
	go listen("слушатель 2", data, cond)

	go broadcast("источник", data, cond)

	ch := make(chan os.Signal, 1)
	signal.Notify(ch, os.Interrupt)
	<-ch
}

В консоль распечатается следующее:

В примере много чего происходит, давайте по пунктам:

• словари передаются по ссылке, и слушатели смогут прочитать данные, которые будут записаны в переданный словарь
• чтобы программа не завершилась до того, как горутины закончат работу, мы ждём сигнал os.Interrupt (ctrl + c в терминале)
• sync.NewCond принимает интерфейс sync.Locker (мьютекс удовлетворяет ему) и возвращает ссылку на инстанс sync.Cond.
• Внутри broadcast мы вызываем time.Sleep(time.Second), чтобы гарантировать, что cond.Broadcast будет вызван после вызова cond.Wait внутри listen
• Вызов cond.Wait обязательно нужно вызывать до вызова cond.Broadcast, иначе слушатели повиснут навсегда