Redis Và Distributed System: Một Hành Trình Thực Tế

Nghiên cứu sâu về việc tận dụng đặc tính của các cấu trúc dữ liệu Redis để xây dựng các mô hình phân tán: mutex, leader election, delayed scheduling, và competing consumers.

Mục Lục

Cấu Trúc Dữ Liệu Redis: Bản Chất Và Đặc Tính
Message Patterns: Pub/Sub vs Queue vs Streams
Sorted Set + List: Anatomy Của Delayed Queue
Distributed Lock: Từ Hash Đến Sorted Set Buffer
Queue-As-Token: Leader Election Bằng Atomic Pop
Distributed State: Shared vs Per-Pod
Atomicity Và Race Conditions
Failure Modes Và Recovery
Tổng Kết: Data Structure Driven Distributed Design

1. Cấu Trúc Dữ Liệu Redis

Bản Chất Của Mỗi Cấu Trúc

Hiểu được bản chất của từng cấu trúc là điều kiện tiên quyết để chọn đúng tool cho bài toán phân tán:

Cấu trúc	Bản chất	Đặc tính phân tán quan trọng
String	Single value	`SET NX EX` atomic — nền tảng của distributed lock
List	Linked list of strings	`LPOP/RPOP` atomic — competing consumers
Hash	Map (field → value)	`HSETNX` atomic — partial update
Set	Tập hợp unordered	`SADD` dedup atomic — distinct counter
Sorted Set	Set có score	`ZRANGEBYSCORE` — time-based queue
Stream	Append-only log	Consumer groups, ack, replay

Properties Quyết Định Hành Vi Phân Tán

1. Atomicity của operation:

SET NX, SADD, ZADD, LPUSH/LPOP → atomic per command
Multi-command sequence → KHÔNG atomic (cần Lua hoặc MULTI/EXEC)

2. Dedup natural:

Set/Sorted Set → tự dedup theo member
List → cho phép duplicate
Hash → dedup theo field

3. Ordering:

List → insertion order (FIFO/LIFO)
Sorted Set → score-based order
Set/Hash → no order
Stream → time-based with ID

4. Persistence của message:

Pub/Sub → ephemeral
List/Stream → persistent
Sorted Set → persistent

Insight: Mỗi Bài Toán Phân Tán Map Tới 1-2 Cấu Trúc

Bài toán	Cấu trúc	Lý do
Mutex	String (SET NX EX)	Atomic create + TTL
Task queue	List (LPUSH/RPOP)	Atomic pop, FIFO
Delayed task	Sorted Set (score=timestamp)	Range query theo thời gian
Distinct counter	Set (SADD)	Tự dedup
Leaderboard	Sorted Set (score=points)	Sort by score
Event sourcing	Stream	Persistent log + consumer groups
Rate limiting	Sorted Set (sliding window)	Time-based count

2. Mô Hình Truyền Tin

Ba Paradigm Khác Nhau Hoàn Toàn

Pub/Sub — Broadcast Pattern:

                    ┌──► Pod A (nhận)
Publisher ─► CH ──┤
                    ├──► Pod B (nhận)
                    └──► Pod C (nhận)

Mọi subscriber active đều nhận message
Không persistent → subscriber offline = miss
Fire-and-forget → không ack

Queue — Competing Consumers:

Producer ─► [m3, m2, m1] ─► Pod A pop m1
                          ─► Pod B pop m2 (atomic, không trùng)
                          ─► Pod C pop m3

1 message → 1 consumer
Persistent → offline khi back lại vẫn nhận
LPOP/RPOP atomic đảm bảo không duplicate

Streams — Hybrid Với Consumer Groups:

Stream ─► Consumer Group A ─► Pod A1 hoặc A2 nhận (competing trong group)
       ─► Consumer Group B ─► Pod B1 hoặc B2 nhận (independent group)
       ─► Consumer Group C ─► ...

Trong cùng group: competing
Khác group: broadcast
Có ack, retry, pending list

Khi Nào Chọn Cái Nào?

Pub/Sub phù hợp khi:

Broadcast event (cache invalidation, config reload)
Subscriber có thể miss message mà không sao
Low latency hơn quan trọng hơn reliability

