Publish-Subscribe

ปัญหา

มีคนสั่ง order แล้วหลายส่วนของธุรกิจต้องรู้เรื่องนี้: Billing ต้องออกใบแจ้งหนี้ Shipping ต้องส่งของ Analytics ต้องนับ ส่วน service ที่รับ order มีสองวิธีที่เห็นชัด ๆ ในการบอกพวกนี้ และทั้งสองวิธีก็ผูกมันไว้กับผู้ฟัง

• เรียกไปหาทีละตัว แบบนี้ Orders ต้องรู้ address ของทุก service ที่สนใจ ต้องรอทุกตัว ต้องตัดสินว่าจะทำยัง⁠ไงเมื่อตัวใดตัวหนึ่งล่ม และต้องแก้แล้ว redeploy ทุกครั้งที่มีผู้ฟังตัวใหม่โผล่มา
• ใส่ข่าวลง queue queue เป็น channel แบบ point-to-point: message แต่ละตัวถูกหยิบไปโดยผู้รับตัวเดียวเท่านั้น ถ้า Billing กับ Shipping อ่านจาก queue เดียวกัน สองตัวนี้ก็จะแย่งกัน แล้วแต่ละ order ก็ไปถึงแค่ตัวเดียว (กรอบเล็กใน step 1) ทำ queue หนึ่งตัวต่อผู้รับหนึ่งตัวก็แก้ได้ แต่ตอนนี้ผู้ส่งต้องรู้จักทุก queue และต้องส่ง message เดียวกันหลายรอบ แถมยังอาจล้มกลางทางระหว่างไล่ส่งได้ด้วย

สิ่งที่ผู้ส่งอยากได้คือพูดครั้งเดียว โดยไม่ได้พูดกับใครเป็นพิเศษ แล้วปล่อยให้ infrastructure จัดการให้ทุกคนที่สนใจได้สำเนาของตัวเอง

ทำงานยัง⁠ไง

มีสองบทบาท และมีของสองอย่างบน broker ที่อยู่ระหว่างทั้งคู่

• publisher ส่ง message แต่ละตัวไปที่ topic หรือ channel ที่มีชื่อบน broker มันไม่ได้ส่งถึงใครเลย พอ broker รับ message ไปแล้ว publisher ก็หมดหน้าที่
• subscription คือสิทธิ์ถาวรของ subscriber ตัวหนึ่งบน topic หนึ่ง ทุก message ที่ publish เข้า topic ตัว subscription แต่ละตัวจะได้สำเนาของตัวเอง (หรือได้ pointer ของตัวเองไปที่ message ตัวเดียวที่เก็บไว้) และมี ตำแหน่ง ของตัวเอง: message ไหนที่ subscriber ของมันได้รับไปแล้วและ acknowledge แล้ว
• subscriber รับจาก subscription ของตัวเองเท่านั้น ไม่รับจากที่อื่น มัน process message แล้ว acknowledge ส่วน message ที่ไม่ได้ acknowledge ก็จะถูกส่งมาให้มันอีกรอบ

เพราะทุก subscription มีตำแหน่งของตัวเอง subscriber เลยไม่กระทบกัน ตัวหนึ่งจะช้า, crash, ถูก redeploy หรือหายไปหนึ่งชั่วโมงก็ได้ ขณะที่ตัวอื่นทำงานต่อไป (step 2) และจะเพิ่มหรือเอาตัวหนึ่งออกก็ได้โดยไม่มีใครรู้สึก (step 4)

Enterprise Integration Patterns ตั้งชื่อ channel สองแบบไว้ บน Point-to-Point Channel นั้น message แต่ละตัวมีผู้รับแค่ตัวเดียวที่ consume มัน ส่วน Publish-Subscribe Channel มี input เดียว และมี output หนึ่งช่องต่อ subscriber หนึ่งตัว แล้วทุก output ก็ได้สำเนาไป งาน survey ของ Eugster, Felber, Guerraoui และ Kermarrec อธิบายสิ่งที่แบบที่สองให้ไว้เป็นการ decouple สามแบบ: ด้าน space (แต่ละฝ่ายไม่รู้จักกัน) ด้าน time (ไม่ต้องรันอยู่พร้อมกัน) และด้าน synchronisation (publisher ไม่ถูก block ระหว่างที่ subscriber ทำงาน)

