7.7Распределённые блокировки и Redis из Go

Разбери ограничения lock через SET NX PX: владелец может пережить TTL, а новый владелец уже начать работу. Сравни lock с fencing token, разберись в pub/sub и Streams, затем проверь, как go-redis ведёт себя при промахе, таймауте и заполненном пуле.

Блокировка на Redis: SET NX PX

Хватает одной команды:

SET lock:job:42 <уникальное-значение> NX PX 10000

• NX - установить, только если ключа нет. Это и есть атомарный захват: команда либо создаёт ключ и возвращает OK, либо возвращает nil. Однопоточный сервер выполняет проверку и установку как одну операцию, так что гонке между ними взяться неоткуда.
• PX 10000 - TTL в миллисекундах.
• В значение кладут случайный UUID держателя.

Зачем TTL

Без TTL блокировка упавшего держателя остаётся навсегда. Процесс убили OOM-killer-ом, под выселили, машина отключилась - defer unlock() не выполнится, потому что выполнять его некому. Ключ висит, работа стоит, чинят руками в три часа ночи. Освободить блокировку без участия упавшего процесса умеет только TTL. Из него же растёт главная проблема таких блокировок, о ней ниже.

Зачем уникальное значение

Чтобы снять свою блокировку, а не чужую. Сценарий: A взял блокировку с TTL 10 секунд, работа затянулась на 12; на десятой секунде ключ истёк, B тут же его взял; на двенадцатой A заканчивает и делает DEL lock:job:42, то есть удаляет блокировку B. Дальше блокировку берёт C, и теперь B и C работают параллельно. Забыл одну деталь, и взаимного исключения больше нет. С уникальным значением блокировку можно снимать с проверкой: «удали, только если там всё ещё моё значение».

Освобождение через Lua

«Проверить и удалить» состоит из двух операций, и между ними ключ может истечь. Классическая гонка check-then-act: GET вернул моё значение, в этот момент TTL истёк, B захватил блокировку, и мой DEL сносит уже его ключ. Атомарность даёт Lua-скрипт: Redis выполняет его целиком, не вклинивая чужие команды.

-- unlock.lua: удалить ключ, только если значение совпадает
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

var unlockScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end`)

type Lock struct {
    rdb   *redis.Client
    key   string
    value string        // уникальный идентификатор владения
}

func Acquire(ctx context.Context, rdb *redis.Client, key string, ttl time.Duration) (*Lock, error) {
    val := uuid.NewString()
    ok, err := rdb.SetNX(ctx, key, val, ttl).Result()   // SET key val NX PX ttl
    if err != nil {
        return nil, err
    }
    if !ok {
        return nil, ErrLockBusy                          // занято кем-то другим
    }
    return &Lock{rdb: rdb, key: key, value: val}, nil
}

func (l *Lock) Release(ctx context.Context) error {
    // NewScript.Run сначала пробует EVALSHA, а при NOSCRIPT переходит на EVAL.
    n, err := unlockScript.Run(ctx, l.rdb, []string{l.key}, l.value).Int64()
    if err != nil {
        return err
    }
    if n == 0 {
        // Ключа нет или он уже не наш: TTL истёк, пока мы работали.
        // Это не безобидно: значит, кто-то мог работать параллельно с нами.
        return ErrLockLost
    }
    return nil
}

Главная проблема: TTL истёк, а работа идёт

Блокировка с TTL держится ровно TTL - и ни микросекундой больше. А задержать держателя дольше могут пауза GC (в Go редко, но stop-the-world есть), вытеснение потока планировщиком ОС, залипший сетевой вызов, «замороженный» контейнер, миграция виртуальной машины или просто медленный запрос в БД. В любой из этих ситуаций два клиента одновременно считают себя держателями блокировки, и оба уверены, что имеют право писать.

Watchdog - продление TTL

Обычно делают так: пока работа идёт, фоновая горутина периодически продлевает TTL (в Redisson это называется watchdog, отсюда термин). Продлевают тем же скриптом с проверкой значения - иначе продлишь чужую блокировку.

var extendScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("PEXPIRE", KEYS[1], ARGV[2])
else
    return 0
end`)

// Запускается сразу после захвата, останавливается через cancel при Release.
func (l *Lock) keepAlive(ctx context.Context, ttl time.Duration) {
    t := time.NewTicker(ttl / 3)      // продлеваем втрое чаще, чем истекает
    defer t.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-t.C:
            ok, err := extendScript.Run(ctx, l.rdb, []string{l.key}, l.value,
                ttl.Milliseconds()).Int64()
            if err != nil || ok == 0 {
                // Блокировку потеряли, значит работу надо прервать, а не продолжать:
                // параллельно с нами уже может идти другой держатель.
                l.lost()          // закрывает канал, по которому работа отменяется
                return
            }
        }
    }
}

