Streams

ทดสอบบน Redis 8.10.2 (redis:8.10.2-alpine), ioredis 6.0.0, Node 24 บน Mac (Apple silicon) transcript แบบ # cli/... คือ session เดียวเรียงตามเอกสารที่ harness ของคอร์สรันซ้ำ ID ที่สร้างด้วย * คือ timestamp จึง ของคุณจะต่างออกไป เอกสารอ้างอิง Streams: Redis Streams docs

Streams — log แบบ append-only

Redis Stream คือโครงสร้างข้อมูลที่ทำงานเหมือน log แบบ append-only ที่เรียงลำดับ ต่างจาก Pub/Sub ที่ข้อความหายไปทันทีที่ส่งแล้ว stream เก็บ entry ไว้ และให้ consumer ใดๆ อ่านซ้ำได้จากตำแหน่งใดก็ได้ consumer ที่มาช้าหรือ restart จึงตามทันได้

แต่ละ entry มีสองส่วน:

ID — <milliseconds>-<sequence> ไม่ซ้ำและเพิ่มขึ้นอย่างเคร่งครัดภายใน stream ส่ง * เพื่อให้ Redis สร้างจากนาฬิกาของเซิร์ฟเวอร์ หรือกำหนดเอง
Fields — คู่ field-value หนึ่งคู่ขึ้นไป คล้าย hash เล็กๆ

entry อยู่ที่ไหน? ในหน่วยความจำ เหมือนทุก type ของ Redis: เอกสารอธิบายว่าเป็น radix tree ของ macro node ที่แต่ละอันเก็บหลาย entry มันถึงดิสก์ผ่านการ persist ที่คุณตั้งค่า (snapshot ของ RDB, AOF) เท่านั้น เหมือน list หรือ hash “Stream” หมายถึง log ในหน่วยความจำ ไม่ใช่ไฟล์บนดิสก์ และหัวข้อเรื่องขนาดด้านล่างแสดงว่ามีต้นทุนเท่าไร

XADD — เพิ่ม entry

XADD key id field value [field value ...] เพิ่ม entry และคืน ID ของมัน เมื่อใช้ * ID คือ <ms ปัจจุบัน>-0 (sequence เพิ่มเมื่อหลาย entry อยู่ใน millisecond เดียวกัน) XLEN นับ entry, TYPE ตอบ stream:

127.0.0.1:6379> XADD events * action "login" user "ada"
"1790750093107-0"
127.0.0.1:6379> XADD events * action "purchase" user "ada" item "book"
"1790750093167-0"
127.0.0.1:6379> XLEN events
(integer) 2
127.0.0.1:6379> TYPE events
stream
127.0.0.1:6379> XRANGE events - +
1) 1) "1790750093107-0"
   2) 1) "action"
      2) "login"
      3) "user"
      4) "ada"
2) 1) "1790750093167-0"
   2) 1) "action"
      2) "purchase"
      3) "user"
      4) "ada"
      5) "item"
      6) "book"
127.0.0.1:6379> XRANGE events - + COUNT 1
1) 1) "1790750093107-0"
   2) 1) "action"
      2) "login"
      3) "user"
      4) "ada"
127.0.0.1:6379> XREVRANGE events + - COUNT 1
1) 1) "1790750093167-0"
   2) 1) "action"
      2) "purchase"
      3) "user"
      4) "ada"
      5) "item"
      6) "book"
127.0.0.1:6379> XREAD COUNT 10 STREAMS events 0
1) 1) "events"
   2) 1) 1) "1790750093107-0"
         2) 1) "action"
            2) "login"
            3) "user"
            4) "ada"
      2) 1) "1790750093167-0"
         2) 1) "action"
            2) "purchase"
            3) "user"
            4) "ada"
            5) "item"
            6) "book"