Durable หรือไม่ subscription แบบ durable อยู่ใน broker และไม่ขึ้นกับ connection ไหน แล้ว message ที่ publish มาระหว่างที่ subscriber ไม่อยู่ก็รออยู่ในนั้น แล้ว subscriber ก็ทำต่อจากตำแหน่งของมันเมื่อกลับมา ส่วน subscription แบบ non-durable มีอยู่แค่ตอนที่ subscriber ต่ออยู่ ช่วงที่หลุดไปก็ไม่มีอะไรเก็บไว้ให้มัน แบบแรกเหมาะกับงานที่ห้ามหาย แบบที่สองเหมาะกับมุมมองแบบ live ที่ message ที่มาช้าไม่มีค่าอะไรแล้ว: dashboard, ป้ายราคาที่วิ่งอยู่, สัญญาณให้ทิ้ง entry ใน cache

Filter subscription บอกได้ว่าอยากได้ message แบบไหน broker ใช้ filter ก่อนส่ง ทำให้ subscriber ไม่เคยได้รับส่วนที่เหลือ และไม่ต้องมี capacity ไว้รองรับมัน (step 3)

สร้างได้สองแบบ broker implement model นี้ได้สองแบบ และความต่างนี้เป็นตัวตัดสินว่า replay อะไรได้บ้าง

• queue หนึ่งตัวต่อ subscription broker ใส่สำเนาของแต่ละ message (หรือ reference ของมัน) ลงในทุก subscription ที่อยากได้ แล้วเอาออกจาก subscription นั้นเมื่อถูก acknowledge ตัว subscription จะมีแค่สิ่งที่มาถึงหลังจากมันถูกสร้าง ส่วน Azure Service Bus, RabbitMQ exchange กับ queue และ Amazon SNS ที่ป้อนเข้า Amazon SQS queue ก็ทำงานแบบนี้
• log ที่ใช้ร่วมกัน กับ cursor หนึ่งตัวต่อ subscriber topic เป็น log แบบ append-only ที่เก็บ message ไว้ตลอดช่วง retention ไม่ว่าจะมีใครอ่านแล้วหรือยัง ส่วน subscription ก็แทบเป็นแค่ offset ที่ชี้เข้าไปใน log การอ่านไม่ได้ลบอะไร เราเลยถอย subscriber กลับได้ และ subscriber ตัวใหม่ก็เริ่มจาก message ที่เก่าที่สุดที่ยังเก็บอยู่ได้ Apache Kafka ทำงานแบบนี้ RabbitMQ stream ก็เหมือนกัน

Google Cloud Pub/Sub อยู่ตรงกลางระหว่างสองแบบ: มันติดตาม acknowledgement ทีละ message สำหรับแต่ละ subscription เหมือน queue และเก็บ message ไว้บน topic แล้วย้อน subscription กลับไปในเวลาได้ เหมือน log

ใช้ตอนไหนดี

• consumer อิสระหลายตัวต้องการ message เดียวกัน และรายชื่อก็น่าจะโตขึ้นเรื่อย ๆ Event-driven architecture สร้างอยู่บนพื้นฐานตัวนี้
• ผู้ส่งไม่ควรต้องรู้จักผู้รับ ไม่ควรต้องรอ และไม่ควรล้มตามผู้รับ
• message บอกข้อเท็จจริง (OrderPlaced) แล้วผู้รับแต่ละตัวตัดสินเองว่าจะทำอะไรต่อ เหมือนใน choreographed saga
• stream เดียวกันป้อนให้ consumer หลายประเภท: service อื่น, data warehouse, audit trail, การ invalidate cache, หน้าจอ live
• consumer แต่ละตัวควรรับ burst ได้ตามจังหวะของตัวเอง durable subscription ทุกตัวเป็น buffer ที่อยู่หน้า subscriber ของมัน: ดู queue-based load leveling

อย่าใช้เมื่อ:

• ผู้ส่งต้องการคำตอบ publisher รู้แค่ว่า broker รับ message ไปแล้ว ไม่มีทางรู้ว่า subscriber ทำอะไรกับมัน ถ้าผู้เรียกต้องการผลลัพธ์ (ราคา ผลการ validate ใช่หรือไม่ใช่) ให้เรียก service ที่มีคำตอบนั้นไปเลย การเอา reply ไปต่อเติมบน topic ก็คือการสร้าง call ขึ้นมาใหม่ที่มีชิ้นส่วนขยับได้มากขึ้น และไม่มี timeout
• ต้องมี handler ตัวเดียวที่ลงมือ command (ตัดเงินบัตรนี้ ส่ง email นี้) ต้องถูกทำครั้งเดียวโดยเจ้าของคนเดียว ถ้า publish เข้า topic มันจะถูกทำโดยทุก subscription และถ้าไม่มี subscription เลยก็จะไม่มีใครทำ เพราะฉะนั้นให้ส่ง command ไปที่ queue แทน
• ต้องมีลำดับเดียวทั้งระบบ subscriber เห็น message ตามลำดับแค่ภายใน partition, key, session หรือ message group เดียวกัน ข้าม key กันหรือข้าม subscriber กันจะไม่มีลำดับ ถ้าจะให้มีลำดับเดียวครอบทุกอย่างก็ต้องมี partition เดียวกับ consumer ตัวเดียว แบบนั้นก็คือทิ้งการ scale ที่เราใช้ pattern นี้มาตั้งแต่แรก
• มี consumer ตัวเดียว และยังไม่เห็นตัวที่สองเลย queue ทำงานเดียวกันได้ โดย configure น้อยกว่าและอธิบายง่ายกว่า

ได้อะไร เสียอะไร

• publisher ไม่มีทางรู้ว่ามีใครจัดการ message แล้วหรือยัง publish สำเร็จแปลว่า broker ได้ message ไปแล้ว ไม่ได้บอกอะไรเกี่ยวกับ subscriber เลย: ไม่ได้บอกว่าได้รับแล้ว ไม่ได้บอกว่า process แล้ว และไม่ได้บอกด้วยซ้ำว่ามี subscriber อยู่หรือเปล่า บน broker ที่เก็บ queue ไว้หนึ่งตัวต่อ subscription ตัว topic ที่ไม่มี subscription จะรับ message แล้วทิ้งไปเฉย ๆ เลยต้องมีคนคอยเฝ้าแต่ละ subscription แทน (ดู เฝ้าทุก subscription ข้างล่าง)
• ตัวซ้ำ broker แทบทุกตัวส่งแบบ at least once แล้ว subscriber ที่ crash หลังทำงานเสร็จแต่ก่อน acknowledge ก็จะได้ message นั้นอีกรอบ
• ลำดับเป็นแค่บางส่วน อย่างดีที่สุดลำดับก็ถูกต้องแค่ต่อ key หรือต่อ partition และการส่งซ้ำก็อาจเอา message เก่าไปไว้หลัง message ที่ใหม่กว่า
• message กลายเป็นสัญญาสาธารณะ publisher ไม่ต้องพึ่ง API ของ subscriber อีกแล้ว แต่ subscriber ทุกตัวพึ่ง message ของมัน และมันก็มองไม่เห็นว่าใครอ่าน field ไหน เท่ากับว่า coupling ย้ายจาก call ไปอยู่ที่ schema
• consistency เป็นแบบ eventual subscriber ลงมือหลังเหตุการณ์ และในเวลาที่ต่างกัน Billing อาจออกใบแจ้งหนี้ order ที่ Shipping ยังไม่เห็นด้วยซ้ำ
• flow ตามยากขึ้น ไม่มี call stack ไหนบอกว่าใครตอบสนองต่อ OrderPlaced: คำตอบอยู่ใน configuration ของ broker ให้ใส่ correlation ID ไว้ในทุก message และใช้ distributed tracing
• ต้นทุนโตตาม fan-out ทุก message ถูกเก็บและถูกส่งหนึ่งครั้งต่อ subscription หนึ่งตัว และ subscription ที่ไม่มีใครอ่านแล้วก็จะเต็มขึ้นเรื่อย ๆ จนกว่าจะหมดอายุหรือถูกลบ broker ยังจำกัด fan-out ไว้ด้วย: 10,000 subscription ต่อ Pub/Sub topic, 2,000 ต่อ Service Bus topic, 12.5 ล้านต่อ SNS topic แบบ standard และ 100 ต่อแบบ FIFO

ข้อควรรู้ตอนลงมือทำ