Fencing token

Идея из статьи Клеппмана. При каждом захвате блокировки выдаётся монотонно растущее число (в Redis это INCR lock:{job:42}:fence: в Cluster фигурные скобки сводят лок и счётчик в один слот, иначе скрипт захвата упадёт с CROSSSLOT). Клиент передаёт этот токен в каждый запрос к защищаемому ресурсу, а ресурс помнит самый большой токен, какой видел, и отвергает всё, что меньше. Тогда «воскресший» старый держатель с токеном 33 физически не сможет испортить работу нового с токеном 34. Оговорка: счётчик в Redis монотонен, пока Redis не теряет записи. После failover с асинхронной репликацией он может откатиться назад, поэтому для строгих гарантий токен берут из хранилища с консенсусом или из той же БД, что защищает ресурс.

-- захват с выдачей токена, одним скриптом
if redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2]) then
    return redis.call("INCR", KEYS[2])   -- монотонный счётчик поколений
else
    return 0
end

Redlock и критика Клеппмана

Алгоритм Redlock Сальваторе Санфилиппо (автор Redis) предложил для блокировки поверх N независимых мастеров Redis (обычно 5), не связанных репликацией. Клиент:

1. берёт текущее время и последовательно пытается захватить ключ на всех N инстансах с малым таймаутом на каждый;
2. считает блокировку взятой, если получил её на большинстве (N/2+1) и суммарно потратил меньше TTL;
3. владеет ею TTL минус потраченное время минус поправка на дрейф часов;
4. при неуспехе снимает блокировку со всех инстансов, где успел взять.

Смысл: пережить падение отдельных инстансов без Sentinel и без потери блокировки при failover.

Критика Мартина Клеппмана (2016)

Статья «How to do distributed locking» и ответ Санфилиппо по пунктам.

1. Разделение целей. Клеппман начинает с вопроса, зачем вообще нужна блокировка: для эффективности (не делать одну работу дважды) или для корректности (двойное выполнение недопустимо). Для эффективности сгодится и одиночный Redis. Для корректности Redlock не подходит, и дальше Клеппман объясняет почему.
2. Опора на часы. Redlock считает истечение по времени, предполагая ограниченный дрейф часов. Но время может скакнуть: NTP-коррекция, живая миграция ВМ, переустановка часов администратором. Время на одном узле прыгнуло вперёд, и блокировка «истекает» раньше, чем думает держатель. Если корректность системы зависит от синхронности часов, в асинхронной модели она небезопасна.
3. Паузы процесса. Даже с идеальными часами держателя может остановить STW-пауза, своп, вытеснение планировщиком или заморозка контейнера. После паузы он не знает, что блокировку потерял, - и продолжает работать. Никакой алгоритм захвата это не ловит.
4. Fencing token как единственное лекарство. Отсюда вывод: нужен монотонный токен, проверяемый ресурсом. Redlock не выдаёт такого токена: у него N независимых инстансов, и получить из них согласованно возрастающее число нельзя. А системы с настоящим консенсусом (ZooKeeper с zxid, etcd с revision, Consul) его выдают.
5. Ответ Санфилиппо. Он возразил, что предположения о часах ограничены и практичны, а про паузы сказал, что fencing token нужен любой блокировке, включая ZooKeeper, так что это не претензия именно к Redlock. Обе стороны частично правы; консенсуса в сообществе нет.

Pub/Sub, Streams и полноценный брокер

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

Что именно теряется в pub/sub

• Нет персистентности. PUBLISH доставляет только тем, кто подписан прямо сейчас. Подписчик отвалился на две секунды, пока переподключался, и сообщений за эти две секунды он не увидит и никогда не узнает об этом.
• Нет подтверждений. PUBLISH возвращает число подписчиков, которым сообщение отправлено, а не число обработавших. Обработал ли кто-то, отправитель не узнает.
• Медленный подписчик отбрасывается. Если он не успевает вычитывать, у него растёт output buffer, и когда тот упрётся в client-output-buffer-limit pubsub, Redis рвёт соединение. Для приложения это происходит тихо.
• В кластере рассылается по всем нодам - это дорого. С Redis 7 есть шардированный SPUBLISH/SSUBSCRIBE, привязанный к слоту канала.

