สารบัญชุดบทความ

  1. Communication Styles และ Trade-offs
  2. แยก Background Job ก่อน Scale ข้าม Process
  3. ส่ง Job ข้าม Process ด้วย NATS
  4. ทำให้ Job ไม่หายด้วย NATS JetStream
  5. เมื่อ Job เติบโตเป็น Event Stream ด้วย Kafka
  6. เลือก Communication Style ให้เหมาะกับแต่ละ Service

ร้านรับ order แล้ว แต่ลูกค้ายังรอหน้าจอหมุน

ในตอนแรก เราแยกได้ว่า job ไหนต้องรอ response และ job ไหนทำแบบ asynchronous ได้ คราวนี้ลองนำแนวคิดนั้นมาใช้กับ food delivery app ของเรา

แม้เราจะแบ่งความรับผิดชอบเป็น Order และ Rider Matching แต่ช่วงเริ่มต้นยังไม่จำเป็นต้อง deploy แยก service เราให้ทั้งสองส่วนเป็น module ใน application เดียวกันก่อน เพื่อให้ทีมเล็กดูแลได้ง่าย

เมื่อร้านกดรับ order request handler จะหา rider ให้เสร็จก่อน แล้วค่อยส่ง response กลับไป ช่วงเที่ยงมีหลายร้านรับ order พร้อมกัน การหา rider จึงใช้เวลานานขึ้น ลูกค้าต้องรอ ทั้งที่เราสามารถแสดงสถานะ “ร้านรับ order แล้ว กำลังหา rider” ได้ก่อน

เราจึงแยกการหา rider เป็น background job และให้ request handler ตอบกลับหลังเพิ่ม job เข้า queue สำเร็จ เริ่มจาก solution ภายใน application ที่สร้างได้ด้วย standard library ก่อน

แยกส่วนที่รับ request ออกจากส่วนที่ทำ background job

เราเพิ่ม in-memory queue เพื่อเก็บ job ที่รอประมวลผลไว้ใน memory ของ process Request handler ส่ง order ID เข้า queue แทนการหา rider เอง จากนั้นให้ worker อ่าน job จาก queue ไปทำ

เราสามารถให้ worker หลายตัวแบ่งกันอ่านจาก queue เดียวกัน เรียกว่า worker pool แต่ละ job จะถูกอ่านโดย worker ตัวใดตัวหนึ่ง จำนวน worker เป็นตัวกำหนดว่าจะประมวลผลได้พร้อมกันกี่ job จึงช่วยจำกัด concurrency แม้มี request เข้ามาจำนวนมาก

สีใน diagram ใช้เหมือนกันตลอดชุดบทความ: Order / Order API = สีฟ้า; Rider Matching = สีเขียว; Queue / broker / event stream = สีเหลือง; Client / database / file / infrastructure = สีเทา สีบอก service หรือประเภทขององค์ประกอบ โดยชื่อในกล่องระบุหน้าที่ ส่วนกรอบแดงเส้นประพร้อมคำว่า “stopped” บอก worker ที่หยุดทำงาน

flowchart LR
    subgraph App["Application process เดียว"]
        Handler["Request handler"] --> Queue["In-memory queue"]
        Queue --> W1["Matching worker 1"]
        Queue --> W2["Matching worker 2"]
    end

    classDef order fill:#dbeafe,stroke:#2563eb,color:#0f172a
    class Handler order
    classDef broker fill:#fef3c7,stroke:#a16207,color:#0f172a
    class Queue broker
    classDef rider fill:#dcfce7,stroke:#15803d,color:#0f172a
    class W1,W2 rider
    style App fill:#f1f5f9,stroke:#64748b,color:#0f172a,stroke-width:2px

ถ้าเขียนด้วย Go เราใช้ buffered channel เป็น queue และ goroutine เป็น worker ได้ ตัวอย่างนี้จำลองการส่งสาม job ให้ worker สองตัว โดย print ข้อความแทนการหา rider จริง:

package main

import (
	"fmt"
	"sync"
)

