singleflight в Go: как объединить одинаковые конкурентные запросы

singleflight в Go запускает одну функцию для одновременных вызовов с одинаковым ключом. Все участники получают её значение и ошибку. На примере загрузки товара разберём ключи, отмену ожидания и проверку числа обращений к источнику.

Когда одинаковая работа становится проблемой

HTTP-сервис хранит карточки товаров в базе, а часто используемые данные в кэше. Срок записи истёк. Почти одновременно пришло сто запросов на один популярный товар.

Каждый обработчик проверяет кэш и не находит запись. Затем каждый обращается в базу. Сто одинаковых запросов читают одни данные, хотя для ответа хватило бы одной загрузки. Такой всплеск после промаха кэша называют cache stampede.

Можно объединить обращения к одному товару: первый обработчик загружает данные, остальные получают тот же результат. Для этого в Go есть пакет golang.org/x/sync/singleflight. Он подходит и без кэша: например, когда несколько частей приложения одновременно запрашивают один внешний ресурс.

Сначала проверьте, действительно ли вызовы дублируются. Сто запросов к разным товарам представляют другую задачу. Если нужно ограничить число операций, используйте пул работников. Если нужно запустить несколько разных операций и собрать ошибки, посмотрите на errgroup. singleflight объединяет работу по совпадающему ключу.

Как работает singleflight.Group

Основной тип пакета называется Group. Его нулевое значение готово к использованию. Группа хранит сведения о текущих вызовах и различает их по строковым ключам.

Первый вызов регистрирует функцию для ключа. Пока функция выполняется, следующие вызовы с тем же ключом присоединяются к ней. Их собственные функции не запускаются. Все участники получают значение и ошибку первой функции. После завершения запись о вызове удаляется.

Схема показывает работу Do для одного ключа в одной группе. Ветвь «Нет» выполняет функцию, ветвь «Да» ждёт её завершения. Пунктир обозначает передачу готового результата ожидающему вызову. Он получает значение и ошибку первой функции, а свою функцию не запускает.

После завершения запись о вызове удаляется из группы. Новое обращение с тем же ключом снова попадёт в ветвь «Нет» и выполнит функцию.

Метод Do возвращает результат после завершения функции. DoChan сразу возвращает канал. Вызывающий код может выбрать в select между ответом и отменой контекста.

Для ключей product:42 и product:43 выполняются разные функции. Группа не ставит все операции в общую очередь. При этом число разных ключей пакет не ограничивает. Сервис всё ещё может перегрузить базу, если одновременно придут запросы на множество разных товаров.

Пакет находится вне стандартной библиотеки. Примеры ниже проверены на Go 1.26.9 с golang.org/x/sync v0.24.0. Эта версия модуля требует Go 1.26.0 или новее. В пустом каталоге создайте модуль и добавьте зависимость:

$ go mod init example.com/singleflight-demo
$ go get golang.org/x/sync@v0.24.0

Версия закреплена для воспроизводимости примеров. У проекта на более старом Go может быть другая версия зависимости. Проверяйте её go.mod, прежде чем обновлять пакет.

Пример: сто вызовов и одна загрузка

Сохраним программу в main.go. Вместо базы используем функцию со счётчиком. Канал release удерживает её, пока все сто вызовов не присоединятся к группе. Так результат не зависит от скорости запуска горутин.

package main

import (
	"fmt"
	"sync/atomic"

	"golang.org/x/sync/singleflight"
)

type Product struct {
	ID   int64
	Name string
}

func main() {
	var group singleflight.Group
	var calls atomic.Int64

	// Держим загрузку открытой, пока все вызовы не присоединятся.
	release := make(chan struct{})
	load := func() (any, error) {
		calls.Add(1)
		<-release
		return Product{ID: 42, Name: "Клавиатура"}, nil
	}

	const requests = 100
	results := make([]<-chan singleflight.Result, requests)
	for i := range results {
		results[i] = group.DoChan("product:42", load)
	}
	close(release)

	shared := 0
	for _, ch := range results {
		result := <-ch
		if result.Err != nil {
			panic(result.Err)
		}
		if result.Shared {
			shared++
		}
	}
	fmt.Printf("Запросов: %d, загрузок: %d, shared: %d\n",
		requests, calls.Load(), shared)

	// Завершённый вызов не сохраняется: функция запустится снова.
	_, _, _ = group.Do("product:42", load)
	fmt.Printf("Загрузок после следующего вызова: %d\n", calls.Load())
}