Queue (List) phù hợp khi:

Mỗi task chỉ cần 1 worker xử lý
Đơn giản, không cần ack
Throughput không quá cao

Streams phù hợp khi:

Cần multiple consumer groups (microservices)
Cần replay/audit log
Cần guarantee delivery (ack, retry, dead letter)

Bài Học Quan Trọng

Pub/Sub không thể dùng cho task queue vì broadcast.

3. Anatomy Delayed Queue

Vấn Đề Cốt Lõi

Redis không có "delayed task" native. Nhưng có:

Sorted Set với score = timestamp → có thể query "task nào đến hạn"
List với atomic pop → competing consumer

Kết hợp 2 → Delayed Queue.

Kiến Trúc

┌────────────────────────────┐
│  Sorted Set                │
│  member = task             │  ← Producer: ZADD score=runAt task
│  score = runAt (timestamp) │
└───────────┬────────────────┘
            │
            │ Mover (scheduler)
            │ ZRANGEBYSCORE 0 now
            ▼
┌────────────────────────────┐
│  List                      │
│  [task ready ...]          │  ← Consumer: RPOP task
└────────────────────────────┘

Mover Logic

protected void drainToQueue() {
    long now = System.currentTimeMillis();
    redisAsyncSortedSet.getSliceByScore(0, now)
        .compose(elements -> {
            if (elements == null || elements.isEmpty()) {
                return Future.succeededFuture();
            }
            return super.multiPut(elements)
                .compose(res -> {
                    if (res == null || !res) return Future.succeededFuture();
                    // QUAN TRỌNG: xóa theo elements thực sự lấy được
                    return redisAsyncSortedSet.remove(elements).mapEmpty();
                });
        });
}

Race Condition Tinh Tế: Get vs Remove

Bug nếu xóa theo range:

T=1000: getSliceByScore(0, 1000) → []
T=1001: ZADD score=999 elem (task nộp muộn)
T=1005: removeSliceByScore(0, 1000) → XÓA NHẦM elem ❌

removeSliceByScore xóa mọi entry có score ≤ 1000, bao gồm cả entry mới ZADD sau khi get đã chạy. Element vào zset trong khoảng giữa bị nuốt mất.

Fix: xóa chính xác elements đã lấy bằng ZREM:

redisAsyncSortedSet.remove(elements)  // ZREM cụ thể, không phải range

Insight Kiến Trúc: Mỗi Pod Một Sorted Set Riêng

Vì sao tách zset per-pod?

Scenario shared zset:

Pod A: ZRANGEBYSCORE 0 now → [t1, t2]
Pod B: ZRANGEBYSCORE 0 now → [t1, t2]  ← cùng kết quả!
Pod A: LPUSH list t1, t2
Pod B: LPUSH list t1, t2  ← duplicate!

→ Phải có distributed lock cho mover. Phức tạp.

Per-pod zset:

Pod A zset: { task1, task2 }  ← chỉ Pod A drain
Pod B zset: { task3, task4 }  ← chỉ Pod B drain
Shared list: [...]            ← consumer competing

→ Mover của mỗi pod hoàn toàn độc lập, không cần lock.

Trade-off: Pod chết → orphan zset. Giải pháp:

onExit flush trước khi shutdown
Hoặc watchdog scan orphan zsets (phức tạp)
Hoặc chấp nhận mất data ở edge case

Score Generation

Bug "anti-postpone" tinh tế:

T=1000: ZADD score=1500 "TOKEN" → ready ở T+500
T=1100: ZADD score=1600 "TOKEN" → ready ở T+500 (đã trượt!)
T=1200: ZADD score=1700 "TOKEN" → ready ở T+500
                                   ↑ infinite postpone

ZADD thường update score nếu member đã tồn tại → producer push liên tục = score liên tục bị đẩy lùi = task không bao giờ ready.

Fix với ZADD NX:

T=1000: ZADD NX score=1500 → add ✅
T=1100: ZADD NX score=1600 → skip (đã có) ✅
T=1500: drain → task ready đúng giờ ✅

⚠️ Pika không support ZADD NX → cần workaround check-then-add hoặc đổi pattern.

4. Distributed Lock