XRANGE key start end [COUNT n] อ่านจากเก่าไปใหม่ โดยมี - (เล็กสุด) และ + (ใหญ่สุด) เป็นปลายเปิด; XREVRANGE key end start ไปจากใหม่สุดก่อน XREAD [COUNT n] STREAMS key id คืน entry หลัง ID ที่คุณส่ง (0 = จากจุดเริ่มต้น) แต่ละ reply คือ ID ของ entry บวกรายการ [field, value, ...] แบบ flat

XADD events * action "login" user "ada"
XADD events * action "purchase" user "ada" item "book"
XLEN events
XRANGE events - +
XREAD COUNT 10 STREAMS events 0

กฎของ ID, การ trim และการลบ

ID ต้องเพิ่มขึ้นอย่างเคร่งครัด และกำหนดบางส่วนของมันได้ (1000-* ให้ Redis เลือก sequence) การลบและการ trim เป็นคนละการดำเนินการ:

XDEL key id ลบหนึ่ง entry แต่ทิ้ง tombstone ไว้ XLEN จึงลดลง และ stream จำ ID สูงสุดที่ถูกลบ (max-deleted-entry-id ใน XINFO STREAM)
XTRIM key MAXLEN n (เก็บ n ตัวใหม่สุด) หรือ MINID id (ทิ้งทุกอย่างที่เก่ากว่า) ลบจากหัว XADD key MAXLEN = n ... trim ระหว่างเพิ่ม NOMKSTREAM ป้องกันไม่ให้ XADD สร้าง stream

127.0.0.1:6379> XADD fixed 1000-1 a 1
"1000-1"
127.0.0.1:6379> XADD fixed 1000-1 a 2
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
127.0.0.1:6379> XADD fixed 999-0 a 3
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
127.0.0.1:6379> XADD fixed 1000-* a 4
"1000-2"
127.0.0.1:6379> XADD log 1-0 n 1
"1-0"
127.0.0.1:6379> XADD log 2-0 n 2
"2-0"
127.0.0.1:6379> XADD log 3-0 n 3
"3-0"
127.0.0.1:6379> XADD log 4-0 n 4
"4-0"
127.0.0.1:6379> XADD log 5-0 n 5
"5-0"
127.0.0.1:6379> XADD log 6-0 n 6
"6-0"
127.0.0.1:6379> XDEL log 3-0
(integer) 1
127.0.0.1:6379> XLEN log
(integer) 5
127.0.0.1:6379> XRANGE log 2-0 5-0
1) 1) "2-0"
   2) 1) "n"
      2) "2"
2) 1) "4-0"
   2) 1) "n"
      2) "4"
3) 1) "5-0"
   2) 1) "n"
      2) "5"
127.0.0.1:6379> XTRIM log MAXLEN 4
(integer) 1
127.0.0.1:6379> XRANGE log - +
1) 1) "2-0"
   2) 1) "n"
      2) "2"
2) 1) "4-0"
   2) 1) "n"
      2) "4"
3) 1) "5-0"
   2) 1) "n"
      2) "5"
4) 1) "6-0"
   2) 1) "n"
      2) "6"
127.0.0.1:6379> XTRIM log MINID 5-0
(integer) 2
127.0.0.1:6379> XADD log MAXLEN = 2 7-0 n 7
"7-0"
127.0.0.1:6379> XRANGE log - +
1) 1) "6-0"
   2) 1) "n"
      2) "6"
2) 1) "7-0"
   2) 1) "n"
      2) "7"
127.0.0.1:6379> XADD nostream NOMKSTREAM * a 1
(nil)
127.0.0.1:6379> EXISTS nostream
(integer) 0

ทุกบรรทัดข้างบนคือ output จริง MAXLEN 4 ลบ 1 entry (stream มี 5 หลัง XDEL) และ MINID 5-0 ลบสอง entry ที่เก่ากว่า