Запустим программу:

$ go run .
Запросов: 100, загрузок: 1, shared: 100
Загрузок после следующего вызова: 2

В цикле выполняется сто вызовов DoChan, но функция load запускается один раз. Закрытие release разрешает ей вернуть результат. Затем каждый участник получает своё сообщение с тем же значением.

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

Что означает shared

У Do третий результат называется shared. В ответе DoChan ему соответствует поле Result.Shared. Значение true означает, что результат разделили несколько вызовов.

Это свойство относится и к первому вызову. Если к нему кто-то присоединился, он тоже получит shared == true. Поэтому по этому полю нельзя определить, какая горутина выполнила загрузку. Для подсчёта обращений к источнику ставьте счётчик внутри переданной функции, как в примере.

Ошибку участники тоже получают общую. Если загрузка завершилась ошибкой, все присоединившиеся вызовы увидят эту ошибку. Следующее обращение после завершения снова запустит функцию. Пакет сам не сохраняет неудачу и не вводит паузу перед повтором.

Когда удобнее Do

Если отменять ожидание не нужно, Do короче. Следующий фрагмент использует переменную group и функцию load из примера выше:

value, err, shared := group.Do("product:42", load)
if err != nil {
    return
}
product := value.(Product)
fmt.Println(product.ID, shared)

Пакет возвращает значение типа any. В нашем примере функция всегда возвращает Product, поэтому приведение к этому типу безопасно. В большом приложении лучше спрятать группу за методом с конкретным возвращаемым типом. Тогда вызывающий код не будет повторять приведение и соглашение о ключах.

Ключ определяет, какие запросы можно объединять

Одинаковый ключ должен означать одинаковую работу и взаимозаменяемый результат. Это часть контракта приложения. singleflight не проверяет, действительно ли два запроса равнозначны.

Для публичной карточки товара может хватить идентификатора. Но персональная цена зависит ещё и от покупателя. Если объединить такие запросы только по товару, один пользователь может получить цену другого. Сначала определите все входные данные, влияющие на результат, затем составьте ключ.

Проверка доступа тоже должна сохраняться. Удобнее выполнять её до получения общих публичных данных, а персональные поля добавлять отдельно. Если сама загрузка учитывает права, ключ должен различать соответствующие контексты доступа.

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

Группа должна жить дольше запроса

Группу обычно хранят в структуре сервиса или загрузчика. Если создавать новую Group внутри каждого HTTP-обработчика, обращения окажутся в разных группах и не объединятся.

Полезно выделить отдельную группу для каждого вида результата. Тогда ключ 42 внутри загрузчика товаров относится только к товару. Общая группа для товаров и пользователей потребует пространства имён, например product:42 и user:42. Иначе столкнутся не только данные, но и ожидаемые типы результата.

Область объединения ограничена конкретным экземпляром Group. Две группы в одном процессе независимы. Отдельные процессы тем более ничего не знают друг о друге. Если у сервиса десять реплик, один ключ может одновременно загружаться на каждой из них.

Поэтому singleflight уменьшает дублирование внутри экземпляра приложения. Для координации между репликами потребуется отдельное решение. Оценивать его стоит по реальной нагрузке: иногда одной загрузки на каждую реплику вполне достаточно.

DoChan: отменяем ожидание, а не общую загрузку

У клиента может закончиться время ожидания, пока источник ещё работает. Do не принимает контекст и сам не умеет возвращаться по его отмене. У DoChan можно ждать канал результата в select рядом с ctx.Done().

При этом есть два разных срока жизни. Первый относится к HTTP-запросу: сколько готов ждать конкретный клиент. Второй относится к общей загрузке: сколько разрешено работать источнику для всех ожидающих.

Если передать в общую функцию контекст первого клиента, его уход может прервать работу для остальных. Например, первый запрос отключился через 50 миллисекунд, а другие готовы ждать секунду. Но все получат ошибку, если обращение к источнику завершится по отмене первого контекста.

Ниже загрузчик, который разделяет эти сроки. Тип Product уже объявлен в main.go. Добавьте рядом файл loader.go:

package main

import (
	"context"
	"strconv"
	"time"

	"golang.org/x/sync/singleflight"
)

type ProductLoader struct {
	group        singleflight.Group
	operationCtx context.Context
	timeout      time.Duration
	load         func(context.Context, int64) (Product, error)
}

