Pub/Sub

ทดสอบบน Redis 8.10.2 (redis:8.10.2-alpine), ioredis 6.0.0, Node 24 บน Mac (Apple silicon) transcript แบบ # cli/... ถูก harness ของคอร์สรันซ้ำ ส่วน output ของ subscriber และสคริปต์ @run สองตัวเป็นการรันจริง (redis-cli ที่ subscribe อยู่จะบล็อก จึงเป็น transcript ที่รันซ้ำไม่ได้) ความหมายของการส่ง: เอกสาร Redis Pub/Sub

Pub/Sub ทำงานอย่างไร

client เรียก SUBSCRIBE channel แล้วเริ่มฟัง ใน RESP2 หลังจากนั้นมันออกได้เฉพาะคำสั่ง subscription; ใน RESP3 รันอะไรก็ได้ (หัวข้อถัดไป)
client อื่นเรียก PUBLISH channel message Redis push ให้ทุก subscriber ที่เชื่อมต่ออยู่ ในขณะนั้น และคืนจำนวนของพวกเขา
ไม่มีอะไรถูกเก็บ ถ้าไม่มี subscriber ข้อความถูกทิ้ง และ subscriber ที่เชื่อมต่อกลับมาจะไม่ได้อะไรที่ส่งระหว่างที่ไม่อยู่ เอกสารเรียกว่าการส่งแบบ at-most-once

channel เป็นแค่ชื่อ: มีอยู่ตราบที่มีคน subscribe และไม่ต้องสร้าง Pub/Sub ไม่ใช่ข้อมูล keyspace ด้วย: มันอยู่นอก dataset จึงไม่มีอะไรถูกเขียนลง RDB/AOF และมันไม่สน SELECT (ผม publish จาก db 5 ด้วย ioredis และ subscriber บน db 0 ได้รับข้อความ)

สิ่งที่ subscriber เห็น

redis-cli ที่ subscribe อยู่ทำเป็น transcript ที่รันซ้ำไม่ได้เพราะมันบล็อก นี่จึงเป็นการจับของจริง: SUBSCRIBE news sports (รันด้วย timeout 3 และ --no-raw) ใน shell หนึ่ง และ PUBLISH สองครั้งจากอีก shell (แต่ละครั้งคืน 1):

$ redis-cli --no-raw SUBSCRIBE news sports
1) "subscribe"
2) "news"
3) (integer) 1
1) "subscribe"
2) "sports"
3) (integer) 2
1) "message"
2) "news"
3) "Breaking: Redis is fast"
1) "message"
2) "sports"
3) "goal"

แต่ละการยืนยันคือ [kind, channel, จำนวน subscription ทั้งหมดของ connection นี้] หนึ่งอันต่อ channel (ตัวนับเป็น 1, 2) แต่ละ push คือ ["message", channel, payload]

ฝั่ง publish ไม่จำเป็นต้องมี subscriber และ PUBSUB ตรวจดูสถานะ ตอนนี้ไม่มีใครฟัง ทุกอย่างจึงเป็นศูนย์:

127.0.0.1:6379> PUBLISH news "nobody is listening"
(integer) 0
127.0.0.1:6379> PUBSUB CHANNELS
(empty array)
127.0.0.1:6379> PUBSUB NUMSUB news
1) "news"
2) (integer) 0
127.0.0.1:6379> PUBSUB NUMPAT
(integer) 0
127.0.0.1:6379> SPUBLISH orders "o-1"
(integer) 0

PUBLISH คืน 0: ข้อความไม่ไปไหนและหายไปแล้ว PUBSUB CHANNELS แสดง channel ที่มี subscriber อย่างน้อยหนึ่ง, PUBSUB NUMSUB ch นับ subscriber ของ channel (ไม่รวม pattern subscriber) และ PUBSUB NUMPAT นับ pattern subscription

PUBLISH news "Breaking: Redis is fast"
PUBLISH news "Another update"
PUBSUB NUMSUB news

UNSUBSCRIBE และตัวนับ