Anatomy

Distributed lock cần 3 properties:

Mutual exclusion — chỉ 1 holder
Auto-release (TTL) — pod chết thì lock tự free
Ownership verification — chỉ owner mới được release

Implementation Cơ Bản (SET NX EX)

public Future<Boolean> tryAcquire(String key, String owner, int ttl) {
    return redisAPI.set(List.of(
        key, owner, "NX", "EX", String.valueOf(ttl)
    )).map(response -> response != null);
}

public Future<Boolean> release(String key, String owner) {
    return redisAPI.get(key)
        .compose(current -> {
            if (current == null || !owner.equals(current.toString())) {
                return Future.succeededFuture(false);
            }
            return redisAPI.del(List.of(key)).map(r -> r.toInteger() > 0);
        });
}

Implementation Phức Tạp Hơn (Hash + Sorted Set Buffer)

Trong production, lock thường được implement với 2 cấu trúc:

Hash: storage cho value (owner ID)
Sorted Set: buffer expire time (score = expireAt)

public Future<Boolean> putWithTtlNX(String key, String owner, int ttl) {
    int expireAt = (int) (System.currentTimeMillis() / 1000) + ttl;

    return buffer.putNX(key, expireAt)  // Sorted Set NX
        .compose(bufferOk -> {
            if (!Boolean.TRUE.equals(bufferOk)) {
                return Future.succeededFuture(false);
            }
            return put(key, owner);  // Hash HSET
        });
}

Lợi ích: flexible — có thể scan keys sắp expire, list all locks, etc.

Trade-off: 2x cost vs SET NX EX native.

Race Condition Trong tryAcquire

Bug — check then put không atomic:

public Future<Boolean> tryAcquire(String key, String owner, int ttl) {
    return contains(key).compose(locked -> {
        if (locked) return Future.succeededFuture(false);
        return putWithTtl(key, owner, ttl);  // ❌ window race
    });
}

T=0: Pod A: contains → false
T=1: Pod B: contains → false   ← cùng thấy "chưa lock"
T=2: Pod A: put owner=A         ← cả 2 đều "acquire" thành công
T=3: Pod B: put owner=B         ← overwrite của A!

Fix: dùng atomic NX, bỏ check riêng:

public Future<Boolean> tryAcquire(String key, String owner, int ttl) {
    return putWithTtlNX(key, owner, ttl);  // atomic ở NX
}

Race Condition Trong release

Khó tránh hơn:

T=0:   Pod A acquire lock, TTL=10s
T=11:  Lock tự expire (Pod A xử lý chậm)
T=12:  Pod B acquire lock thành công
T=13:  Pod A finally release:
       GET → "B" ≠ "A" → KHÔNG DEL ✅ (owner check bảo vệ)

NHƯNG:
T=0:    Pod A acquire lock
T=11:   Lock expire
T=12:   Pod B acquire
T=13:   Pod A release:
        GET → "B" ≠ "A" → KHÔNG DEL ✅
        
NHƯNG case khác:
T=0:    Pod A acquire, owner=A
T=13.0: Pod A release: GET → "A" ✅
T=13.1: Lock expire (đúng TTL)
T=13.2: Pod B acquire, owner=B
T=13.3: Pod A: DEL → xóa nhầm lock của B! ❌

→ Window race giữa GET và DEL. Lua script là cách duy nhất để atomic check-and-del:

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

Get Phải Filter Expired

Bug — return expired data:

public Future<String> get(String key) {
    return buffer.get(key)
        .compose(expireAt -> {
            if (expireAt != null && expireAt <= now) {
                remove(key);   // cleanup async
            }
            return super.get(key);  // ❌ vẫn return data đã expired!
        });
}

Hậu quả nghiêm trọng cho distributed lock:

Pod A acquire lock TTL=10s, owner=A
T=20s: Pod B tryAcquire:
       contains() → get("lock") → trả "A" (đã expired nhưng vẫn return)
       contains = true → KHÔNG acquire được
       → Lock effectively never expires

Fix:

public Future<String> get(String key) {
    return buffer.get(key)
        .compose(expireAt -> {
            if (expireAt == null || expireAt <= now) {
                if (expireAt != null) {
                    remove(key);
                    buffer.remove(key);
                }
                return Future.succeededFuture(null);  // return null cho expired
            }
            return super.get(key);
        });
}

