Consumer Groups

ทดสอบบน Redis 8.10.2 (redis:8.10.2-alpine), ioredis 6.0.0, Node 24 บน Mac (Apple silicon) transcript แบบ # cli/... คือ session เดียวเรียงตามเอกสารที่ harness ของคอร์สรันซ้ำ ผมใช้ ID ของ entry แบบกำหนดเอง (1-0..5-0) เพื่อให้ session ทำซ้ำได้ และค่า idle เป็นมิลลิวินาทีจากการรันของผม ของคุณจึงต่างออกไป พื้นฐานของ stream อยู่ใน Streams

Consumer group คืออะไร

consumer group ให้ worker (consumer) หลายตัวอ่าน stream เดียวกัน โดย แต่ละ entry ไปถึง consumer ของ group เพียงตัวเดียว Redis จำต่อ group ว่า entry ไหนถูกส่งและส่งให้ใคร และเก็บทุก entry ที่ส่งแล้วแต่ยังไม่ acknowledge ไว้ใน Pending Entries List (PEL) จนกว่า consumer จะเรียก XACK การจดบันทึกนี้เองที่เปลี่ยน log ให้เป็น work queue ที่ประมวลผลแบบ at-least-once group ต่างๆ บน stream เดียวกันเป็นอิสระต่อกัน: แต่ละ group เห็นทุก entry

สร้าง group และอ่านในฐานะ consumer

XGROUP CREATE key group id [MKSTREAM] กำหนดจุดเริ่มของ group $ = เฉพาะ entry ที่เพิ่มจากนี้ไป; 0 = ตั้งแต่ต้น MKSTREAM สร้าง stream ว่างถ้ายังไม่มี XREADGROUP GROUP group consumer [COUNT n] STREAMS key > ส่ง entry ที่ group ไม่เคยส่งให้ใครเลย; consumer ถูกสร้างเมื่อใช้ครั้งแรก

cli/pubsub-streams-consumer-groups-1.redis

127.0.0.1:6379> XGROUP CREATE jobs workers $ MKSTREAM
OK
127.0.0.1:6379> XLEN jobs
(integer) 0
127.0.0.1:6379> XADD jobs 1-0 type email to ada
"1-0"
127.0.0.1:6379> XADD jobs 2-0 type sms to bob
"2-0"
127.0.0.1:6379> XADD jobs 3-0 type email to cy
"3-0"
127.0.0.1:6379> XADD jobs 4-0 type push to dee
"4-0"
127.0.0.1:6379> XADD jobs 5-0 type email to eve
"5-0"
127.0.0.1:6379> XGROUP CREATE jobs audit 0
OK
127.0.0.1:6379> XREADGROUP GROUP workers alice COUNT 2 STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "1-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "ada"
      2) 1) "2-0"
         2) 1) "type"
            2) "sms"
            3) "to"
            4) "bob"
127.0.0.1:6379> XREADGROUP GROUP workers bob COUNT 2 STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "3-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "cy"
      2) 1) "4-0"
         2) 1) "type"
            2) "push"
            3) "to"
            4) "dee"
127.0.0.1:6379> XREADGROUP GROUP workers alice COUNT 10 STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "5-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "eve"
127.0.0.1:6379> XREADGROUP GROUP workers alice COUNT 10 STREAMS jobs >
(nil)
127.0.0.1:6379> XREADGROUP GROUP workers alice COUNT 10 STREAMS jobs 0
1) 1) "jobs"
   2) 1) 1) "1-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "ada"
      2) 1) "2-0"
         2) 1) "type"
            2) "sms"
            3) "to"
            4) "bob"
      3) 1) "5-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "eve"
127.0.0.1:6379> XREADGROUP GROUP audit reader COUNT 1 STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "1-0"
         2) 1) "type"
            2) "email"
            3) "to"
            4) "ada"

session นั้นแสดงว่า:

• alice ได้ 1-0 และ 2-0, bob ได้ 3-0 และ 4-0 และการอ่านครั้งถัดไปของ alice ได้ 5-0: ไม่มี entry ไปถึงสอง consumer การอ่านด้วย > ครั้งต่อมาคืน (nil): ไม่มีอะไรใหม่
• > เทียบกับ 0 > หมายถึง “entry ใหม่ที่ไม่เคยส่ง” ID จริงเช่น 0 หมายถึง “ขอ entry pending ของฉันเอง หลัง ID นั้น”: alice ได้ 1-0, 2-0 และ 5-0 กลับมา ซึ่งเป็น entry ที่เธอได้รับและยังไม่ acknowledge นี่คือวิธีที่ consumer ที่ restart ทำงานต่อ (ในการรันนี้ การอ่านประวัติยังทำให้ตัวนับการส่งของ 5-0 เป็น 2 ดังที่เห็นด้านล่าง)
• group audit ที่สร้างที่ 0 ได้รับ 1-0 อีกครั้งโดยอิสระ: group ไม่แชร์ความคืบหน้า