UNSUBSCRIBE [channel ...] ตอบหนึ่งครั้งต่อ channel พร้อมจำนวน ที่เหลือ ถ้าไม่ใส่อาร์กิวเมนต์จะทิ้งทั้งหมด ตอบทีละอัน reply ดิบที่ผมจับได้ (SUBSCRIBE a b, แล้ว UNSUBSCRIBE a, แล้ว UNSUBSCRIBE): ["unsubscribe","a",1] แล้ว ["unsubscribe","b",0] เมื่อตัวนับถึง 0 connection กลับสู่โหมดปกติ (reply (nil) ที่เห็นในบทเรียนเก่าจะเกิดเมื่อ unsubscribe ทั้งที่ไม่ได้ subscribe อะไรเลยเท่านั้น)

connection ของ subscriber: RESP2 เทียบ RESP3

ใน RESP2 connection ที่ subscribe อยู่ถูกล็อกไว้กับ SUBSCRIBE/UNSUBSCRIBE/PSUBSCRIBE/PUNSUBSCRIBE/SSUBSCRIBE/SUNSUBSCRIBE, PING, QUIT และ RESET reply ดิบของเซิร์ฟเวอร์ต่อ GET บน connection แบบนั้น:

-ERR Can't execute 'get': only (P|S)SUBSCRIBE / (P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context

หลัง HELLO 3 (RESP3) GET เดียวกันตอบตามปกติ (null) เพราะ push เดินทางเป็น push type แยกและไม่ชนกับ reply เอกสาร ระบุไว้เหมือนกัน ioredis 6 พูด RESP3 เป็นค่าเริ่มต้น ซึ่งเปลี่ยนสิ่งที่บทเรียนเก่าพูดเรื่อง “ต้องมีสอง connection”:

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

const url = process.env.REDIS_URL!;
const sub = new Redis(url);   // connection 1: subscribe only
const pub = new Redis(url);   // connection 2: everything else

const got: string[] = [];
sub.on('message', (channel, message) => got.push(`${channel}: ${message}`));

// subscribe() resolves with the number of channels this connection now listens to
console.log('subscribed to', await sub.subscribe('news', 'sports'), 'channels');   // 2

// PUBLISH returns how many subscribers received it
console.log('delivered to', await pub.publish('news', 'Redis 8 is out'));          // 1
console.log('delivered to', await pub.publish('nobody-listens', 'hello?'));        // 0 -> gone forever

// PUBSUB introspection runs on the normal connection
console.log(await pub.pubsub('CHANNELS'));                                          // [ 'news', 'sports' ] (any order)
console.log(await pub.pubsub('NUMSUB', 'news', 'nobody-listens'));                  // [ 'news', 1, 'nobody-listens', 0 ]

// ioredis 6 speaks RESP3 by default, so a subscribed connection may still run normal commands
console.log('RESP3 GET on the subscriber ->', await sub.get('x'));                  // null
// force RESP2 and the subscriber is locked to (P|S)SUBSCRIBE/UNSUBSCRIBE/PING/QUIT/RESET
const legacy = new Redis(url, { protocol: 2 });
await legacy.subscribe('legacy');
try { await legacy.get('x'); } catch (e) { console.log('RESP2 error:', (e as Error).message); }
legacy.disconnect();

await new Promise((r) => setTimeout(r, 50));   // let the push arrive
console.log(got);                              // [ 'news: Redis 8 is out' ]

// unsubscribe, then a publish reaches nobody
await sub.unsubscribe('news', 'sports');
console.log('delivered to', await pub.publish('news', 'after unsubscribe'));        // 0

await sub.quit();
await pub.quit();

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

subscribed to 2 channels
delivered to 1
delivered to 0
[ 'sports', 'news' ]
[ 'news', 1, 'nobody-listens', 0 ]
RESP3 GET on the subscriber -> null
RESP2 error: Connection in subscriber mode, only subscriber commands may be used
[ 'news: Redis 8 is out' ]
delivered to 0

แม้ RESP3 อนุญาต ก็ควร publish บน connection แยกต่างหาก: connection ของ subscriber เป็นตัวรับที่อยู่ยาวและมี backpressure ของตัวเอง (ด้านล่าง) และไลบรารีที่ใช้ protocol: 2 (ioredis รุ่นเก่า หรือบังคับใช้) จะโยน error ข้างบน ลำดับของ PUBSUB CHANNELS ไม่ได้ถูกกำหนด ข้อความ RESP2 error เป็นข้อความฝั่ง client ของ ioredis เอง; ข้อความดิบของเซิร์ฟเวอร์คือบรรทัด -ERR Can't execute 'get'...