trim แบบแม่นยำเทียบกับแบบประมาณ MAXLEN ~ n ได้รับอนุญาตให้หยุดก่อน เพราะ Redis คืน internal node ทั้งก้อน (stream-node-max-entries ค่าเริ่มต้น 100) บน stream 1,000 entry XTRIM s MAXLEN ~ 250 ลบ 700 และเหลือ 300; จากนั้น XTRIM s MAXLEN 250 แบบแม่นยำลบเพิ่มอีก 50 จนถึง 250 แบบประมาณถูกกว่าและเป็นสิ่งที่ต้องการบน hot path: XADD key MAXLEN ~ 10000 * ...

XREAD BLOCK จาก Node.js

XREAD BLOCK ms STREAMS key $ รอ entry ที่ใหม่กว่า “ตอนนี้” ($) และ BLOCK 0 รอตลอดไป คำสั่งที่บล็อกยึด connection ไว้ จึงให้ reader มี connection ของตัวเอง reply เป็น null เมื่อหมดเวลาโดยไม่มีอะไรใหม่:

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

const url = process.env.REDIS_URL!;
const redis = new Redis(url);
const reader = new Redis(url);   // XREAD BLOCK holds its connection, so it gets its own
const key = 'b2:events';
await redis.del(key);

// XADD returns the generated ID: "<ms timestamp>-<sequence>"
const id1 = await redis.xadd(key, '*', 'action', 'login', 'user', 'ada');
console.log('first id:', id1, ' has ms-seq shape:', /^\d{13}-\d+$/.test(id1!));

// a reader that blocks for up to 2 s waiting for entries newer than "$"
const waiting = reader.xread('BLOCK', 2000, 'STREAMS', key, '$');
await new Promise((r) => setTimeout(r, 100));
await redis.xadd(key, '*', 'action', 'purchase', 'user', 'ada', 'item', 'book');
const res = await waiting;   // [ [ streamKey, [ [ id, [field, value, ...] ], ... ] ] ]  or null on timeout
const [, entries] = res![0]!;
const [id2, fields] = entries[0]!;
console.log('blocked reader got', id2, fields);                       // [ 'action', 'purchase', 'user', 'ada', 'item', 'book' ]

// fields are a FLAT [field, value, ...] array: turn it into an object
const obj: Record<string, string> = {};
for (let i = 0; i < fields.length; i += 2) obj[fields[i]!] = fields[i + 1]!;
console.log(obj);

// nothing new within the timeout -> null, not an error
console.log(await reader.xread('BLOCK', 200, 'STREAMS', key, '$'));   // null

// bound the stream while appending; ~ trims in whole internal nodes, so the length lands near the target, not on it
for (let i = 0; i < 1000; i++) await redis.xadd(key, 'MAXLEN', '~', 250, '*', 'n', i);
console.log('length after 1000 appends with MAXLEN ~ 250:', await redis.xlen(key));   // 250..349

// XDEL leaves a tombstone: XLEN drops, XINFO remembers the highest deleted id
const [firstId] = (await redis.xrange(key, '-', '+', 'COUNT', 1))[0]!;
await redis.xdel(key, firstId);
const info = await redis.xinfo('STREAM', key) as (string | number)[];
const at = (name: string) => info[info.indexOf(name) + 1];
console.log('XINFO length:', at('length'), ' max-deleted-entry-id is the id we deleted:', at('max-deleted-entry-id') === firstId);

await redis.del(key);
reader.disconnect();
await redis.quit();

Output จริง (Mac เครื่องนี้; ID ต่างกันในแต่ละครั้ง):

first id: 1790750068817-0  has ms-seq shape: true
blocked reader got 1790750068920-0 [ 'action', 'purchase', 'user', 'ada', 'item', 'book' ]
{ action: 'purchase', user: 'ada', item: 'book' }
null
length after 1000 appends with MAXLEN ~ 250: 302
XINFO length: 301  max-deleted-entry-id is the id we deleted: true

XDELEX และ XACKDEL (8.2+)

