Messaging

บริบท

เมื่อลูกค้าสั่งซื้อ มีหลายอย่างที่ต้องเกิดขึ้น: การเรียกเก็บเงินต้องตัดบัตร คลังสินค้าต้องเตรียมจัดส่ง และอีเมลยืนยันต้องถูกส่งออกไป ไม่มีอันใดเลยที่ต้องเสร็จก่อนที่ผู้ซื้อจะเห็นคำว่า “สั่งซื้อสำเร็จ” ถ้าบริการ Order เรียกแต่ละบริการปลายน้ำแบบ synchronous ผู้ให้บริการอีเมลที่ช้าอาจทำให้ขั้นตอนชำระเงินค้าง และบริการใดบริการหนึ่งล่มก็จะทำให้การขายล้มเหลวทันที

ปัญหา

คุณอยากให้ service หนึ่งจุดชนวนงานใน service อื่นได้ โดยไม่ต้องรอผลและไม่ต้องผูกกับว่าตอนนั้นปลายทางพร้อมใช้งานหรือเปล่า งานบางอย่างควรกระจายออกไปหลายทาง เช่น event การสั่งซื้อครั้งเดียวควรขับเคลื่อนทั้งการเก็บเงิน การจัดส่ง และการแจ้งเตือนแยกกันเป็นอิสระ แถมคุณยังอยากเพิ่มปฏิกิริยาที่สี่เข้ามาทีหลังได้โดยไม่ต้องแตะ Order service เลย การเรียกแบบ synchronous ตรง ๆ ให้สิ่งเหล่านี้ไม่ได้เลย เพราะบล็อกการทำงาน ผูกสอง service ไว้ด้วยกันในเชิงเวลา และรู้จักผู้รับได้แค่รายเดียว คำถามคือจะให้ service ทำงานร่วมกันโดยลดการผูกทั้งเชิงเวลาและเชิงตัวตนได้อย่างไร

วิธีแก้

Messaging คือรูปแบบ asynchronous แทนที่จะเรียก service ตรง ๆ ฝั่ง producer จะส่งข้อความให้ message broker ที่เป็น infrastructure ที่เก็บข้อความไว้อย่างคงทนแล้วส่งต่อให้ consumer producer ไม่ต้องรอให้งานเสร็จ และไม่จำเป็นต้องรู้ด้วยซ้ำว่าใครเป็น consumer broker จึงตัดการผูกเชิงเวลาระหว่างสองฝั่งออกไป consumer ที่ล่มอยู่ก็แค่กลับมาทยอยประมวลผลงานที่ค้างไว้เมื่อฟื้น

broker มีรูปแบบการส่งสองแบบ:

• Queues (point-to-point) — แต่ละข้อความถูกส่งไปยัง consumer เพียงรายเดียว สิ่งนี้นำพา command: “ตัดเงินคำสั่งซื้อนี้” instance หลายตัวของ consumer แข่งกันรับข้อความ ซึ่งทำให้เกิดการกระจายโหลด
• Topics (publish/subscribe) — แต่ละข้อความถูกส่งไปยังผู้สมัครรับทุกราย สิ่งนี้นำพา event: “มีคำสั่งซื้อถูกสร้างขึ้น” การเรียกเก็บเงิน การจัดส่ง และการแจ้งเตือนต่างก็สมัครรับอย่างอิสระ และสามารถเพิ่มผู้สมัครรับรายใหม่ได้โดยไม่ต้องเปลี่ยน producer

flowchart LR
  P[Order Service<br/>producer] -->|OrderPlaced| Broker[(Message Broker<br/>topic)]
  Broker --> B[Billing Service]
  Broker --> S[Shipping Service]
  Broker --> N[Notification Service]
  B -.->|buffered if down| Broker

Messaging — producer เผยแพร่ event หนึ่งไปยัง broker แล้ว broker กระจายต่อไปยัง consumer ที่เป็นอิสระจากกัน

ตัวอย่าง

ตัวอย่างนี้คือ wrapper เล็ก ๆ ครอบ broker ที่เผยแพร่ event OrderPlaced แล้ว consume ที่ปลายอีกฝั่ง การเรียก publish จะคืนค่าทันทีที่ broker รับข้อความไว้ ส่วน consumer จะ acknowledge ก็ต่อเมื่อทำงานเสร็จแล้วเท่านั้น ข้อความที่ยังประมวลผลไม่สำเร็จจึงถูกส่งซ้ำ แต่ละตัวอย่างรันได้ด้วยตัวเองและเขียนตามสไตล์ของภาษานั้น ๆ

TypeScript

type OrderPlaced = { orderId: string; customerId: string; total: number };

// Publish: hand the event to the broker and return immediately.
async function publishOrderPlaced(broker: Broker, event: OrderPlaced): Promise<void> {
  await broker.publish('orders.placed', JSON.stringify(event));
}