• model เดียว หลายชื่อ สิ่งที่หน้านี้เรียกว่า subscription คือ consumer group ใน Kafka, queue ที่ bind กับ exchange ใน RabbitMQ, SQS queue ที่อยู่หลัง SNS topic และ session ใน MQTT

  Product	publish ไปที่	subscriber แต่ละตัวอ่านจาก	การ filter ใน broker
  Apache Kafka	topic ที่เป็น log แบ่ง partition	partition ของ topic ในฐานะ consumer group ของตัวเอง โดยมี offset ที่ commit ไว้ต่อ partition	ไม่มี: group ได้ทุก record ของ topic ที่มัน subscribe
  Google Cloud Pub/Sub	topic	subscription ของตัวเอง (pull, push หรือ export)	filter บน attribute ของ message กำหนดตายตัวตอนสร้าง subscription
  Amazon SNS กับ SQS	SNS topic	SQS queue ของตัวเองที่ subscribe topic ไว้	filter policy ต่อ subscription บน attribute ของ message หรือบน JSON body
  Azure Service Bus	topic	subscription ของตัวเอง ที่ทำตัวเหมือน queue	rule ต่อ subscription: SQL filter หรือ correlation filter บน property ของ message
  RabbitMQ	exchange	queue ของตัวเองที่ bind กับ exchange	ประเภท exchange กับ binding: ทุกอย่าง (fanout), routing key (direct, topic) หรือ header
  MQTT	ชื่อ topic	session ที่ broker เก็บไว้ให้ client นั้น	topic filter ที่มี wildcard
  Redis pub/sub	channel	connection ของมัน	channel pattern

• product แต่ละตัวสร้างมันยัง⁠ไง ตัวอย่าง เช็กเมื่อเดือนตุลาคม 2026:

  • Apache Kafka (4.3) Kafka ส่งแต่ละ record ไปให้ member หนึ่งตัวของทุก consumer group ที่ subscribe topic นั้น subscriber แต่ละตัวเลยเป็น consumer group ของตัวเอง ตำแหน่งของ group คือ offset ที่ commit ไว้ต่อ partition โดย Kafka เก็บ offset นี้ไว้ใน internal topic ส่วน record ไม่ถูกลบตอนถูกอ่าน: log ถูกเก็บไว้ตาม retention ของ topic คือ 7 วันถ้าไม่ได้ตั้งเป็นอย่างอื่น group ที่ไม่มี offset ที่ commit ไว้จะเริ่มตามที่ auto.offset.reset บอก ค่า default คือ latest แปลว่าเอาแค่ record ใหม่ ส่วน earliest อ่านทุกอย่างที่ยังเก็บอยู่ group ที่ว่างอยู่นานเกิน offsets.retention.minutes (default 7 วัน) จะเสีย offset ไป แล้วถูกนับเป็น group ใหม่ตอนต่อเข้ามาครั้งถัดไป
  • Google Cloud Pub/Sub topic หนึ่งมีได้หลาย subscription และ subscription แต่ละตัวเป็นของ topic เดียว subscription เห็นแค่สิ่งที่ publish หลังจากมันถูกสร้าง ได้รับแบบ at least once และไม่มีลำดับถ้าไม่ได้ใช้ ordering key ส่วน message ที่ยังไม่ acknowledge ถูกเก็บไว้ 7 วันโดย default (ตั้งได้ 10 นาทีถึง 31 วัน) และ subscription ที่ไม่มีความเคลื่อนไหวจาก subscriber เลย 31 วันจะถูกลบ เว้นแต่ expiration policy ของมันจะบอกเป็นอย่างอื่น ตัว topic เองก็เก็บ message ได้ นานสูงสุด 31 วัน ทำให้ subscription seek ย้อนไปที่ timestamp หนึ่งได้ รวมถึง timestamp ก่อนที่ subscription จะมีอยู่ด้วย
  • Amazon SNS กับ Amazon SQS SNS push แต่ละ message ไปที่ endpoint ที่ subscribe topic ไว้ และไม่เก็บอะไรไว้ให้ consumer มาดึงทีหลัง การ subscribe SQS queue หนึ่งตัวต่อ consumer หนึ่งตัวทำให้มันกลายเป็น durable subscription ที่ consumer มา poll เอา การจัดแบบนี้มักเรียกว่า fan-out ส่วน policy ของ queue ต้องยอมให้ topic ส่งเข้ามาได้ ถ้าไม่ได้เปิด raw message delivery ตัว queue จะได้ JSON envelope ที่ห่อ message ที่ publish มา ตัว topic แบบ standard ส่งแบบ at least once และพยายามรักษาลำดับแบบ best-effort ส่วน topic แบบ FIFO เพิ่มการรักษาลำดับภายใน message group และการตัดตัวซ้ำ
  • Azure Service Bus subscription ทำตัวเหมือน virtual queue ที่ได้สำเนาของทุก message ที่ส่งเข้า topic และผู้รับก็ใช้มันแบบเดียวกับที่ใช้ queue เป๊ะ ๆ แต่ละ subscription มี rule ที่มีชื่อ คือ filter ที่มี action ใส่หรือไม่ใส่ก็ได้ และ subscription ใหม่จะเริ่มด้วย default rule ที่รับทุกอย่าง filter เห็น system property กับ user property แต่ไม่เคยเห็น body ของ message
  • RabbitMQ publisher ส่งไปที่ exchange แล้ว exchange ก็ route ไปที่ queue ที่ bind กับมัน exchange แบบ fanout route ไปทุกตัว แบบ direct หรือ topic route ตาม routing key แบบ headers route ตามค่าใน header ส่วน subscriber แต่ละตัวประกาศ queue ของตัวเอง durable queue ที่มีชื่อตายตัวคือ durable subscription ส่วน exclusive queue ที่ server ตั้งชื่อให้จะถูกลบเมื่อ connection ปิด และเป็นแบบ non-durable ส่วน message ที่ไม่ตรงกับ binding ไหนเลยจะถูกทิ้ง (หรือส่งต่อไปที่ alternate exchange ถ้าตั้งไว้) หรือถูกส่งคืน publisher ถ้า publish แบบ mandatory ส่วน stream คือทางเลือกที่เป็นรูปร่าง log: การอ่านไม่ได้ลบ message และ consumer จะเกาะที่ message แรก message สุดท้าย message ถัดไป offset หรือ timestamp ก็ได้
  • MQTT 5.0 client subscribe ด้วย topic filter บนชื่อ topic ที่คั่นด้วย /: + ตรงกับหนึ่ง level พอดี ส่วน # ที่ต้องอยู่ท้ายสุดตรงกับกี่ level ก็ได้ ส่วนการที่ subscription จะ durable หรือไม่ขึ้นกับ session ถ้า Session Expiry Interval มากกว่าศูนย์ server จะเก็บ subscription ของ client กับ message QoS 1 และ QoS 2 ที่ค้างอยู่ไว้นานเท่านั้นหลัง client disconnect แต่ถ้า interval เป็นศูนย์ session ก็จบไปพร้อมกับ network connection ส่วน message แบบ retained คือ message ล่าสุดที่ publish เข้า topic พร้อม retain flag และ server จะส่งมันให้ subscriber ใหม่ทุกตัว คนที่เพิ่งมาเลยรู้ค่าปัจจุบัน แต่ไม่ได้ประวัติ
  • Redis pub/sub channel ที่ไม่มี storage เลย: ส่งแบบ at most once และ subscriber ที่หลุดหรือล้มก็เสีย message ไปเฉย ๆ subscriber จับคู่ชื่อ channel ด้วย pattern แบบ glob ได้ นี่คือแบบ non-durable ในรูปที่บริสุทธิ์ที่สุด และ Redis ก็แนะนำให้ไปใช้ Redis Streams เมื่อต้องการการรับประกันที่แข็งกว่านี้