5. Queue-As-Token

Ý Tưởng

Tận dụng đặc tính atomic pop của List để làm leader election. Nhiều pod cùng "fight" cho 1 token, ai pop được thì chạy job:

Init: LPUSH queue "1"  ← seed token

Pod A: RPOP queue → "1"  ← thắng
Pod B: RPOP queue → null
Pod C: RPOP queue → null

Pod A: doWork() → done → LPUSH queue "1"  ← trả token
... (chu kỳ tiếp tục)

Tại Sao Pattern Này Clever

Tận dụng atomic của LPUSH/RPOP — không cần distributed lock
Tự built-in throttle — chỉ 1 pod chạy mỗi chu kỳ
Không phụ thuộc framework — chỉ Redis primitive

Tại Sao Pattern Này Fragile

Failure mode 1: Pod chết giữa chừng → token mất:

Pod A: RPOP → "1"  ← giữ token
Pod A: doWork...
Pod A CHẾT trước khi LPUSH "1" trả token
→ Queue rỗng vĩnh viễn
→ Không pod nào chạy được job nữa

Failure mode 2: Race khi token đang trong "limbo":

T=0:    Pod A RPOP → "1" → doWork (10s)
T=5:    Pod B watchdog: queue rỗng → seed token mới
T=6:    Pod C RPOP token mới → doWork (Pod A vẫn chạy!)
        → 2 pod cùng chạy ❌
T=10:   Pod A xong → LPUSH "1" → giờ có 2 token

Cải Tiến: Delayed Queue Với Token

Dùng Delayed Queue thay List:

Pod A xong → put("1") với delay 10s
T+10: token mới ready → pod khác có cơ hội pickup

Lợi ích:

Pod vừa chạy không sticky leader (token có delay)
Anti-postpone bug được giảm bớt (ZADD NX)
Built-in cooldown

Vấn Đề Fundamental: Queue Dedup ≠ Mutex

Đây là insight quan trọng nhất của pattern này:

T=10:    mover drain → list ["1"], zset rỗng
T=10.1:  Pod A RPOP → "1" → doWork (5s)
T=10.5:  Pod B put → zset có (zset rỗng nên NX add OK)
T=11:    mover drain → list ["1"]
T=11:    Pod C RPOP → doWork (Pod A vẫn chạy)
         → 2 pod cùng chạy ❌

Lý do: NX chỉ dedup khi zset đang có entry. Khi mover vừa drain xong → zset rỗng → NX bất lực → producer khác có thể put trùng.

Kết luận: Queue-as-token không đảm bảo mutex một cách reliable. Để mutex thật sự → phải có distributed lock bao quanh doWork:

return queue.dequeue()
    .compose(token -> {
        if (isEmpty(token)) return Future.succeededFuture();
        
        return distributedLock.tryAcquire("job_" + name, ownerId, ttl)
            .compose(acquired -> {
                if (!acquired) return Future.succeededFuture();  // pod khác đang chạy
                
                return doWork()
                    .onComplete(ar -> {
                        distributedLock.release("job_" + name, ownerId);
                        queue.put("1");
                    });
            });
    });

→ Queue làm scheduler/trigger, Lock làm mutex. Cả 2 đóng vai trò khác nhau.

6. Shared vs Per-Pod

Một Quyết Định Quan Trọng Trong Distributed Design

Khi xây dựng pattern phân tán, luôn có câu hỏi: state nên share giữa các pod, hay mỗi pod giữ riêng?

Per-Pod State

Ưu điểm:

Không race condition giữa pod
Performance cao (không lock)
Đơn giản về code

Nhược điểm:

Pod chết → state mất
Khó coordinate giữa pod
Memory tốn N lần

Shared State

Ưu điểm:

1 source of truth
Persistent qua pod restart
Coordinate dễ

Nhược điểm:

Race condition phải xử lý
Performance thấp hơn (Redis round-trip)
Phức tạp hơn

Case Study: Delayed Queue Sorted Set

Lựa chọn 1: Shared zset