// Consume: process each message, then acknowledge it.
function consumeOrders(broker: Broker): void {
  broker.subscribe('orders.placed', async (raw, ack) => {
    const event = JSON.parse(raw) as OrderPlaced;
    await chargeCustomer(event.customerId, event.total);
    await ack(); // ack only after the work succeeded
  });
}

Python

import json
from dataclasses import dataclass, asdict

@dataclass
class OrderPlaced:
    order_id: str
    customer_id: str
    total: float

# Publish: hand the event to the broker and return immediately.
async def publish_order_placed(broker, event: OrderPlaced) -> None:
    await broker.publish("orders.placed", json.dumps(asdict(event)))

# Consume: process each message, then acknowledge it.
def consume_orders(broker) -> None:
    async def handler(raw: str, ack) -> None:
        data = json.loads(raw)
        await charge_customer(data["customer_id"], data["total"])
        await ack()  # ack only after the work succeeded

    broker.subscribe("orders.placed", handler)

Go

type OrderPlaced struct {
    OrderID    string  `json:"orderId"`
    CustomerID string  `json:"customerId"`
    Total      float64 `json:"total"`
}

// Publish: hand the event to the broker and return immediately.
func PublishOrderPlaced(broker Broker, e OrderPlaced) error {
    payload, err := json.Marshal(e)
    if err != nil {
        return err
    }
    return broker.Publish("orders.placed", payload)
}

// Consume: process each message, then acknowledge it.
func ConsumeOrders(broker Broker) {
    broker.Subscribe("orders.placed", func(raw []byte, ack func() error) {
        var e OrderPlaced
        _ = json.Unmarshal(raw, &e)
        if err := ChargeCustomer(e.CustomerID, e.Total); err != nil {
            return // not acked -> redelivered later
        }
        _ = ack()
    })
}

Rust

use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize)]
struct OrderPlaced {
    order_id: String,
    customer_id: String,
    total: f64,
}

// Publish: hand the event to the broker and return immediately.
async fn publish_order_placed(broker: &Broker, event: &OrderPlaced) -> anyhow::Result<()> {
    let payload = serde_json::to_vec(event)?;
    broker.publish("orders.placed", payload).await?;
    Ok(())
}

// Consume: process each message, then acknowledge it.
async fn consume_orders(broker: &Broker) -> anyhow::Result<()> {
    broker.subscribe("orders.placed", |raw, ack| async move {
        let event: OrderPlaced = serde_json::from_slice(&raw)?;
        charge_customer(&event.customer_id, event.total).await?;
        ack().await?; // ack only after the work succeeded
        Ok(())
    }).await
}

ผลลัพธ์ที่ตามมา

สิ่งที่คุณได้:

• loose coupling producer ไม่บล็อก ไม่รู้จัก consumer และไม่สนใจว่าปลายทางกำลังทำงานอยู่หรือเปล่า consumer ตัวใหม่สมัครเข้ามารับได้โดยไม่ต้องแก้ producer เลย ระบบจึงโตด้วยการเพิ่มของใหม่ ไม่ใช่การไปแก้ของเดิม
• การบัฟเฟอร์และความทนทาน (buffering and resilience) broker ดูดซับช่วงที่ทราฟฟิกพุ่งสูงและเก็บข้อความไว้ขณะที่ consumer ล่มหรือกำลัง deploy ความช้าใน consumer หนึ่งไม่มีวันแพร่กลับไปยัง producer
• การกระจายตามธรรมชาติ (natural fan-out) event ที่เผยแพร่หนึ่งครั้งขับเคลื่อนปฏิกิริยาอิสระได้หลายอย่าง นั่นคือสิ่งที่การโต้ตอบแบบ one-to-many ต้องการพอดี

สิ่งที่คุณต้องจ่าย:

• broker กลายเป็น infrastructure ที่สำคัญที่สุด ทุกอย่างขึ้นกับว่า broker ยัง available คงทน และถูกดูแลอย่างถูกต้องหรือไม่ ต้องทำเป็น cluster ต้องเฝ้าติดตาม และกลายเป็นจุดใหม่ที่พังได้
• Eventual consistency เพราะงานเกิดขึ้นในภายหลัง จึงมีช่วงที่คำสั่งซื้อมีอยู่แล้วแต่อีเมลยังไม่ถูกส่ง ผู้เรียกไม่สามารถอ่านผลลัพธ์กลับมาได้ทันทีในแบบที่การเรียก synchronous คืนค่ามาให้
• ความซับซ้อนในการทำงาน (operational complexity) คุณต้องจัดการกับการส่งซ้ำ (broker ส่วนใหญ่เป็นแบบ at-least-once ดังนั้น consumer ต้อง idempotent) ข้อความที่มาผิดลำดับ และ poison message ที่ต้องใช้ dead-letter queue การ debug flow ที่กระจายอยู่ในหลาย publisher และ subscriber ยากกว่าการไล่ตาม call stack เดียว