Sharded Pub/Sub (7.0+)

SSUBSCRIBE, SPUBLISH และ SUNSUBSCRIBE คือตัวแปร sharded ส่งมาเป็น smessage ใน cluster shard channel ถูก hash ไปยัง slot เหมือน key และข้อความอยู่ภายใน shard ของ slot นั้นแทนที่จะ broadcast ไปทุก node (เอกสาร หัวข้อ “Sharded Pub/Sub”; ไม่ได้พิสูจน์ที่นี่ เรารันเซิร์ฟเวอร์ standalone) มันใช้บนเซิร์ฟเวอร์ standalone ได้ด้วย: ผมรัน ssubscribe('orders') และ spublish('orders','o-1') ใน ioredis ได้ส่งถึง 1 และ event smessage orders o-1; SPUBLISH ไปยัง channel ที่ไม่มีใครฟังคืน 0 และ PUBSUB SHARDCHANNELS / SHARDNUMSUB แสดง orders มี subscriber 1 ราย ใช้เมื่อรัน Cluster และอยากกระจายภาระ Pub/Sub ไปตาม shard

subscriber ที่ช้าถูกตัดออก

Pub/Sub เก็บ push ไว้ใน output buffer ต่อ client ของเซิร์ฟเวอร์ ค่าเริ่มต้นของคลาส pubsub คือ client-output-buffer-limit pubsub 32mb 8mb 60: hard limit 32 MB หรือ 8 MB ต่อเนื่อง 60 วินาที เกินนั้นเซิร์ฟเวอร์ปิด connection ผมสร้าง subscriber ที่ไม่เคยอ่านและ publish ข้อความขนาด 1 MiB:

// @run
import net from 'node:net';
import { Redis } from 'ioredis';

const url = new URL(process.env.REDIS_URL!);
const slow = net.connect(Number(url.port), url.hostname);   // raw TCP subscriber
slow.on('error', () => {});
slow.write('SUBSCRIBE flood\r\n');
await new Promise((r) => setTimeout(r, 200));
slow.pause();                                               // ...that never reads its socket again

const pub = new Redis(process.env.REDIS_URL!);
const chunk = 'x'.repeat(1024 * 1024);                      // 1 MiB messages
let firstZero = -1;
for (let i = 0; i < 100; i++) {
  if ((await pub.publish('flood', chunk)) === 0 && firstZero < 0) firstZero = i;   // 0 = nobody left to deliver to
}
console.log('subscriber dropped after', firstZero, 'MiB published');
console.log((await pub.info('stats')).split('\r\n').filter((l) => l.includes('output_buffer')));
console.log((await pub.config('GET', 'client-output-buffer-limit'))[1]);           // ... pubsub 33554432 8388608 60

pub.disconnect();
slow.destroy();

Output จริง:

subscriber dropped after 38 MiB published
[ 'client_output_buffer_limit_disconnections:2' ]
normal 0 0 0 slave 268435456 67108864 60 pubsub 33554432 8388608 60

หลังผ่านราว 32 MB ของ buffer บวก socket buffer ของ kernel PUBLISH เริ่มคืน 0 และตัวนับใน INFO stats เพิ่มขึ้น (ตัวนับสะสมตั้งแต่เซิร์ฟเวอร์เริ่ม ตัวเลขของคุณจึงขึ้นกับการรันก่อนหน้า) consumer ที่ตามไม่ทันจะเสีย ทุก ข้อความในอนาคตจนกว่าจะเชื่อมต่อใหม่ ซึ่งเป็นอีกเหตุผลที่ Pub/Sub เหมาะกับสัญญาณสด ไม่ใช่ข้อมูลที่ต้องประมวลผล

🟥 Pub/Sub ส่งแล้วลืม: subscriber ที่ออฟไลน์หรือช้าจะพลาดข้อความ

ไม่มี buffer สำหรับ subscriber ที่ไม่อยู่ ไม่มี acknowledgement และไม่มีประวัติ: PUBLISH คืน 0 ข้างบนแล้วข้อความก็สิ้นสภาพ subscriber ที่หายไปเพียงมิลลิวินาทีเดียวจะไม่เห็นมัน และตัวที่ตามไม่ทันจะถูกตัดการเชื่อมต่อ ถ้าข้อความต้องถูกประมวลผล ให้ใช้ Streams (เก็บถาวร, consumer group, acknowledgement)