• Durable ไม่ได้แปลว่าตลอดกาล broker ส่วนใหญ่จำกัดว่า message ของ subscriber ที่ไม่อยู่จะรอได้นานแค่ไหน: retention ของ Pub/Sub subscription (default 7 วัน) ของ SQS queue (default 4 วัน สูงสุด 14) หรือของ Kafka log หลังจากนั้น message ก็หายไป และไม่มีอะไรรายงาน error ให้เลือก retention จาก outage ที่นานที่สุดที่ subscriber ต้องรอดให้ได้ และตั้ง alarm ไว้ให้ดังก่อนถึงจุดนั้นนาน ๆ ให้ตัดสินแยกเป็นราย subscriber ด้วยว่าควร durable หรือเปล่า Jakarta Messaging ทำให้ทางเลือกนี้ชัดเจน: non-durable subscription มีอยู่แค่ตอนที่มี consumer ที่ active ส่วน durable subscription จะเก็บ message ต่อไปจนกว่าจะถูกลบด้วย unsubscribe

• การรับประกันการส่งเป็นของ subscription at-least-once คือเรื่องปกติ: acknowledge หลังทำงานเสร็จ และเตรียมรับ message เดิมอีกรอบหลัง crash, timeout หรือ acknowledgement ที่หายไป ให้ subscriber ทุกตัวเป็น idempotent consumer การรับประกันที่แข็งกว่านี้มีอยู่ แต่แคบ: Pub/Sub มี exactly-once delivery สำหรับ pull subscription ภายใน region เดียว ส่วน SNS FIFO topic ทิ้ง message ที่ deduplication ID ซ้ำภายในห้านาที และ Kafka transaction ทำให้ขั้น read-process-write เป็น atomic ตราบใดที่ output อยู่ใน Kafka ส่วนใน MQTT การรับประกันเป็นราย subscription: message มาถึงด้วย QoS ตัวที่ต่ำกว่าระหว่าง QoS ตอน publish กับ QoS ที่ให้ subscription ไว้