func (l *ProductLoader) Get(ctx context.Context, id int64) (Product, error) {
	if err := ctx.Err(); err != nil {
		return Product{}, err
	}

	key := strconv.FormatInt(id, 10)
	ch := l.group.DoChan(key, func() (any, error) {
		// Общая загрузка зависит от жизни сервиса, а не первого клиента.
		opCtx, cancel := context.WithTimeout(l.operationCtx, l.timeout)
		defer cancel()
		return l.load(opCtx, id)
	})

	select {
	case <-ctx.Done():
		return Product{}, ctx.Err()
	case result := <-ch:
		if result.Err != nil {
			return Product{}, result.Err
		}
		return result.Val.(Product), nil
	}
}

Здесь operationCtx задаёт время жизни сервиса. Его передают при создании загрузчика и отменяют при остановке приложения. Поле timeout ограничивает длительность каждой общей загрузки. Функция load должна учитывать переданный ей контекст, например передавать его в запрос к базе или HTTP-клиенту.

При создании ProductLoader нужно заполнить все три поля: operationCtx, timeout и load. Это минимальный пример устройства загрузчика, без конструктора и подключения конкретной базы. Пустое значение этой структуры к работе не готово.

Метод Get сначала проверяет, не отменён ли запрос. Затем регистрирует ожидание и выбирает между ответом и отменой. Если клиент ушёл, метод возвращает его ошибку. Остальные участники продолжают ждать общую операцию.

Сам пакет не отменяет загрузку, когда ушли все ожидающие. В нашем примере она завершится по ответу источника, собственному таймауту или отмене operationCtx. Ограничение общей операции нужно и тогда, когда каждый вызывающий умеет прерывать ожидание.

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

Две особенности ожидания

Канал, возвращённый DoChan, не закрывается. Из него читают один результат. Цикл for range после первого сообщения останется ждать следующие значения.

У каждого вызова DoChan свой канал с буфером на один результат. Если клиент перестал ждать, отправка его ответа не требует получателя. Но память и общая операция освободятся только по мере завершения работы. Эта особенность не заменяет контроль сроков жизни.

Есть ещё обычное свойство select: если результат и отмена готовы одновременно, может быть выбрана любая ветвь. Наш Get допускает оба исхода. Если API требует всегда предпочитать отмену, это отдельное правило, которое нужно реализовать и проверить.

Механизм context.Context передаёт сигнал отмены. Код операции должен на него реагировать. Ни context, ни singleflight не останавливают произвольную горутину принудительно.

Как совместить singleflight с кэшем

В конце основной программы счётчик вырос до двух. Это ожидаемое поведение: завершённый результат не остаётся в группе. Если нужна повторная выдача результата без загрузки, добавьте отдельный кэш.

Обычная последовательность такая: проверить кэш; при промахе войти в singleflight; внутри функции ещё раз проверить кэш; если запись всё ещё отсутствует, загрузить данные и сохранить их. Ключи кэша и группы должны одинаково описывать результат.

Повторная проверка закрывает промежуток между первой проверкой и регистрацией вызова. За это время другая операция могла уже загрузить запись и завершиться. В группе её больше нет, а в кэше данные есть. Без второй проверки новая функция зря обратится к источнику.

Здесь возникает решение об ошибках. singleflight передаст ошибку всем участникам текущего вызова, но следующая группа начнёт работу заново. При недоступном источнике может получиться серия повторных загрузок. Нужны ли пауза, ограничение повторов или короткое хранение ошибки, решает приложение.

Уточните и требование к свежести данных. Новый HTTP-запрос может присоединиться к загрузке, которая началась чуть раньше. Для карточки товара это часто допустимо. Для операции, которая должна читать состояние строго после конкретного обновления, одного совпадения ключа может оказаться недостаточно.

Общий результат не становится независимой копией

Пакет передаёт участникам значение, возвращённое функцией. Он не создаёт глубокую копию для каждого ожидающего. В примере Product содержит число и строку, но в реальном результате могут быть указатели, срезы и карты.

Если один участник начнёт менять общую карту, это затронет остальных и может привести к гонке. Договоритесь, что результат доступен только для чтения, либо создавайте нужные копии на границе загрузчика. Не возвращайте объект в пул, пока другие участники ещё могут им пользоваться.

Какие операции объединять