เมื่อใช้ consumer group (บทถัดไป) entry อาจถูกลบออกจาก stream ขณะที่ group ยังมี ID ของมันเป็น pending เหลือ reference ที่ค้างอยู่ Redis 8.2 เพิ่ม XDELEX key [KEEPREF|DELREF|ACKED] IDS n id... และ XACKDEL key group IDS n id... เพื่อลบอย่างปลอดภัย ACKED ปฏิเสธการลบ entry ที่บาง group ยังไม่ acknowledge reply ต่อ ID: 1 ลบแล้ว, 2 ไม่ลบเพราะยังมี reference, -1 ไม่มี entry นั้น (XDELEX) ต้องใช้ Redis 8.2+ (ไม่มีบน Redis 7); ผมไม่ได้ทดสอบ Valkey

127.0.0.1:6379> XADD jobs 1-0 task a
"1-0"
127.0.0.1:6379> XADD jobs 2-0 task b
"2-0"
127.0.0.1:6379> XADD jobs 3-0 task c
"3-0"
127.0.0.1:6379> XGROUP CREATE jobs workers 0
OK
127.0.0.1:6379> XREADGROUP GROUP workers w1 COUNT 3 STREAMS jobs >
1) 1) "jobs"
   2) 1) 1) "1-0"
         2) 1) "task"
            2) "a"
      2) 1) "2-0"
         2) 1) "task"
            2) "b"
      3) 1) "3-0"
         2) 1) "task"
            2) "c"
127.0.0.1:6379> XDELEX jobs ACKED IDS 1 1-0
1) (integer) 2
127.0.0.1:6379> XACK jobs workers 1-0
(integer) 1
127.0.0.1:6379> XDELEX jobs ACKED IDS 1 1-0
1) (integer) 1
127.0.0.1:6379> XACKDEL jobs workers IDS 1 2-0
1) (integer) 1
127.0.0.1:6379> XRANGE jobs - +
1) 1) "3-0"
   2) 1) "task"
      2) "c"
127.0.0.1:6379> XPENDING jobs workers
1) (integer) 1
2) "3-0"
3) "3-0"
4) 1) 1) "w1"
      2) "1"

XDELEX ACKED กับ 1-0 ที่ยังไม่ acknowledge ตอบ 2 และเก็บไว้; หลัง XACK ตอบ 1 XACKDEL acknowledge และลบในขั้นตอน atomic เดียว entry 3-0 ถูกส่งให้ w1 แต่ไม่เคย acknowledge จึงยังอยู่ใน stream และใน pending list

Reading AI-generated code

AI เพิ่มการติดตามการคลิกให้ API ทุก request เพิ่มลง stream ที่งานวิเคราะห์ดึงออกวันละครั้ง:

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

app.use(async (req, _res, next) => {
await redis.xadd('clicks', '*', 'path', req.path, 'user', req.user?.id ?? 'anon');
next();
});

เฉลย

ไม่มีอะไร trim clicks เลย และ Stream อยู่ใน RAM ผมวัด stream ของ entry เล็กๆ (สาม field ต่อ entry) ด้วย MEMORY USAGE:

Entry	หน่วยความจำ
1,000	23,597 byte
10,000	240,726 byte
100,000	2,510,255 byte
100,000 ที่เพิ่มด้วย `MAXLEN ~ 1000`	25,753 byte (`XLEN` 1000)

ราว 25 byte ต่อ entry ที่หน้าตาแบบนี้ เป็นเส้นตรงตลอดไป ที่ traffic จริงนั่นคือ memory leak ช้าๆ ที่จบด้วย eviction หรือ OOM command not allowed ให้ตั้งเพดานตอนเขียนด้วย MAXLEN แบบประมาณตามช่วงที่ consumer ล่มนานที่สุด (หรือ MINID สำหรับ “เก็บ 24 ชั่วโมง”) และให้งานรายวันอ่านด้วย XREAD/XRANGE จาก ID สุดท้ายที่ประมวลผลแล้ว:

await redis.xadd('clicks', 'MAXLEN', '~', 100000, '*', 'path', req.path, 'user', req.user?.id ?? 'anon');

เอกสารระบุแนวคิดเดียวกันไว้ใต้ “Capped Streams” pubsub-streams-streams-basic.ts ที่รันได้ข้างบนแสดงการเพิ่มแบบมีเพดาน (MAXLEN ~ 250 ให้ 302 entry หลังเพิ่ม 1,000 ครั้ง)

Mental Model

stream คือ ไฟล์ log ที่อยู่ใน RAM: เพิ่มที่ท้าย อ่านจากตำแหน่งใดก็ได้ด้วย ID ทิ้งจากหัว มันแก้สิ่งที่ Pub/Sub ทำไม่ได้ (ตามทัน, replay) และรับต้นทุนของ log ในหน่วยความจำ (คุณตัดสินใจว่าจะเก็บประวัติเท่าไร) ID คือทั้งลำดับและบุ๊กมาร์กที่ consumer เก็บไว้

Trade-offs

ตัวเลือก	ข้อดี	ข้อเสีย
Streams เก็บ entry	อ่านซ้ำได้ consumer ที่มาช้าและ restart ตามทัน	หน่วยความจำ: ราว 25 byte ต่อ entry เล็ก ไม่จำกัดถ้าไม่ตั้งเพดาน
`XADD` กับ `*`	ID เรียงตามเวลาโดยไม่ต้องดูแล	ผูกกับนาฬิกาเซิร์ฟเวอร์; กำหนด ID เองถ้าต้องการความหมายทางธุรกิจ
`MAXLEN ~ n`	trim ถูกบนเส้นทางเขียน	อาจเก็บเกิน n ได้ถึงหนึ่ง node (100 entry); ใช้ `MAXLEN` แบบแม่นยำเมื่อสำคัญ
`XREAD BLOCK`	ส่งแบบใกล้เคียง push โดยไม่ต้อง poll	ยึด connection; ต้องจัดการ timeout และ reconnect; ไม่มี ack (ใช้ consumer group)
`XDEL`	ลบหนึ่ง entry	tombstone ยังอยู่; group อาจเหลือ reference ค้าง; ควรใช้ `XDELEX`/`XACKDEL`

Common Misconceptions

“stream เก็บทุกอย่างบนดิสก์” มันอยู่ในหน่วยความจำ; ดิสก์ผ่าน RDB/AOF เท่านั้น และประวัติคือสิ่งที่คุณไม่ได้ trim
“XLEN บอกว่าเคยเพิ่ม entry กี่ตัว” มันบอก entry ที่มีอยู่ตอนนี้; XINFO STREAM มี entries-added
“MAXLEN ~ 250 เหลือ 250 เป๊ะ” เหลือ 300 (ตัวอย่าง 1,000 entry); ใช้แบบแม่นยำเมื่อต้องการเป๊ะ
“Streams ให้การส่งแบบ exactly-once” consumer group ให้ at-least-once; ประมวลผลแบบ idempotent

Common mistakes

ไม่มี MAXLEN/MINID บน stream ที่ร้อน — หน่วยความจำโตเป็นเส้นตรง (ดูแบบฝึกหัด)
XRANGE - + โดยไม่มี COUNT บน stream ใหญ่ — แบ่งหน้าด้วย COUNT และ ID สุดท้ายที่เห็น
ใช้ connection ของ XREAD ที่บล็อกร่วมกับคำสั่งอื่น — connection ไม่ว่างจนกว่าจะ timeout
XDEL กับ entry ที่ group ยังติดตามอยู่ — ใช้ XDELEX ACKED (8.2+)

Real-world example

เอกสารของ Redis เอง วาง Streams เป็นทางเลือกที่คงทนและ replay ได้แทน Pub/Sub และระบุ “Capped Streams” (MAXLEN/MINID) เป็นวิธีจำกัดขนาด (Redis Streams)