• ลำดับเป็นแบบต่อ key Kafka รักษาลำดับภายใน partition และส่ง record ที่ key เดียวกันไปที่ partition เดียวกัน Pub/Sub เรียงลำดับ message ที่ใช้ ordering key เดียวกัน ถ้า publish ใน region เดียวกันและเปิด ordering ไว้บน subscription ส่วน SNS FIFO topic เรียงลำดับภายใน message group ส่วน Service Bus เรียงภายใน session เลือก key ที่ต้องรักษาลำดับ (order ID, account) และไม่ต้องคาดหวังอะไรข้าม key การส่งซ้ำก็ซ้ำเป็นชุดด้วย: ใน Pub/Sub การส่ง message หนึ่งซ้ำจะส่ง message ที่ตามมาทีหลังใน key เดียวกันซ้ำด้วย

• Filter ตาม topic ตาม attribute หรือตามเนื้อหา

  • ตามลำดับชั้นของ topic publish เข้า topic ที่แคบ แล้วให้ subscriber เลือกตามชื่อ ส่วน topic filter ของ MQTT กับ topic exchange ของ RabbitMQ เพิ่ม wildcard ให้ (ใน binding key * แทนหนึ่งคำพอดี ส่วน # แทนศูนย์คำขึ้นไป) และ Redis ก็มี channel pattern ส่วน topic ที่แคบทำให้ต้อง filter น้อยลง แต่ก็มี topic ให้ดูแลมากขึ้น
  • ตาม attribute publisher ใส่ metadata ไว้ข้าง payload (ประเภท, region, tenant) แล้ว subscription ก็ filter ตามนั้น filter ของ Pub/Sub เห็นแค่ attribute ส่วน filter ของ Service Bus เห็นแค่ property และ attribute ก็เป็นขอบเขต default ของ SNS filter policy
  • ตามเนื้อหา broker มองเข้าไปใน payload ตัว SNS filter policy กำหนดขอบเขตให้ดู body ของ message ได้ ถ้า body เป็น JSON
  • filter เป็นส่วนหนึ่งของสัญญา: publisher ต้องใส่ attribute ในทุก message และ message ที่ถูก filter ทิ้งไปก็หายไปจาก subscription นั้นแล้ว ระวังรายละเอียดด้าน operation ด้วย อย่าง filter ของ Pub/Sub แก้ไม่ได้หลังสร้าง subscription แล้ว ส่วนการแก้ SNS filter policy อาจใช้เวลาถึง 15 นาทีกว่าจะมีผลทุกที่ แล้ว SQL filter ใน Service Bus ก็กิน throughput เลยควรใช้ correlation filter ถ้ามันพอ ใน Kafka ตัว group ได้ทุก record เลยต้องให้ consumer filter เอง หรือให้ stream-processing job เขียน topic ที่แคบกว่าออกมา

• Push หรือ pull ถ้าเป็น pull subscriber เป็นคนขอ message: Kafka consumer, การ poll SQS, Pub/Sub pull subscription ตัว subscriber จะกำหนดจังหวะเอง ตามไม่ทันได้โดยไม่โดนถล่ม และต้องมีแค่ connection ขาออก ถ้าเป็น push broker เป็นคนเริ่มส่ง: SNS, Pub/Sub push subscription ที่ POST ไปที่ HTTPS endpoint และนับ success status เป็น acknowledgement, RabbitMQ และ MQTT ผ่าน connection ที่ subscriber เปิดไว้ ส่วน push ไม่ต้องมี loop สำหรับ poll แต่ broker เป็นคนกำหนดจังหวะ subscriber เลยต้องจำกัดจำนวนที่ in flight และ HTTP endpoint ก็ต้องเข้าถึงได้และต้อง authenticate ผู้เรียกด้วย ส่วน RabbitMQ แนะนำให้ register consumer มากกว่าการ poll queue

• Retention, replay และสิ่งที่ subscriber ใหม่เห็น โดย default ตัว subscription ใหม่จะเริ่มที่ตอนนี้ แบบใน step 4 ถ้าจะให้ subscriber ใหม่ได้ประวัติ หรือจะ process ใหม่หลังเจอ bug ตัว broker ต้องเก็บ message ไว้:

  • Kafka: เริ่ม group ใหม่ด้วย auto.offset.reset=earliest หรือย้าย group ที่หยุดอยู่ด้วย kafka-consumer-groups.sh --reset-offsets ไปที่ offset แรกสุดหรือไปที่จุดเวลาหนึ่ง
  • Pub/Sub: เปิด message retention บน topic แล้ว seek subscription ไปที่ timestamp หรือ snapshot
  • SNS: FIFO topic archive message ไว้ได้นานถึง 365 วัน และ replay policy ของ subscription ก็ส่ง message พวกนั้นซ้ำได้จากจุดเริ่มที่กำหนด
  • RabbitMQ stream: ให้ consumer เกาะที่ first ที่ offset หรือที่ timestamp
  • การ replay ส่ง message ที่ subscriber อาจจัดการไปแล้ว นี่เป็นอีกเหตุผลที่ต้อง idempotent ส่วนถ้า broker ไม่เก็บอะไรไว้เลย คนที่เพิ่งมาก็ต้องหาแหล่งอื่นสำหรับอดีต: API ของ publisher, export หรือ snapshot แล้วตามด้วย stream

• message คือสัญญา ให้ทุกประเภท message มี schema (Avro, Protocol Buffers, JSON Schema) และมีกติกาการเปลี่ยน เพิ่ม field ที่เป็น optional ได้ แต่ห้ามเปลี่ยนชื่อ field หรือเอา field เดิมไปใช้กับอย่างอื่น breaking change คือประเภทใหม่หรือ topic ใหม่ ที่ publish คู่กับตัวเก่าจนกว่า subscriber ตัวสุดท้ายจะย้ายไป ลำดับเดียวกับ expand and contract ส่วน schema registry บังคับกติกานี้ตรงจุดที่ publish message ตัว Confluent Schema Registry ปฏิเสธ version ใหม่ที่ทำลาย compatibility ที่ตั้งไว้ ค่า default ของมันคือ BACKWARD แปลว่า consumer ที่ใช้ schema ใหม่อ่านข้อมูลที่เขียนด้วย schema ก่อนหน้าได้ เลยต้อง upgrade consumer ก่อน ส่วน Pub/Sub ผูก schema แบบ Avro หรือ Protocol Buffers เข้ากับ topic ได้ และปฏิเสธ message ที่ไม่ตรง schema ส่วน envelope อย่าง CloudEvents ทำให้ metadata เป็นมาตรฐาน (id, source, type) ทำให้ filter กับการเช็ก idempotence มีของที่หน้าตาเหมือนกันให้ใช้

• ความล้มเหลวอยู่แค่ใน subscription เดียว message ที่ Billing process ไม่ได้จะถูก retry ใน subscription ของ Billing และ Shipping ไม่มีวันรู้ ตั้ง retry กับ dead-letter queue ต่อ subscription: dead-letter topic บน Pub/Sub subscription, dead-letter subqueue ที่ Service Bus subscription ทุกตัวมี, redrive policy บน SQS queue ส่วน SNS มี dead-letter queue ของตัวเองบนแต่ละ subscription ไว้สำหรับ message ที่มันส่งไปที่ endpoint ไม่ได้ มัน retry ไปที่ SQS queue หรือ Lambda function 100,015 ครั้งใน 23 วัน และไปที่ endpoint แบบ email, SMS หรือ mobile push 50 ครั้งใน 6 ชั่วโมง ส่วน HTTP/S endpoint มี policy ที่ตั้งค่าได้ และนานสูงสุดหนึ่งชั่วโมง เมื่อ retry หมดแล้ว message จะถูกทิ้ง เว้นแต่จะตั้ง dead-letter queue นั้นไว้ ให้ redrive message ที่พักไว้กลับไปที่ subscription ที่มันล้ม ไม่ใช่ไปที่ topic ไม่งั้น subscriber ตัวอื่นทุกตัวจะได้มันเป็นรอบที่สอง

• การ scale subscriber ตัวเดียวเป็นอีกเรื่อง ถ้าอยาก process เร็วขึ้น ให้รันหลาย instance ของ subscriber บน subscription เดียวกัน: เพิ่ม member ใน Kafka consumer group (ต่อ partition หนึ่งจะมี member ที่ได้ทำงานจริงไม่เกินหนึ่งตัว) หลาย receiver บน Service Bus subscription ตัวเดียวหรือ SQS queue ตัวเดียว หรือ MQTT shared subscription ($share/<name>/<filter>) นั่นคือ pattern competing consumers และ instance ของมันใช้ตำแหน่งร่วมกันตำแหน่งเดียว ความผิดพลาดคลาสสิกสองแบบมาจากการสับสนสองเรื่องนี้: service ต่างกันสองตัวบน subscription เดียวกันจะต่างเห็น message แค่บางส่วน และ instance สองตัวของ service เดียวกันที่มี subscription คนละตัวจะทำงานทั้งหมดซ้ำสองรอบ

• subscriber ที่อยู่นอกองค์กร อย่าให้ partner เข้าถึง broker ของเรา ให้ส่งผ่าน HTTP ด้วย webhooks: subscription ของเราเองรับ message แล้วมีตัวส่งที่เซ็น, POST และ retry ให้ ส่วน push subscription ก็เป็นกลไกเดียวกันภายใน cloud เดียว

• Publish ให้เชื่อถือได้ การบันทึก order กับการ publish OrderPlaced เป็นการเขียนสองครั้ง และถ้า crash ระหว่างสองครั้งนี้ ก็จะเหลือ order ที่ไม่มีใครได้ยิน หรือ event ของ order ที่ถูก rollback ไปแล้ว ให้ใช้ transactional outbox หรือ change data capture ตัว relay อาจ publish message ซ้ำสองครั้ง เลยต้องให้ทุก message มี ID ที่ subscriber ใช้ตัดตัวซ้ำได้

• เฝ้าทุก subscription นี่คือราคาของการไม่รู้จัก subscriber สำหรับแต่ละ subscription ให้ติดตาม backlog (message ที่ยังไม่ acknowledge) และ อายุของ message ที่ยังไม่ acknowledge ตัวที่เก่าที่สุด และให้ alarm ไปหาทีมที่เป็นเจ้าของ subscriber ส่วน metric ใน Pub/Sub คือ subscription/num_unacked_messages_by_region กับ subscription/oldest_unacked_message_age_by_region ใน SQS คือ ApproximateNumberOfMessagesVisible กับ ApproximateAgeOfOldestMessage และใน Kafka คือ lag ต่อ partition ที่ kafka-consumer-groups.sh --describe พิมพ์ออกมา อายุเป็น alarm ที่ดีกว่า: backlog ใหญ่เป็นเรื่องปกติของ subscription ที่งานยุ่ง ส่วน backlog เล็กที่แก่ขึ้นเรื่อย ๆ แปลว่ามี message บางตัวติดอยู่ เทียบอายุกับ retention เพราะ message ที่เก่ากว่านั้นหายไปแล้ว คอยหา subscription ที่ไม่มีใคร consume และ topic ที่ไม่มีใคร subscribe ด้วย ถ้า publisher ต้องรู้ผลจริง ๆ ให้ subscriber publish event ของตัวเอง (OrderInvoiced) แล้วให้ publisher subscribe ตัวนั้น

• ใคร publish ได้ และใคร subscribe ได้ ให้สิทธิ์สองอย่างนี้แยกกันและให้แคบ ใครที่ publish ได้ก็ยัด message ที่ subscriber ทุกตัวจะลงมือทำตามได้ ใครที่สร้าง subscription ได้ก็ได้สำเนาของทุกอย่าง รวมถึง field ที่อ่อนไหวด้วย ใน SNS ตัว topic policy ให้สิทธิ์ sns:Publish และ sns:Subscribe ส่วนใน Pub/Sub ตัว Publisher role อยู่บน topic ส่วน Subscriber role อยู่บน subscription และการผูก subscription ใหม่ต้องมี permission บน topic ส่วนใน Kafka ตัว ACL ให้ producer เขียน topic ได้ และให้ consumer อ่าน topic กับ group ของตัวเองได้ แล้ว Service Bus ก็แยกสิทธิ์ Send กับ Listen ออกจากกัน อย่าใส่ secret หรือข้อมูลส่วนบุคคลที่ไม่จำเป็นไว้ใน message: ผู้ฟังของ topic โตขึ้นเรื่อย ๆ ตามเวลา และจะไม่มีใครมาถาม publisher ก่อน