Отсюда единственно корректное применение pub/sub: сигналы-ускорители, потеря которых безопасна. Сюда хорошо ложится инвалидация локальных кэшей: если сигнал не дошёл, ключ всё равно протухнет по TTL через секунду. А бизнес-события («заказ оплачен») на pub/sub гарантированно потеряются, причём тихо и в первый же деплой.

Когда Streams достаточно

• Умеренный объём, который помещается в память с запасом, и есть MAXLEN.
• Retention нужен часами, а не неделями: перечитать вчерашний день не потребуется.
• Redis уже есть в инфраструктуре, и заводить Kafka ради одной очереди задач избыточно.
• Нужны подтверждения и повторная выдача зависших сообщений - PEL и XAUTOCLAIM это дают.

Когда нужен настоящий брокер

• Retention днями и неделями, replay. В Redis лог живёт в памяти - терабайт истории там никто держать не станет.
• Много независимых потребителей одного потока. В Kafka это естественно: каждая consumer group читает свой offset по одному и тому же логу.
• Строгие требования к durability. Kafka пишет на диск и реплицирует с acks=all; Redis теряет до секунды при everysec и подтверждённые записи при failover.
• Экосистема: Kafka Connect, Streams/Flink, схемы (Avro/Protobuf + registry), compaction, транзакции. Ничего этого в Redis нет.
• Сложная маршрутизация уже по части RabbitMQ: exchange, routing key, TTL очередей, DLQ из коробки.

Redis из Go: go-redis

github.com/redis/go-redis/v9 стал стандартом де-факто (второй вариант, rueidis, быстрее за счёт автопайплайнинга по умолчанию, но встречается реже; RESP3 go-redis v9 включает и сам). Клиент потокобезопасен, держит внутри пул соединений и создаётся один на приложение.

rdb := redis.NewClient(&redis.Options{
    Addr:     "redis:6379",
    Password: os.Getenv("REDIS_PASSWORD"),
    DB:       0,

    // --- пул соединений ---
    PoolSize:        50,               // максимум сокетов; по умолчанию 10 * GOMAXPROCS
    MinIdleConns:    10,               // держать прогретыми, чтобы не платить за dial в пике
    MaxIdleConns:    20,
    ConnMaxIdleTime: 30 * time.Minute, // закрывать простаивающие
    ConnMaxLifetime: 0,                // 0 = не пересоздавать по возрасту
    PoolTimeout:     4 * time.Second,  // сколько ждать свободное соединение из пула

    // --- таймауты ---
    DialTimeout:  2 * time.Second,
    ReadTimeout:  500 * time.Millisecond,   // Redis быстрый: секунды тут не нужны
    WriteTimeout: 500 * time.Millisecond,

    // --- ретраи ---
    MaxRetries:      3,
    MinRetryBackoff: 8 * time.Millisecond,
    MaxRetryBackoff: 512 * time.Millisecond,
})

Про пул - то, что спрашивают

• Одна команда занимает одно соединение на время round-trip. Значит, нужный размер пула ≈ пиковая конкурентность обращений к Redis, а не число ядер. При латентности 0,3 мс одно соединение выдаёт ~3000 команд в секунду - 50 соединений закрывают 150 000 rps.
• Исчерпание пула даёт не отказ, а ожидание: горутина ждёт свободное соединение до PoolTimeout, потом получает redis: connection pool timeout. Это главный симптом: либо пул мал, либо Redis отвечает медленно (большие ключи, медленные команды). Поэтому метрики rdb.PoolStats() (Timeouts, TotalConns, IdleConns) надо снимать в Prometheus.
• Блокирующие команды съедают соединение целиком. BLPOP, XREAD BLOCK, подписка pub/sub держат сокет всё время ожидания. Их нельзя гонять через общий пул в большом количестве: заведи для них отдельный клиент или хотя бы помни, что N воркеров навсегда съедят N соединений.
• MaxRetries и идемпотентность. Ретраи включены по умолчанию. Для GET это безопасно, для INCR уже нет: команда могла выполниться, а ответ потеряться, и повтор посчитает второй раз. Для неидемпотентных операций либо MaxRetries: -1 (выключить), либо операция, устойчивая к повтору.

Контекст и таймауты

Каждая команда принимает ctx, но по умолчанию go-redis v9 его дедлайн на сокет не переносит (ContextTimeoutEnabled: false): контекст учитывается, пока ждёшь соединение из пула и между ретраями, а ответ ждут по ReadTimeout. С ContextTimeoutEnabled: true дедлайн контекста становится дедлайном чтения; оборванное посреди ответа соединение go-redis закрывает, так что массовые таймауты выглядят как шторм переподключений. И в любом случае уже отправленная команда в Redis выполнится. На практике ставят короткий ReadTimeout (порядка сотен миллисекунд) и не рассчитывают, что отмена «остановит работу в Redis».