func main() {
	jobs := make(chan string, 10)
	var workers sync.WaitGroup

	for i := 0; i < 2; i++ {
		workers.Add(1)
		go func() {
			defer workers.Done()
			for orderID := range jobs {
				fmt.Println("หา rider ให้", orderID)
			}
		}()
	}

	for _, orderID := range []string{"order-1", "order-2", "order-3"} {
		select {
		case jobs <- orderID:
			fmt.Println("enqueue job แล้ว:", orderID)
		default:
			fmt.Println("คิวเต็ม รับ job เพิ่มไม่ได้:", orderID)
		}
	}

	close(jobs)
	workers.Wait()
}

ตัวอย่างนี้กำหนดให้ queue เก็บ job ที่รอได้ 10 รายการ และมี worker ประมวลผลพร้อมกันได้สองตัว ใน application จริง เราควรสร้าง worker pool ตอนเริ่ม process และใช้ pool เดิมร่วมกันทุก request

เมื่อส่ง job ครบแล้ว close(jobs) จะปิด channel สำหรับการส่ง job ใหม่ แต่ worker ยังอ่าน job ที่เหลือใน channel ต่อได้ จากนั้น workers.Wait() รอให้ worker ทุกตัวทำ job เสร็จ ก่อนปล่อยให้โปรแกรมจบ

ถ้าส่ง job เข้ามาเร็วกว่าที่ worker ทำเสร็จล่ะ?

Buffer คือพื้นที่ใน queue สำหรับเก็บ job ที่ worker ยังไม่ได้รับไปทำ หากมี job เข้ามาจำนวนมากในช่วงสั้น ๆ buffer จะช่วยให้ job เหล่านั้นรอใน queue ได้ โดยผู้ส่งยังไม่ต้องรอ worker ว่าง แต่ buffer ไม่ได้ทำให้ worker ประมวลผลเร็วขึ้น

สมมติมี job ใหม่เข้ามา 100 รายการต่อวินาที แต่ worker ทั้งหมดรวมกันทำเสร็จเพียง 60 รายการต่อวินาที job ที่ยังไม่เสร็จจะเพิ่มขึ้นประมาณ 40 รายการทุกวินาที และเมื่อ worker รับ job เพิ่มไม่ทัน queue ก็จะเต็มในที่สุด

เราจึงต้องออกแบบ backpressure หรือวิธีควบคุมการรับ job เมื่อระบบประมวลผลไม่ทัน จะให้ผู้ส่งรอจน queue มีที่ว่าง หรือ reject job ใหม่แล้วแจ้งให้ผู้ส่ง retry ภายหลัง?

ถ้าเลือกให้ผู้ส่งรอ request ก็อาจกลับมาติดรอที่ queue ถ้าเลือก reject ต้องแจ้งให้ชัดว่า job นั้นยังไม่ได้เข้า queue ในตัวอย่าง Go เมื่อ queue เต็ม select จะเข้า default ทันที แทนที่จะรอให้ queue มีที่ว่าง

สำหรับ rider job การ print log ว่า queue เต็มยังไม่เพียงพอ เพราะ order นั้นยังต้องมีคนรับผิดชอบ หาก API ยังไม่ได้รับ job ไว้ อาจตอบ error ให้ผู้เรียกรู้ว่าต้อง retry โดยต้องป้องกันไม่ให้ retry แล้วสร้าง job ซ้ำด้วย

แต่ถ้า database save status “ร้านรับ order แล้ว” ไปก่อนหน้า การตอบ error จะไม่ rollback status นั้นกลับให้เอง เราต้อง save status เพิ่มว่า order นี้ยังรอ submit rider job พร้อมมีขั้นตอนอ่านรายการเหล่านี้กลับมา submit ใหม่ Local queue ในตัวอย่างยังไม่ได้รองรับขั้นตอนนี้ จึงยังไม่พอสำหรับรับผิดชอบ order จริง