Reading AI-generated code

AI ต่อ “order paid” เข้ากับ worker จัดส่งด้วย Pub/Sub:

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

async function onPaymentSucceeded(order: Order) {
await db.markPaid(order.id);
await redis.publish('order:paid', JSON.stringify(order));   // fulfilment worker listens
}

เฉลย

ถ้า worker จัดส่งล่ม กำลัง restart หรือหลุดการเชื่อมต่อชั่วครู่ publish จะ resolve สำเร็จพร้อม 0 และออเดอร์ไม่เคยถูกจัดส่ง: ไม่มีอะไรถูกเก็บ ไม่มีอะไรลองใหม่ ข้างบน publish('nobody-listens', ...) คืน 0 และ PUBSUB NUMSUB แสดง subscriber 0 ราย ค่าที่คืนเป็นสัญญาณเดียวและโค้ดนี้เพิกเฉยมัน สำหรับงานที่ห้ามหาย ให้เขียนลง Stream (XADD order:paid * ...) และให้ consumer group อ่าน หรืออย่างน้อยถือว่า 0 เป็น incident ใช้ Pub/Sub กับสิ่งที่มันถนัด: สัญญาณสดที่ทิ้งได้ เช่น hint ให้ล้างแคช หรือตัวบอก “กำลังพิมพ์…”

Mental Model

Pub/Sub คือ การออกอากาศวิทยุสด: เฉพาะคนที่จูนอยู่ตอนนี้เท่านั้นที่ได้ยิน สถานีไม่เก็บการบันทึก และผู้ฟังที่ตามไม่ทันจะถูกวางสาย ใช้กับสัญญาณที่คุณค่าหมดทันที ใช้ Stream เมื่อต้องมีคนตามทัน

Trade-offs

ตัวเลือก	ข้อดี	ข้อเสีย
Pub/Sub สำหรับ fan-out	latency ต่ำมาก ไม่เก็บอะไร publish ครั้งเดียวถึงทุก subscriber ปัจจุบัน	at-most-once: subscriber ที่ไม่อยู่เสียข้อความ ไม่มี replay ไม่มี ack
หลาย channel บน connection subscriber เดียว	connection น้อยลง	ผู้อ่านช้าตัวเดียวถูกตัดที่ขีดจำกัด output-buffer และเสียทุกอย่างหลังจากนั้น
Sharded Pub/Sub ใน Cluster	traffic อยู่ภายใน shard ของ slot	ไม่มี pattern subscription สำหรับ shard channel (ไม่มี `SPSUBSCRIBE`); ต้อง 7.0+
connection publisher แยก	ตัวรับเรียบง่ายและไม่ขึ้นกับเวอร์ชัน protocol	มีอีกหนึ่ง connection ต้องจัดการ

Common Misconceptions

“connection ที่ subscribe อยู่รันคำสั่งไม่ได้เลย” จริงเฉพาะ RESP2; ioredis 6 (RESP3 เป็นค่าเริ่มต้น) รัน GET บน subscriber และได้ null
“PUBLISH คืน 1 แปลว่าข้อความถูกประมวลผลแล้ว” มันนับ subscriber ที่เซิร์ฟเวอร์ push ให้ ไม่มากไปกว่านั้น
“เปิด appendonly แล้ว Pub/Sub คงทน” ข้อความไม่เคยถูกเขียนลง RDB/AOF
“channel เป็นของ database” มันเป็น global; SELECT ไม่ได้จำกัดขอบเขต

Common mistakes

ใช้ Pub/Sub กับงานที่ห้ามหาย — ดูแบบฝึกหัดข้างบน ใช้ Stream
เพิกเฉยค่าที่ PUBLISH คืน — 0 คือหลักฐานเดียวของข้อความที่หาย
ทำงานช้าใน handler ของ message — subscriber ที่ตามไม่ทันถูกตัดที่ 32 MB (หรือ 8 MB นาน 60 วินาที) ส่งข้อความไปที่ queue แล้วคืนให้เร็ว

Real-world example

เอกสารของ Redis เอง ระบุว่า Pub/Sub “exhibits at-most-once message delivery semantics” และชี้ผู้ที่ต้องการการรับประกันที่แข็งกว่าไปที่ Streams (Redis Pub/Sub)