Начать проще с чтения: получить карточку товара, настройки сервиса или публичные данные внешнего API. Для этих операций несколько клиентов могут использовать один ответ. Даже здесь проверьте, совпадает ли требование к свежести результата.

С записью нужно осторожнее. Два запроса на создание заказа с одним покупателем могут означать две покупки. Если использовать идентификатор покупателя как ключ, одна функция создаст один заказ, а оба вызова получат его результат. Пакет выполнил своё соглашение, но приложение потеряло отдельную операцию.

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

Для защиты от повторной отправки одной команды обычно нужен идентификатор операции и правило идемпотентности на стороне хранилища. singleflight может дополнить такую схему внутри процесса. Но после завершения функции, перезапуска сервиса или обращения к другой реплике его запись не защитит от повторного выполнения.

Общий ключ также объединяет ошибки. Если права или допустимость операции зависят от конкретного пользователя, один отказ не должен автоматически становиться ответом для другого. Проверка ключей и полномочий нужна до того, как функция попадёт в группу.

Forget разрешает новую загрузку

Метод Forget(key) удаляет регистрацию ключа из группы. Следующее обращение сможет запустить новую функцию, даже если старая ещё выполняется. Участники старого вызова продолжат ждать его результат.

Поэтому Forget не останавливает работу и не является обычным способом отменить запрос. Если вызвать его при каждом клиентском таймауте, можно снова получить несколько одновременных загрузок одного ключа.

Метод нужен, когда приложение сознательно разрешает начать работу заново. Но он сам не решает вопрос свежести. Старая загрузка может завершиться после новой и записать устаревшие данные в кэш. Для такого сценария отдельно определяют порядок записи, версии значений или другой способ проверки результата.

Как проверить объединение вызовов

Тест должен гарантировать, что вызовы действительно пересеклись во времени. Если просто запустить сто горутин, короткая функция может успеть завершиться до регистрации остальных. Тогда несколько загрузок будут корректным поведением, а тест окажется нестабильным.

В примере мы удерживаем функцию каналом. DoChan регистрирует ожидание до возврата, поэтому после заполнения среза все участники уже относятся к одному вызову. Только затем закрываем release.

Минимальный тест для singleflight_test.go выглядит так:

package main

import (
    "sync/atomic"
    "testing"

    "golang.org/x/sync/singleflight"
)

func TestSameKey(t *testing.T) {
	var group singleflight.Group
	var calls atomic.Int64
	release := make(chan struct{})
	load := func() (any, error) {
		calls.Add(1)
		<-release
		return "result", nil
	}

	results := make([]<-chan singleflight.Result, 100)
	for i := range results {
		results[i] = group.DoChan("same-key", load)
	}
	close(release)

	for _, ch := range results {
		result := <-ch
		if result.Err != nil || result.Val != "result" || !result.Shared {
			t.Fatalf("неверный результат: %+v", result)
		}
	}
	if got := calls.Load(); got != 1 {
		t.Fatalf("вызовов: %d, нужно 1", got)
	}

	_, _, shared := group.Do("same-key", load)
	if got := calls.Load(); got != 2 || shared {
		t.Fatalf("после завершения: вызовов=%d, shared=%t", got, shared)
	}
}

Запуск:

$ go test -race ./...

Проверяйте и границы: разные ключи должны запускать разные функции; новый вызов после завершения должен загружать данные снова; отмена одного ожидающего не должна случайно отменять общий источник. Для Forget нужен сценарий, где старая и новая функции некоторое время работают одновременно.

В приложении отдельно считайте входящие запросы и реальные обращения к источнику. Одно поле shared не показывает, насколько уменьшилась нагрузка. Сравните также время ожидания: участники общей операции зависят от её длительности и получат одну ошибку, если она завершится неудачно.

Если после добавления загрузчика растёт число ожидающих горутин, разберите сроки работы источника и отмену. Для поиска зависших вызовов пригодится goroutine-профиль pprof. Начать стоит с жизненного цикла операции, а не с увеличения числа горутин.

Итог

singleflight.Group объединяет выполняющиеся вызовы с одинаковым ключом. Храните группу в загрузчике, задавайте ключ по всем входным данным и отдельно ограничивайте время общей операции.

Do подходит для обычного ожидания. DoChan позволяет клиенту уйти раньше. Для хранения завершённых результатов нужен кэш, а для ограничения разных операций отдельный механизм управления нагрузкой.