Worker pool ช่วยควบคุม concurrency แต่ไม่ได้จัดการ business workflow เหล่านี้ให้เรา การเพิ่ม worker ก็ต้องดู capacity ของ database และ service ที่ worker เรียกใช้ด้วย ไม่เช่นนั้นระบบเหล่านั้นอาจรับโหลดไม่ไหวแทน

เพิ่มเป็นสอง instance แล้ว job จะกระจายตามไหม?

เมื่อ request มากขึ้น เราเพิ่ม app เป็นสอง instance และใช้ load balancer กระจาย request แต่ละ instance เป็นคนละ process จึงมี queue และ worker ของตัวเอง

flowchart LR
    LB["Load balancer"] --> A["Instance A: local queue + workers"]
    LB --> B["Instance B: local queue + workers"]
    A --> QA["Jobs รอใน queue ของ A"]
    B --> QB["Jobs รอใน queue ของ B"]

    classDef neutral fill:#f1f5f9,stroke:#64748b,color:#0f172a
    class LB,A,B neutral
    classDef broker fill:#fef3c7,stroke:#a16207,color:#0f172a
    class QA,QB broker

เรามีกำลังประมวลผลรวมเพิ่มขึ้น แต่ถ้า A มี job รออยู่มาก ขณะที่ worker ของ B ว่าง B ก็เข้ามาอ่าน queue ใน memory ของ A ไม่ได้ การเพิ่ม instance ไม่ได้เปลี่ยน local queue ให้กลายเป็น shared queue Load balancer กระจายเฉพาะ request ที่เข้ามา ไม่ได้ย้าย job ที่อยู่ใน queue ของ A ไปให้ B

อีกปัญหาจะเห็นตอน restart หรือ deploy ถ้า process หยุดกะทันหัน job ที่รอใน queue จะหายไปพร้อม memory ของ process และ job ที่ worker กำลังทำอาจไม่เสร็จ

การหยุดรับ job ใหม่แล้วรอให้ worker ทำ job ที่เหลือจนเสร็จ เรียกว่า graceful shutdown วิธีนี้ช่วยได้เมื่อเราควบคุมการปิด process ได้ แต่ไม่ครอบคลุมกรณี process crash หรือเครื่องดับ

ขอบเขตของสิ่งที่เราทำได้ตอนนี้

เราแยก job ออกจาก request และควบคุม concurrency ได้ โดยยังไม่มี infrastructure ใหม่ให้ดูแล แต่ solution นี้ยังมีข้อจำกัดสองเรื่อง: worker อีก process อ่าน local queue นี้โดยตรงไม่ได้ และ job ใน memory ไม่อยู่ต่อหลัง process หยุด

ถ้าต้องการแชร์ job ข้าม process เราต้องมีช่องทางส่ง message ผ่าน network ถ้าต้องการให้ job ยังอยู่หลัง restart เราต้องมี persistent storage เช่นเก็บ job ลง disk หรือ database พร้อมวิธีอ่าน job ที่ยังไม่เสร็จกลับมาทำต่อ

ข้อจำกัดเหล่านี้มาจาก architecture ที่เลือก ไม่ได้ผูกกับภาษาโปรแกรม เราสร้างกลไกเพิ่มเติมด้วย standard library ได้ แต่ก็ต้อง implement acknowledgement, retry และ recovery เองด้วย

เมื่อ worker อยู่คนละ process เราจะส่ง job ให้กันอย่างไร?

สำหรับ food delivery app ถ้า rider job หาย order นั้นก็อาจไม่มีใครไปรับ การลดเวลารอของ request จึงเป็นเพียงปัญหาแรกที่เราแก้ได้ เรายังต้องตอบให้ได้ว่า จะส่ง job ให้ worker อีก process อย่างไร และใครจะเก็บ job ไว้เมื่อ worker ยังไม่พร้อม?

ตอนที่ 3 จะเริ่มจากแยก Rider Matching worker ออกจาก application แล้วใช้ตัวกลางส่ง message ให้ worker ที่เชื่อมต่ออยู่ จากนั้นเราจะตรวจว่า solution นี้ตอบโจทย์ job ที่หายไม่ได้แล้วหรือยัง