Пакетная обработка в Go: рабочие приёмы работы с батчами

Задача звучит просто: приходят события по одному, а записывать их в базу или отправлять в API надо пачками — так дешевле и быстрее. Наивное решение «накопил 100 штук в слайс и слил» разваливается на трёх вопросах: что делать с остатком, если событий пришло 97 и поток замолчал; как не держать данные в буфере минутами при редком трафике; как не превратить один батч в блокировку всего конвейера. Разберём рабочие приёмы для Go 1.22+ — с каналами, таймерами и контролем параллельности.

Почему «накопил и слил» не работает

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

// Go 1.22 — наивный вариант, у него две дыры
func processNaive(in <-chan Event, batchSize int, flush func([]Event)) {
    batch := make([]Event, 0, batchSize)
    for event := range in {
        batch = append(batch, event)
        if len(batch) >= batchSize {
            flush(batch)
            batch = batch[:0]
        }
    }
    // а вот про этот остаток забывают
}

Дыра первая: если канал закрыли, а в batch лежит 40 элементов из 100 — они пропадут. Дыра вторая: пока не набралось batchSize, данные висят в памяти сколько угодно долго. При потоке в одно событие в минуту первый батч уйдёт через час с лишним. Для метрик или логов это неприемлемо.

Приём первый: батч по размеру плюс остаток

Первое, что чинится, — потеря хвоста. После выхода из range нужно слить то, что осталось.

func processWithTail(in <-chan Event, batchSize int, flush func([]Event)) {
    batch := make([]Event, 0, batchSize)
    for event := range in {
        batch = append(batch, event)
        if len(batch) >= batchSize {
            flush(batch)
            batch = batch[:0]
        }
    }
    // сливаем неполный последний пакет после закрытия канала
    if len(batch) > 0 {
        flush(batch)
    }
}

Одна деталь про batch[:0]: слайс переиспользуется, а flush получает его же нижележащий массив. Если flush сохраняет слайс куда-то асинхронно (пишет в другой канал, кладёт в структуру), следующая итерация затрёт данные. Тогда передавайте копию: flush(append([]Event(nil), batch...)). Если flush синхронный и полностью отрабатывает до возврата — переиспользование безопасно и экономит аллокации.

Приём второй: батч по размеру ИЛИ по таймеру

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

import "time"

func processWithTimer(in <-chan Event, batchSize int, maxWait time.Duration, flush func([]Event)) {
    batch := make([]Event, 0, batchSize)
    ticker := time.NewTicker(maxWait)
    defer ticker.Stop()

    sendBatch := func() {
        if len(batch) == 0 {
            return
        }
        flush(batch)
        batch = batch[:0]
        ticker.Reset(maxWait) // отсчёт заново после отправки
    }

    for {
        select {
        case event, ok := <-in:
            if !ok {
                sendBatch() // канал закрыт — сливаем остаток
                return
            }
            batch = append(batch, event)
            if len(batch) >= batchSize {
                sendBatch()
            }
        case <-ticker.C:
            sendBatch()
        }
    }
}

Здесь важен ticker.Reset(maxWait) после отправки по размеру. Без него таймер может сработать сразу после того, как батч ушёл по достижению лимита, и вы отправите пустой или почти пустой пакет через долю секунды. sendBatch с проверкой len(batch) == 0 защищает от пустых отправок, но сброс таймера всё равно нужен, чтобы интервал отсчитывался от момента реальной отправки.

Батч по таймеру — это компромисс между задержкой и эффективностью. Меньше интервал — свежее данные, но больше мелких пакетов и накладных расходов. Больше интервал — крупнее пакеты, но данные стареют. Цифру подбирают под SLA приёмника, а не «на глаз».

Приём третий: несколько воркеров на отправку

Пока flush синхронный, весь конвейер стоит на время отправки батча. Если flush — это запрос в базу на 200 мс, а батчи набираются каждые 50 мс, чтение из канала встанет. Решение — вынести отправку в пул воркеров через отдельный канал батчей.