Tất cả pod cùng ZADD vào "delayed_jobs"
Tất cả pod cùng ZRANGEBYSCORE để drain

→ Race khi drain: 2 pod cùng đọc → cùng LPUSH → duplicate. → Phải có distributed lock cho mover.

Lựa chọn 2: Per-pod zset (chosen)

Pod A zset: "delayed_jobs_uuidA"
Pod B zset: "delayed_jobs_uuidB"
Shared list: "ready_queue"

→ Mover độc lập, không race. → Pod chết → zset orphan, cần onExit flush.

Kết luận: chọn per-pod cho mover state (ephemeral, tốc độ cao), shared cho task data (cần persistent, cross-pod).

Case Study: Dedup Cache

Lựa chọn 1: In-memory Set per-pod

private final Set<String> recentlyPushed = ConcurrentHashSet.newSet();

→ Race window giữa contains() và add(). → Pod A có item X, pod B push trùng X → vẫn duplicate. → Memory leak nếu không clear.

Lựa chọn 2: Redis Set (shared)

redisAPI.sadd("dedup_set", item);  // atomic dedup

→ Distributed dedup đúng. → Tốn round-trip Redis cho mỗi check.

Lựa chọn 3: Caffeine cache TTL per-pod (chosen)

Cache<String, Boolean> recentlyPushed = Caffeine.newBuilder()
    .expireAfterWrite(1, TimeUnit.SECONDS)
    .maximumSize(10_000)
    .build();

→ Tự expire, không stuck. → Per-pod nhưng đủ tốt cho mục đích "giảm" (không phải "tuyệt đối"). → Trade-off: vẫn có thể duplicate cross-pod, nhưng rate rất thấp.

Insight: chọn theo mức độ chấp nhận inconsistency. Nếu chỉ cần "giảm rate", per-pod cache là đủ.

7. Atomicity

Mức Độ Atomicity Trong Redis

1. Single command atomic
   SET key value → atomic
   ZADD key score member → atomic
   LPOP key → atomic

2. Multi command với MULTI/EXEC
   Atomic nhưng không có conditional logic

3. Lua script EVAL
   Atomic + có thể có logic
   Pattern an toàn nhất cho compound ops

4. Application-level "atomic"
   Compose ops + check + retry
   Có race, phải chấp nhận edge case

Pattern Phổ Biến Phá Vỡ Atomicity

Check-then-act:

if (exists(key)) {
    return false;
}
set(key, value);  // ❌ race với check

Get-modify-put:

value = get(key);
value++;
set(key, value);  // ❌ race với get
// Fix: INCR atomic

Multi-step transaction:

zadd(key, member);     // ✅ atomic
lpush(list, member);   // ❌ giữa 2 ops, có thể fail
zrem(key, member);     // → state inconsistent nếu lpush fail

Khi Nào Cần Atomic, Khi Nào Không?

Cần atomic:

Mutex/lock acquire
Decrement counter
Compare-and-swap
Critical financial ops

Không cần atomic (chấp nhận eventual):

Cache update
Best-effort dedup
Metric aggregation
Logging

Trade-off Atomicity vs Complexity

Approach	Atomicity	Complexity	Performance
Single Redis cmd	✅ Full	Thấp	Cao
Lua script	✅ Full	Trung bình	Cao
MULTI/EXEC	⚠️ Partial	Trung bình	Trung bình
App-level + retry	❌ Partial	Cao	Thấp

Insight: trong production thực tế, chấp nhận một số race là OK nếu nghiệp vụ cho phép. Đừng over-engineer bằng Lua script khi 99.9% chính xác là đủ.

8. Failure Modes

Phân Loại Failure

1. Process failure: pod chết đột ngột (OOM, kill, crash) 2. Network partition: pod tách khỏi Redis 3. Slow process: pod sống nhưng chậm (GC pause, high CPU) 4. Clock skew: thời gian giữa pod và Redis lệch 5. Redis failure: master down, failover, data loss

Recovery Strategies

Strategy 1: TTL Auto-Expire

acquire lock với TTL=30s
pod chết → 30s sau lock tự free
→ pod khác acquire được

Ưu: đơn giản, không cần watchdog. Nhược: trong khoảng TTL, job bị dừng.