redis.Nil - не ошибка, а результат

val, err := rdb.Get(ctx, key).Result()
switch {
case errors.Is(err, redis.Nil):
    // Ключа нет. Это штатная ветка, а не сбой: промах кэша.
    return loadFromDB(ctx, key)
case err != nil:
    // Настоящая ошибка: сеть, таймаут, пул. Тоже обычно ведёт в БД,
    // но с метрикой и логом, иначе не заметишь, что Redis лежит.
    metrics.CacheErrors.Inc()
    return loadFromDB(ctx, key)
default:
    return val, nil
}

Отдельно: redis.Nil прилетает не только из GET. Пустой результат BLPOP по таймауту, ZSCORE несуществующего элемента, HGET отсутствующего поля - всё это redis.Nil. А вот Exists и SetNX возвращают ноль/false без ошибки, и на этом тоже часто путаются.

Пайплайнинг

Пайплайн отправляет N команд одним пакетом и читает N ответов, экономя N−1 round-trip. Это самый дешёвый способ ускорить работу с Redis: при RTT 0,3 мс сто последовательных GET займут 30 мс, а пайплайном - 0,4 мс.

// Pipeline не атомарен: просто пакет команд, между ними могут выполниться чужие.
pipe := rdb.Pipeline()
cmds := make([]*redis.StringCmd, 0, len(ids))
for _, id := range ids {
    cmds = append(cmds, pipe.Get(ctx, keyOf(id)))
}
_, err := pipe.Exec(ctx)
// Exec возвращает ошибку, если упала хотя бы одна команда, включая redis.Nil.
// Результаты всё равно надо разбирать по каждой команде отдельно.
if err != nil && !errors.Is(err, redis.Nil) {
    return nil, err
}
for i, c := range cmds {
    v, err := c.Result()
    if errors.Is(err, redis.Nil) {
        continue                 // ключа нет: промах, не ошибка
    }
    ...
}

// TxPipeline: то же самое, но обёрнуто в MULTI ... EXEC.
// Все команды выполнятся подряд, без чужих между ними.
pipe := rdb.TxPipeline()
incr := pipe.Incr(ctx, "counter")
pipe.Expire(ctx, "counter", time.Hour)
_, err := pipe.Exec(ctx)          // MULTI, INCR, EXPIRE, EXEC одним round-trip
fmt.Println(incr.Val())           // 1 - счётчик после первого INCR
// До Exec поля команд пустые: incr.Val() вернул бы 0. Заполняются они
// в момент Exec, поэтому читать результат раньше - типичная ошибка.

Lua: EVAL, EVALSHA и redis.NewScript

Lua-скрипт выполняется на сервере атомарно: пока он работает, ни одна другая команда не выполнится. Только так в Redis получается настоящее «прочитать-проверить-записать». EVAL отправляет текст скрипта каждый раз, EVALSHA - только SHA1-хеш, экономя трафик. go-redis прячет оба за одним вызовом:

// Скрипт объявляется один раз как переменная пакета.
var rateLimit = redis.NewScript(`
local current = redis.call("INCR", KEYS[1])
if current == 1 then
    redis.call("PEXPIRE", KEYS[1], ARGV[1])
end
if current > tonumber(ARGV[2]) then
    return 0
end
return 1`)

// Run сначала пробует EVALSHA; если сервер ответил NOSCRIPT
// (перезапуск, SCRIPT FLUSH), сам переключается на EVAL и кеширует скрипт заново.
allowed, err := rateLimit.Run(ctx, rdb,
    []string{"rl:" + userID},                 // KEYS
    int64(time.Minute/time.Millisecond), 100, // ARGV: окно и лимит
).Int64()

if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

var unlockScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end`)

func (l *Lock) Release(ctx context.Context) error {
    n, err := unlockScript.Run(ctx, l.rdb, []string{l.key}, l.value).Int64()
    if err != nil {
        return err
    }
    if n == 0 {
        // Ключа уже нет или он не наш: TTL истёк, пока мы работали.
        // Это сигнал, что параллельно с нами кто-то мог работать.
        return ErrLockLost
    }
    return nil
}

var extendScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("PEXPIRE", KEYS[1], ARGV[2])
else
    return 0
end`)