เนื้อหาที่เกี่ยวข้อง

• Remote Procedure Invocation — ทางเลือกแบบ synchronous เมื่อผู้เรียกต้องการคำตอบทันทีจริง ๆ
• Saga — ใช้ messaging เพื่อประสานทรานแซกชันที่ครอบหลายบริการ
• Event Sourcing — วิธีที่เชื่อถือได้ในการสร้าง event ที่คุณจะเผยแพร่

ข้อดี	ข้อแลกเปลี่ยน
producer และ consumer decoupled — ไม่ต้องรู้จักกัน	eventual consistency — ข้อมูลไม่ sync ทันที
ทนต่อ spike ของ traffic — queue รับ message สะสม	message ordering ซับซ้อน โดยเฉพาะ parallel consumer
consumer ล้มได้โดยไม่กระทบ producer	debugging ยากกว่า synchronous call
scale consumer อิสระตาม backlog	message schema เปลี่ยนต้องระวัง backward compatibility

ข้อผิดพลาดที่พบบ่อย

Chatty Messaging — ส่ง message เล็ก ๆ จำนวนมากแทนที่จะ batch

อาการ:

• message broker ได้รับ event หลายพันตัวต่อวินาทีสำหรับ operation เดียว
• consumer ประมวลผลไม่ทัน backlog สะสม
• network overhead สูงเกินจำเป็น

Shared Schema Without Versioning — เปลี่ยน message schema โดยไม่ manage version

อาการ:

• consumer เก่า deserialize message จาก producer ใหม่ไม่ได้
• deploy producer ใหม่ทำให้ consumer crash พร้อมกัน

💡 ตัวอย่างจากของจริง

Uber:

• ใช้ Kafka สำหรับ real-time event streaming ระหว่าง service
• driver location update ส่งเป็น event — ไม่ใช่ synchronous API call
• ทำให้ scale แต่ละ consumer (pricing, ETA, map) แยกกันได้

LinkedIn (ผู้สร้าง Kafka):

• สร้าง Kafka เพื่อแก้ปัญหา messaging ภายใน
• ปัจจุบันประมวลผล trillions of messages ต่อวัน

🧩 consumer ที่เป็น idempotent ไม่ใช่ทางเลือก

broker ส่วนใหญ่รับประกันการส่งแบบ at-least-once ซึ่งหมายความว่า consumer ของคุณจะได้เห็นข้อความเดิมซ้ำสองครั้งในที่สุด ถ้าการประมวลผล “ตัดเงินคำสั่งซื้อ 17” ไม่ idempotent การส่งซ้ำจะตัดเงินลูกค้าสองครั้ง ทำให้ consumer ปลอดภัยที่จะทำซ้ำได้ — ลบข้อมูลซ้ำโดยอ้างอิงจาก message id หรือออกแบบการทำงานให้การใช้สองครั้งมีผลเหมือนกับการใช้ครั้งเดียว

messaging ลดการผูกมัดของสองบริการในเชิงเวลาได้อย่างไร?

บีบอัดข้อความให้ส่งได้เร็วขึ้น producer ส่งข้อความให้ broker แล้วไปทำงานต่อ consumer จึงมาประมวลผลตอนที่พร้อมได้ แม้จะล่มอยู่ตอนที่ข้อความถูกส่งก็ตาม บังคับให้ทั้งสอง service ใช้นาฬิกาเดียวกัน เรียก consumer แบบ synchronous แต่ตั้ง timeout ยาว ๆ

ความแตกต่างระหว่าง queue กับ topic คืออะไร?

queue เร็วกว่า; topic ช้ากว่า queue ส่งแต่ละข้อความไปยัง consumer เพียงรายเดียว (command); topic ส่งแต่ละข้อความไปยังผู้สมัครรับทุกราย (event) queue คงทน; topic อยู่ในหน่วยความจำเสมอ ไม่ต่างกัน เป็นชื่อสองชื่อของสิ่งเดียวกัน

เหตุใด consumer ของ messaging มักต้องเป็น idempotent?

เพราะ broker เข้ารหัสข้อความ เพราะการส่งแบบ at-least-once หมายความว่าข้อความเดิมอาจมาถึงมากกว่าหนึ่งครั้ง และการประมวลผลซ้ำต้องไม่ก่อให้เกิดอันตราย เพราะ topic สลับลำดับข้อความตามตัวอักษร เพราะ producer retry เฉพาะวันอังคารเท่านั้น

ข้อใดเป็นต้นทุนที่แท้จริงของการนำ messaging มาใช้?

ทำให้ทุกการเรียกกลายเป็น synchronous broker กลายเป็นโครงสร้างพื้นฐานสำคัญที่ต้องพร้อมใช้งานสูงและทำงานอย่างระมัดระวัง กันไม่ให้เพิ่ม consumer ใหม่ได้ กำจัด eventual consistency ออกไป

Submit

Mark as complete