func processPipeline(in <-chan Event, batchSize int, maxWait time.Duration, workers int, flush func([]Event)) {
    batches := make(chan []Event, workers) // буфер по числу воркеров

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for batch := range batches {
                flush(batch)
            }
        }()
    }

    batch := make([]Event, 0, batchSize)
    ticker := time.NewTicker(maxWait)
    defer ticker.Stop()

    sendBatch := func() {
        if len(batch) == 0 {
            return
        }
        batches <- batch          // передаём воркеру
        batch = make([]Event, 0, batchSize) // новый слайс, старый ушёл в воркер
        ticker.Reset(maxWait)
    }

    for {
        select {
        case event, ok := <-in:
            if !ok {
                sendBatch()
                close(batches)
                wg.Wait() // ждём, пока воркеры дожуют
                return
            }
            batch = append(batch, event)
            if len(batch) >= batchSize {
                sendBatch()
            }
        case <-ticker.C:
            sendBatch()
        }
    }
}

Здесь batch[:0] уже нельзя — слайс уходит воркеру и живёт своей жизнью, поэтому после отправки выделяем новый через make. Это плата за параллельность: аллокация на каждый батч. Буфер канала batches размером с число воркеров сглаживает всплески, но не делает очередь бесконечной — если приёмник стабильно медленнее источника, канал заполнится и чтение всё равно затормозит. Это правильное поведение: обратное давление лучше, чем разрастание памяти.

Порядок при нескольких воркерах

С workers > 1 вы теряете гарантию порядка отправки батчей: воркер, взявший второй пакет, может закончить раньше первого. Если приёмник требует строгого порядка (например, событий по одному ключу), либо оставляйте один воркер, либо шардируйте поток по ключу — тогда каждый воркер обрабатывает свой набор ключей и порядок внутри ключа сохраняется.

Приём четвёртый: отмена через context

В боевом сервисе конвейер должен корректно останавливаться по сигналу — при graceful shutdown нужно слить накопленное, а не бросить его. Для этого добавляют context.Context в select.

import "context"

func processWithContext(ctx context.Context, in <-chan Event, batchSize int, flush func([]Event)) {
    batch := make([]Event, 0, batchSize)
    for {
        select {
        case <-ctx.Done():
            if len(batch) > 0 {
                flush(batch) // финальный слив перед выходом
            }
            return
        case event, ok := <-in:
            if !ok {
                if len(batch) > 0 {
                    flush(batch)
                }
                return
            }
            batch = append(batch, event)
            if len(batch) >= batchSize {
                flush(batch)
                batch = batch[:0]
            }
        }
    }
}

Тонкость: при отмене финальный flush сам должен получить контекст с таймаутом, иначе на «мёртвом» контексте запрос в базу зависнет. Передавайте в приёмник отдельный context.WithTimeout(context.Background(), ...), а не тот же ctx, который уже отменён.

Сравнение подходов

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

Что ломается чаще всего

• Переиспользованный слайс в асинхронном flush. batch[:0] при передаче в горутину или канал приводит к гонке данных — go test -race её ловит, глазами не видно.
• Забытый сброс таймера. Без ticker.Reset после отправки по размеру получаете дребезг: пакет ушёл, а через миг таймер шлёт почти пустой.
• Отсутствие финального слива. При закрытии канала или отмене контекста остаток теряется — самая частая потеря данных в батчинге.
• Слишком большой буфер канала батчей. Маскирует медленный приёмник: память растёт, а вы узнаёте о проблеме по OOM, а не по замедлению.
• time.After в цикле вместо Ticker. time.After внутри select в цикле создаёт новый таймер на каждой итерации и не даёт сборщику его убрать до срабатывания — на горячем цикле это утечка. Используйте time.NewTicker или time.NewTimer с ручным Reset.

Что делать дальше

Соберите минимальный конвейер из приёма с таймером, прогоните под go test -race с несколькими воркерами — гонку по слайсу проще всего словить именно так. Померьте две метрики в проде: размер уходящих батчей (гистограмма) и время в буфере до отправки. Если гистограмма забита единицами и двойками — интервал таймера слишком мал под ваш трафик, увеличивайте. Если время в буфере регулярно упирается в maxWait, а размер не добирает — поток реже, чем вы рассчитывали, и таймер работает вместо размера. Инвалидация или дедупликация внутри батча — отдельная тема, здесь мы её не трогали.