Strategy 2: Watchdog Re-Seed

watchdog tick mỗi 60s
nếu queue rỗng → seed token mới

Ưu: recovery nhanh hơn TTL. Nhược: false positive khi doWork chạy lâu.

Quy tắc vàng: watchdogInterval > maxDoWorkTime + buffer.

Strategy 3: Reliable Queue (Processing List)

BRPOPLPUSH queue processing  ← atomic move sang processing
doWork()
LREM processing            ← xóa khỏi processing

Watchdog: scan processing list, item nào quá lâu → move lại queue

Ưu: không mất task khi pod chết. Nhược: phức tạp, cần tune timeout.

Strategy 4: Heartbeat Renewal

acquire lock TTL=30s
mỗi 10s → renew TTL=30s
pod chết → không renew → 30s sau lock free

Ưu: lock không expire khi pod sống chậm. Nhược: nhiều round-trip Redis.

Bài Học: Không Có Silver Bullet

Mỗi strategy có trade-off:

Recovery time vs complexity
False positive vs false negative
Performance vs safety

Chọn theo business requirement:

Job critical, không được trùng → reliable queue + lock
Job idempotent, có thể trùng nhẹ → TTL auto-expire đủ
Job rate cao → tránh per-job lock, dùng leader election

9. Tổng Kết: Data Structure Driven Distributed Design

Nguyên Tắc Thiết Kế

1. Bắt đầu từ data structure:

Bài toán → Cấu trúc Redis phù hợp → Properties tự nhiên

2. Tận dụng atomicity built-in:

SET NX EX cho lock
ZADD NX cho dedup
LPOP/RPOP cho competing consumers
SADD cho distinct count

3. Hiểu rõ failure semantics:

Mỗi pattern phải hỏi: "Pod chết thì sao?", "2 pod race thì sao?"
Test concrete scenario, không assume

4. Per-pod vs shared là quyết định kiến trúc:

Per-pod: tốc độ, không race
Shared: consistency, recovery

5. Atomicity là spectrum:

Single command < Lua < MULTI/EXEC < App-level
Chọn mức phù hợp business requirement

6. Trade-off rõ ràng > Pattern clever:

Distributed lock chuẩn > queue-as-token clever
Recovery rõ ràng > false hope
Code readable > tinh tế khó hiểu

Anti-Pattern Trong Distributed Design

❌ Dùng Pub/Sub làm task queue — broadcast không phải competing

❌ Static field cho instance UUID — tất cả instance share

❌ Multi-step không atomic mà không lường race — eventually buggy

❌ TTL quá ngắn so với processing time — lock expire khi vẫn cần

❌ TTL quá dài so với recovery requirement — pod chết, job stuck lâu

❌ Queue dedup tin là mutex — không phải

❌ Release lock trong singleton job — chạy lặp lại

Pattern Tinh Túy

✅ Sorted Set + List = Delayed Queue — tận dụng score + atomic pop

✅ String + SET NX EX = Distributed Lock — atomic create, auto-expire

✅ Hash + Sorted Set buffer = Expiring Map — flexible với metadata

✅ Per-pod zset + shared list — tránh race ở mover, share ở consumer

✅ Lock + Queue + Watchdog — combination cho periodic job

✅ PutNX trong delayed queue — tránh anti-postpone bug

Câu Hỏi Cuối Cùng Cho Mọi Design Phân Tán

Trước khi commit design, hỏi:

Pod chết giữa chừng thì state đi đâu?
2 pod race ở step nào? Có thể bị duplicate không?
Network delay/jitter ảnh hưởng đến đâu?
TTL/timeout giá trị bao nhiêu là hợp lý?
Recovery time tối đa là bao nhiêu? Business chấp nhận không?
Operation nào atomic, operation nào không?
Có thể test concrete scenario không?

Distributed system không có magic — chỉ có understand trade-off và handle edge case explicit.

Bài viết tổng hợp từ kinh nghiệm xây dựng các pattern phân tán trên Redis (Pika) + Vert.x trong production. Mọi pattern đều xuất phát từ bug thực tế, race condition thật, và lessons learned từ hệ thống live.

Command Palette