t := time.NewTicker(ttl / 3)            // продлеваем втрое чаще, чем истекает
for {
    select {
    case <-ctx.Done():
        return
    case <-t.C:
        ok, err := extendScript.Run(ctx, rdb, []string{key}, val, ttl.Milliseconds()).Int64()
        if err != nil || ok == 0 {
            l.lost()   // прервать работу, а не продолжать: рядом уже может идти другой
            return
        }
    }
}

-- захват с выдачей токена
if redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2]) then
    return redis.call("INCR", KEYS[2])
else
    return 0
end

-- проверка на стороне ресурса, например в Postgres
UPDATE jobs SET result = $1, fence = $2 WHERE id = $3 AND fence <= $2;
-- 0 строк --> наш токен устарел, мы больше не держатель, результат отбрасываем

rdb := redis.NewClient(&redis.Options{
    Addr: "redis:6379",

    PoolSize:        50,               // дефолт: 10 * GOMAXPROCS
    MinIdleConns:    10,               // прогретые соединения, чтобы не платить за dial в пике
    ConnMaxIdleTime: 30 * time.Minute,
    ConnMaxLifetime: 0,                // 0 = не пересоздавать по возрасту
    PoolTimeout:     4 * time.Second,  // сколько ждать свободное соединение

    DialTimeout:  2 * time.Second,
    ReadTimeout:  500 * time.Millisecond,   // Redis быстрый, секунды тут не нужны
    WriteTimeout: 500 * time.Millisecond,

    MaxRetries:      3,
    MinRetryBackoff: 8 * time.Millisecond,
    MaxRetryBackoff: 512 * time.Millisecond,
})

pipe := rdb.Pipeline()
cmds := make([]*redis.StringCmd, 0, len(ids))
for _, id := range ids {
    cmds = append(cmds, pipe.Get(ctx, keyOf(id)))
}
_, err := pipe.Exec(ctx)
// Exec вернёт ошибку, если упала хотя бы одна команда, включая redis.Nil.
if err != nil && !errors.Is(err, redis.Nil) {
    return nil, err
}
for _, c := range cmds {
    v, err := c.Result()
    if errors.Is(err, redis.Nil) {
        continue          // просто нет ключа
    }
    ...
}

pipe := rdb.TxPipeline()          // MULTI ... EXEC
incr := pipe.Incr(ctx, "counter")
pipe.Expire(ctx, "counter", time.Hour)
_, err := pipe.Exec(ctx)
fmt.Println(incr.Val())           // 1 - счётчик после первого INCR
                                  // (до Exec было бы 0: результаты приезжают в Exec)

// Объявляется один раз, на уровне пакета.
var rateLimit = redis.NewScript(`
local current = redis.call("INCR", KEYS[1])
if current == 1 then
    redis.call("PEXPIRE", KEYS[1], ARGV[1])
end
if current > tonumber(ARGV[2]) then
    return 0
end
return 1`)

// Run: сначала EVALSHA, при NOSCRIPT сам вызывает EVAL и снова кеширует скрипт.
allowed, err := rateLimit.Run(ctx, rdb,
    []string{"rl:" + userID},                  // KEYS
    int64(time.Minute/time.Millisecond), 100,  // ARGV
).Int64()

val, err := rdb.Get(ctx, key).Result()
switch {
case errors.Is(err, redis.Nil):
    // Ключа нет: штатный промах.
    metrics.CacheMiss.Inc()
    return loadFromDB(ctx, key)
case err != nil:
    // Настоящая ошибка: сеть, таймаут, пул. Мы не знаем, есть ключ или нет.
    metrics.CacheErrors.Inc()
    slog.Warn("cache unavailable", "err", err)
    return loadFromDB(ctx, key)     // деградация, а не отказ
default:
    metrics.CacheHit.Inc()
    return val, nil
}

Что стоит запомнить

• Случайное значение владельца и Lua unlock не дают удалить чужой lock после истечения TTL.
• Fencing token защищает ресурс только когда сам ресурс отвергает устаревшие токены.
• Pub/Sub теряет события для отключённого подписчика, Streams хранят записи и поддерживают consumer groups.

Частые ошибки

• Redis lock без fencing token не гарантирует защиту долгой операции после истечения TTL.
• Pipeline экономит round trips, но не изолирует последовательность команд от других клиентов.
• redis.Nil означает отсутствие ключа, а timeout/connection error означает неопределённый исход запроса.

SET lock token-A NX PX 3000
...work takes 5000 ms...
SET lock token-B NX PX 3000
-- A runs owner-check Lua and finishes late

func (l *Lock) Release(ctx context.Context) error {
    return l.rdb.Del(ctx, l.key).Err()
}