XACK, XPENDING และ XINFO

XACK key group id [id ...] ลบ entry ออกจาก PEL และคืนจำนวนที่ลบ (0 ถ้าไม่ได้ pending อยู่ การ ack ซ้ำจึงไม่เป็นอันตราย) XPENDING key group สรุป PEL: จำนวน, ID pending ที่เล็กที่สุดและใหญ่ที่สุด และ pending ต่อ consumer รูปแบบขยาย XPENDING key group start end count แสดงแต่ละ entry พร้อม consumer, เวลา idle เป็น ms และ จำนวนครั้งที่ส่ง:

cli/pubsub-streams-consumer-groups-2.redis

127.0.0.1:6379> XPENDING jobs workers
1) (integer) 5
2) "1-0"
3) "5-0"
4) 1) 1) "alice"
      2) "3"
   2) 1) "bob"
      2) "2"
127.0.0.1:6379> XACK jobs workers 1-0 2-0
(integer) 2
127.0.0.1:6379> XACK jobs workers 1-0
(integer) 0
127.0.0.1:6379> XPENDING jobs workers - + 10
1) 1) "3-0"
   2) "bob"
   3) (integer) 518
   4) (integer) 1
2) 1) "4-0"
   2) "bob"
   3) (integer) 518
   4) (integer) 1
3) 1) "5-0"
   2) "alice"
   3) (integer) 335
   4) (integer) 2
127.0.0.1:6379> XINFO GROUPS jobs
1)  1) "name"
    2) "audit"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 1
    7) "last-delivered-id"
    8) "1-0"
    9) "entries-read"
   10) (integer) 1
   11) "lag"
   12) (integer) 4
2)  1) "name"
    2) "workers"
    3) "consumers"
    4) (integer) 2
    5) "pending"
    6) (integer) 3
    7) "last-delivered-id"
    8) "5-0"
    9) "entries-read"
   10) (integer) 5
   11) "lag"
   12) (integer) 0

XINFO GROUPS แสดงต่อ group pending, last-delivered-id, entries-read และ lag (entry ที่ยังไม่ส่ง: audit ตามหลัง 4, workers 0) XINFO CONSUMERS แสดงจำนวน pending และเวลา idle ของแต่ละ consumer ซึ่งเป็นวิธีจับ consumer ที่ตาย: pending ไม่ใช่ศูนย์และ idle โตขึ้น (หน่วยมิลลิวินาที ผลนี้จึงมาจากการรันของผม และของคุณจะต่างออกไป)

127.0.0.1:6379> XINFO CONSUMERS jobs workers
1) 1) "name"
   2) "alice"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 482
   7) "inactive"
   8) (integer) 604
2) 1) "name"
   2) "bob"
   3) "pending"
   4) (integer) 2
   5) "idle"
   6) (integer) 665
   7) "inactive"
   8) (integer) 665

XADD jobs 1-0 type "email" to "[email protected]"
XADD jobs 2-0 type "sms" to "+6681234567"
XGROUP CREATE jobs workers 0
XREADGROUP GROUP workers consumer-1 COUNT 10 STREAMS jobs >
XACK jobs workers 1-0
XPENDING jobs workers - + 10

Copy the sequence ⧉

กู้คืนจาก consumer ที่ตาย

ถ้า bob crash 3-0 และ 4-0 จะค้างใน PEL ภายใต้ชื่อเขาตลอดไปเว้นแต่มีใครรับไป XAUTOCLAIM key group consumer min-idle-time start [COUNT n] (6.2+) โอนความเป็นเจ้าของ entry ที่ pending มาอย่างน้อย min-idle-time ms และคืน entry พร้อม cursor สำหรับการเรียกถัดไป XCLAIM ทำเช่นเดียวกันสำหรับ ID ที่ระบุ ใน session ด้านล่าง alice acknowledge 5-0 ของตัวเองก่อน แล้วลอง claim ด้วย min-idle-time 60000 ซึ่งไม่พบอะไรเพราะ entry เพิ่งผ่านมาไม่กี่วินาที แล้ว claim ด้วย 0 (เลือกเพื่อให้ transcript แน่นอน; ค่าใน production คือหลายวินาทีถึงหลายนาที) ซึ่งรับ entry ทั้งสองของ bob:

cli/pubsub-streams-consumer-groups-3.redis

127.0.0.1:6379> XACK jobs workers 5-0
(integer) 1
127.0.0.1:6379> XAUTOCLAIM jobs workers alice 60000 0-0 COUNT 10
1) "0-0"
2) (empty array)
3) (empty array)
127.0.0.1:6379> XAUTOCLAIM jobs workers alice 0 0-0 COUNT 10
1) "0-0"
2) 1) 1) "3-0"
      2) 1) "type"
         2) "email"
         3) "to"
         4) "cy"
   2) 1) "4-0"
      2) 1) "type"
         2) "push"
         3) "to"
         4) "dee"
3) (empty array)
127.0.0.1:6379> XPENDING jobs workers - + 10
1) 1) "3-0"
   2) "alice"
   3) (integer) 53
   4) (integer) 2
2) 1) "4-0"
   2) "alice"
   3) (integer) 53
   4) (integer) 2
127.0.0.1:6379> XACK jobs workers 3-0 4-0
(integer) 2
127.0.0.1:6379> XPENDING jobs workers
1) (integer) 0
2) (nil)
3) (nil)
4) (nil)
127.0.0.1:6379> XREADGROUP GROUP audit reader COUNT 1 NOACK STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "2-0"
         2) 1) "type"
            2) "sms"
            3) "to"
            4) "bob"
127.0.0.1:6379> XPENDING jobs audit
1) (integer) 1
2) "1-0"
3) "1-0"
4) 1) 1) "reader"
      2) "1"
127.0.0.1:6379> XGROUP DELCONSUMER jobs workers bob
(integer) 0
127.0.0.1:6379> XGROUP DESTROY jobs audit
(integer) 1

สังเกตตัวนับการส่ง: entry ที่ claim ตอนนี้เป็น 2 ตัวนับที่โตไปเรื่อยๆ คือสัญญาณ poison message; ให้ dead-letter entry ที่เกินเกณฑ์แทนการลองซ้ำตลอดไป NOACK บน XREADGROUP ข้าม PEL ไปเลย (audit อ่าน 2-0 และ XPENDING jobs audit ยังแสดงแค่ 1-0 ก่อนหน้า): ใช้เมื่อยอมเสีย entry ตอน crash ได้เท่านั้น

worker ใน Node.js: crash, กู้คืน และกับดัก ack-ก่อน

สคริปต์ด้านล่างเล่นทั้งหมดใน ioredis worker A อ่านสองงานแล้วตาย worker B ประมวลผลที่เหลือ โดย acknowledge หลัง ประมวลผล เมื่อเกินเกณฑ์ idle XAUTOCLAIM ให้งานของ A แก่ B บล็อกสุดท้ายทำ crash เดิมซ้ำโดย XACK ก่อน ทำงาน:

src/pubsub-streams-consumer-groups-worker.ts

// @run
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);
const stream = 'b2:orders';
const group = 'fulfil';
await redis.del(stream);
await redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
for (let i = 1; i <= 5; i++) await redis.xadd(stream, '*', 'order', `o-${i}`);

type Entry = [id: string, fields: string[]];
// XREADGROUP / XAUTOCLAIM return nested arrays; the entries part is [[id, [field, value, ...]], ...]
async function readNew(consumer: string, count: number): Promise<Entry[]> {
  const res = await redis.xreadgroup('GROUP', group, consumer, 'COUNT', count, 'BLOCK', 200, 'STREAMS', stream, '>');
  return res ? (res[0]![1] as Entry[]) : [];
}
const processOrder = async (e: Entry) => console.log('  processed', e[1][1]);

// worker A takes two jobs and "crashes": no processing, no XACK
const lost = await readNew('worker-a', 2);
console.log('worker-a read', lost.length, 'and died; pending:', (await redis.xpending(stream, group))[0]);

// worker B handles the rest: process first, THEN acknowledge
for (let batch = await readNew('worker-b', 10); batch.length; batch = await readNew('worker-b', 10)) {
  for (const e of batch) { await processOrder(e); await redis.xack(stream, group, e[0]); }
}
console.log('worker-b done; still pending (worker-a\'s jobs):', (await redis.xpending(stream, group))[0]);

// recovery: claim entries idle for more than 300 ms (in production 30 s or more), process, ack
await new Promise((r) => setTimeout(r, 400));
const [, claimed] = (await redis.xautoclaim(stream, group, 'worker-b', 300, '0-0', 'COUNT', 10)) as [string, Entry[], string[]];
for (const e of claimed) { await processOrder(e); await redis.xack(stream, group, e[0]); }
console.log('reclaimed', claimed.length, '-> pending now:', (await redis.xpending(stream, group))[0]);

// ...the same crash if you ACK BEFORE working: nothing is left to recover
const s2 = 'b2:orders-ackfirst';
await redis.del(s2);
await redis.xgroup('CREATE', s2, group, '0', 'MKSTREAM');
await redis.xadd(s2, '*', 'order', 'o-99');
const res = await redis.xreadgroup('GROUP', group, 'worker-c', 'STREAMS', s2, '>');
await redis.xack(s2, group, (res![0]![1] as Entry[])[0]![0]);      // ack first...
/* worker-c crashes here, before doing the work */
await new Promise((r) => setTimeout(r, 400));
const [, gone] = (await redis.xautoclaim(s2, group, 'worker-d', 300, '0-0')) as [string, Entry[], string[]];
console.log('ack-first: job o-99 recoverable?', gone.length > 0, '| pending:', (await redis.xpending(s2, group))[0]);

await redis.del(stream, s2);
await redis.quit();

Output จริง (Mac เครื่องนี้):

worker-a read 2 and died; pending: 2
  processed o-3
  processed o-4
  processed o-5
worker-b done; still pending (worker-a's jobs): 2
  processed o-1
  processed o-2
reclaimed 2 -> pending now: 0
ack-first: job o-99 recoverable? false | pending: 0

worker จริงรันสิ่งนี้ใน loop พร้อม BLOCK และเรียก XAUTOCLAIM เป็นระยะด้วย min-idle-time ที่ยาวกว่างานที่ช้าที่สุดที่ชอบธรรมของคุณ ioredis คืน reply แบบ array ซ้อนตามที่ type ระบุ ([[id, [field, value, ...]], ...] หรือ null เมื่อ BLOCK หมดเวลา)

🟥 Consumer group ให้ at-least-once ดังนั้นทำให้งานเป็น idempotent

งานอาจรันสองครั้ง: worker ที่ช้าซึ่ง entry ถูก auto-claim ไปยังทำงานต่อขณะที่เจ้าของใหม่เริ่มมันด้วย PEL รับประกันว่า crash จะไม่ทำให้ entry หายเงียบๆ ไม่ใช่ว่ามันรันครั้งเดียว ใช้ idempotency key (ID ของ entry เหมาะมาก) ในสิ่งที่งานเขียน

Reading AI-generated code

AI เขียน loop ของ consumer สำหรับ stream การจัดส่งออเดอร์ มัน acknowledge ก่อน “เพื่อไม่ให้ประมวลผลซ้ำ”:

🐛 หา bug ในโค้ดนี้

for (const [id, fields] of entries) {
await redis.xack('orders', 'fulfil', id);   // ack first: avoid duplicates
await shipOrder(fields);                    // may throw or the process may crash
}

เฉลย

การ acknowledge ลบ entry ออกจาก PEL ดังนั้นถ้า process ตาย (หรือ shipOrder โยน error) หลัง XACK จะไม่เหลืออะไรให้ XAUTOCLAIM พบ ในบล็อกสุดท้ายของสคริปต์ งาน o-99 แบบ ack-ก่อนจบด้วย recoverable? false | pending: 0: มันหายไป ขณะที่งานแบบ ack-หลังทั้งหมดถูกกู้คืน ประมวลผลก่อน acknowledge ทีหลัง และทำงานให้ idempotent เพื่อรับมือกับตัวซ้ำที่เกิดไม่บ่อย:

for (const [id, fields] of entries) {
  await shipOrder(fields);                    // idempotent: keyed by the entry id
  await redis.xack('orders', 'fulfil', id);   // ack only once the work is done
}

Mental Model

consumer group คือ คิวที่มีใบเสร็จ: XREADGROUP ส่ง entry ให้คุณและเขียนชื่อคุณลงใน PEL; XACK ฉีกใบเสร็จ ถ้าคุณหายไป ใบเสร็จจะค้างอยู่และเก่าลงเรื่อยๆ จนกว่า XAUTOCLAIM ของ worker อื่นรับช่วงไป สิ่งที่ group ให้คือใบเสร็จ ความถูกต้องมาจากการ acknowledge หลังทำงานและการทำงานแบบ idempotent

Trade-offs

ตัวเลือก	ข้อดี	ข้อเสีย
consumer group แทน XREAD ธรรมดา	ขยายแนวนอนได้ แต่ละ entry ไป consumer เดียว จดบันทึกที่ปลอดภัยต่อ crash	ต้องจัดการ XACK, การ reclaim และ dead-letter เอง
at-least-once ผ่าน PEL	ไม่มีการสูญหายเงียบๆ เมื่อ crash	อาจซ้ำหลัง reclaim; งานต้อง idempotent
XAUTOCLAIM ด้วย min-idle-time เล็ก	กู้คืนเร็ว	แย่ง entry จาก worker ที่ช้าแต่ยังมีชีวิต ทำให้ทำงานซ้ำ
NOACK	ไม่มีต้นทุน PEL	crash ทำให้ entry หาย (เหมือน Pub/Sub)
หลาย group บน stream เดียว	ผู้บริโภคอิสระของ event ชุดเดียวกัน	stream ต้องเก็บ entry จนกว่า group ที่ช้าที่สุดจะเสร็จ (trim อย่างระวัง)

Common Misconceptions

• “Consumer group ให้ exactly-once” ให้ at-least-once entry ที่ reclaim อาจรันสองครั้ง
• “XACK ลบ entry” มันลบออกจาก PEL ของ group เท่านั้น entry ยังอยู่ใน stream (ใช้ XACKDEL/XDELEX 8.2+ หรือ trim)
• “entry ของ consumer ที่ตายถูกย้ายให้อัตโนมัติ” ไม่มีอะไรย้ายจนกว่าจะมี worker รัน XAUTOCLAIM/XCLAIM
• “> และ 0 ใช้แทนกันได้” > คือ entry ใหม่; ID จริงเล่น entry pending ของคุณเองซ้ำ

Common mistakes

• XACK ก่อนงานเสร็จ — crash แล้วงานหาย (แสดงข้างบน)
• ไม่เคยเรียก XACK — PEL โตตลอดไปและ XINFO GROUPS แสดง pending ที่เพิ่มขึ้น
• ไม่มีกระบวนการ reclaim — entry ของ consumer ที่ตายไม่ถูกลองใหม่
• ลอง poison message ซ้ำไม่จบ — ดูตัวนับการส่งใน XPENDING และ dead-letter เมื่อเกินขีดจำกัด
• งานที่ไม่ idempotent — at-least-once หมายถึงตัวซ้ำเกิดขึ้นได้

Real-world example

เอกสารของ Redis เอง อธิบาย PEL, การกู้คืน consumer ที่ตายด้วย XAUTOCLAIM และความหมายแบบ at-least-once ใน Redis Streams (หัวข้อ “Recovering from permanent failures”)

XREADGROUP GROUP g consumer STREAMS s > หมายความว่าอย่างไร?

ส่ง entry ใหม่ที่ไม่เคยถูกส่งให้ consumer นี้ ส่ง entry ที่ pending ทั้งหมดซ้ำตั้งแต่ต้น acknowledge ทุก entry ใน group ลบ consumer group

เกิดอะไรขึ้นกับ entry ใน stream หลังเรียก XACK?

ถูกลบออกจาก stream ทั้งหมด ถูกย้ายไป dead-letter stream ถูกลบออกจาก Pending Entries List ของ group ถูกส่งซ้ำให้ consumer อื่น

XPENDING แสดงอะไร?

ทุก entry ที่เคยเขียนลง stream entry ที่ส่งให้ consumer ของ group แล้วแต่ยังไม่ acknowledge entry ที่รอการส่ง จำนวน consumer group ที่ active

ใน XGROUP CREATE jobs workers 0 MKSTREAM เลข 0 หมายถึงอะไร?

ส่งครั้งละศูนย์ entry เริ่มจาก entry ล่าสุดเท่านั้น เริ่มอ่านจากจุดเริ่มต้นของ stream สร้างศูนย์ consumer ในตอนแรก

worker เรียก XACK ก่อนแล้ว crash ก่อนทำงานเสร็จ XAUTOCLAIM พบอะไร?

งานนั้น พร้อมตัวนับการส่ง 2 ไม่พบอะไร: entry ออกจาก PEL ไปแล้ว งานจึงหาย งานนั้นใน dead-letter stream error

consumer ตายทั้งที่มี 3 entry pending คำสั่งไหนทำให้ worker อื่นรับช่วง?

XGROUP DELCONSUMER XAUTOCLAIM พร้อม min-idle-time XREADGROUP กับ > XTRIM

ตัวนับการส่งของ entry ใน XPENDING เพิ่มขึ้นเรื่อยๆ โดยทั่วไปหมายความว่า?

entry นั้นได้รับความนิยม poison message ที่ล้มเหลวซ้ำๆ: dead-letter เมื่อเกินขีดจำกัด stream ต้อง trim group ไม่ sync

Submit

Mark as complete