Golang The Series EP.161: Async AI Tasks การใช้ Worker Pools จัดการคิวงาน AI

ยินดีต้อนรับเข้าสู่ EP.161 ครับ! หลังจากที่เราได้ร่วมกันสร้าง Internal AI Tool แยกเลเยอร์อย่างเป็นระบบไปในตอนที่แล้ว เมื่อคลังข้อมูลเริ่มใหญ่ขึ้นและมีพนักงานเข้ามาใช้งานพร้อมกัน สาย Backend จะต้องเจอกับความท้าทายใหม่ นั่นคือ "ปัญหาคอขวดเมื่อมีงานประมวลผลขนาดใหญ่เข้ามาพร้อมกัน"

ลองนึกภาพว่าถ้ามีพนักงานอัปโหลดไฟล์ PDF หนา 500 หน้าเข้ามาให้ AI สรุป หรือสั่งรัน Ingestion Pipeline แปลงเอกสารนับร้อยไฟล์พร้อมๆ กัน การประมวลผลผ่าน HTTP Request ตรงๆ (Synchronous) จะทำให้ผู้ใช้ต้องนั่งรอจนหน้าเว็บเกิด Timeout แน่นอน แถมยังเสี่ยงทำเซิร์ฟเวอร์ล่มจากการกิน CPU และ Memory จนทะลุขีดจำกัด

ทางออกที่ดีที่สุดในสถาปัตยกรรมของ Go คือการเปลี่ยนงานเหล่านั้นให้เป็น Async Tasks (งานประมวลผลเบื้องหลัง) และควบคุมปริมาณการทำงานขนานด้วย Worker Pools Pattern ครับ!

แนวคิด Worker Pools ในการควบคุมทรัพยากร AI

Worker Pools คือการสถาปนา Goroutines ขึ้นมาจำนวนหนึ่งที่คงที่ในระบบ (เช่น กำหนดให้มีพนักงานคอยประมวลผล 3 ตัว) เพื่อให้คอย "ดึงงาน" ออกมาจากคิวกลาง (Go Channel) ไปทำทีละงานแบบอซิงโครนัส (Asynchronous) ซึ่งช่วยแก้ปัญหาหลักๆ ได้ดังนี้ครับ:

• ป้องกัน Boundless Concurrency: ช่วยควบคุมไม่ให้ระบบสร้าง Goroutine ออกมาใหม่แบบไร้ขีดจำกัดตามจำนวน Request ซึ่งเป็นสาเหตุหลักที่ทำให้ Memory เต็ม (OOM - Out of Memory)

• Rate Limiting ในตัว: ช่วยให้เราควบคุมอัตราการยิงไปหา OpenAI, Claude หรือ Local Ollama ไม่ให้เกินโควตาจราจร (Rate Limit) ของ API Key หรือเกินขีดจำกัดพละกำลังของฮาร์ดแวร์เราด้วยครับ

ดีไซน์ระบบโครงสร้างคิวงาน (Job Queue Structural Design)

เพื่อให้โค้ดอ่านง่ายและนำไปขยายต่อในโปรเจกต์จริงได้ทันที เราจะแบ่งประเภทข้อมูลออกเป็น 2 ส่วนหลัก คือ Job (ตัวแทนชิ้นงานที่วิ่งเข้าคิว) และ Result (ผลลัพธ์ที่ได้หลังจากพนักงานประมวลผลเสร็จ) ดังนี้ครับ:

Go

package main

import (
	"context"
)

// Job แทนโครงสร้างงาน AI ที่เราต้องการทำเบื้องหลัง
type Job struct {
	ID       int
	FilePath string
	Ctx      context.Context // พ่วง Context สำหรับควบคุม Timeout ของแต่ละชิ้นงาน
}

// Result แทนผลลัพธ์ที่ได้จากการประมวลผลเสร็จสิ้น
type Result struct {
	JobID int
	Data  string
	Error error
}

ตัวอย่างการเขียน Go สร้าง Worker Pools จัดการคิวงาน AI

มาดูตัวอย่างการเขียนโค้ดเพื่อควบคุมการทำงานแบบ Async โดยจำลองการดึงไฟล์ PDF ในคิวไปประมวลผลสกัดข้อความและทำ Embedding ขนานกันครับ

Go

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// worker ทำหน้าที่เสมือนพนักงานเบื้องหลังที่คอยดึงงานจากช่อง jobs ไปทำ
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()

	// วนลูปดึงงานจาก Channel ไปทำเรื่อยๆ จนกว่า Channel จะถูกปิด
	for job := range jobs {
		fmt.Printf("👷 [Worker %d] เริ่มประมวลผลงานชิ้นที่ %d: %s\n", id, job.ID, job.FilePath)
		
		// จำลองกระบวนการที่ใช้เวลาสูง เช่น สกัดข้อความ PDF หรือยิงไปแปลงเวกเตอร์
		err := processAITask(job.Ctx, job.FilePath)
		
		if err != nil {
			fmt.Printf("❌ [Worker %d] งานชิ้นที่ %d ล้มเหลว: %v\n", id, job.ID, err)
			results <- Result{JobID: job.ID, Error: err}
			continue
		}
		
		fmt.Printf("✅ [Worker %d] ทำงานชิ้นที่ %d สำเร็จ!\n", id, job.ID)
		results <- Result{
			JobID: job.ID, 
			Data:  fmt.Sprintf("สกัดข้อความสำเร็จจากไฟล์ %s", job.FilePath), 
			Error: nil,
		}
	}
}

// ฟังก์ชันจำลองการทำงานประมวลผล AI
func processAITask(ctx context.Context, path string) error {
	select {
	case <-time.After(2 * time.Second): // จำลองการทำงาน 2 วินาที
		return nil
	case <-ctx.Done(): // ถ้าระบบหลักสั่ง Cancel หรือ Timeout ก่อน ให้เลิกทำทันที
		return ctx.Err()
	}
}

func main() {
	numJobs := 10
	numWorkers := 3 // โควตาพนักงาน: ควบคุมให้รันพร้อมกันสูงสุดแค่ 3 งานเพื่อประหยัด RAM/CPU

	jobs := make(chan Job, numJobs)
	results := make(chan Result, numJobs)

	var wg sync.WaitGroup
	ctx := context.Background()

	// 1. สตาร์ท Workers ตามจำนวนที่กำหนดไว้ (Fixed Pool Size)
	for w := 1; w <= numWorkers; w++ {
		wg.Add(1)
		go worker(w, jobs, results, &wg)
	}

	// 2. โยนงาน 10 ชิ้น เข้าสู่ท่อ Queue (จำลองสถานการณ์พนักงานรุมอัปโหลดไฟล์พร้อมกัน)
	for j := 1; j <= numJobs; j++ {
		jobCtx, _ := context.WithTimeout(ctx, 5*time.Second) // แต่ละงานมีเวลารันไม่เกิน 5 วินาที
		jobs <- Job{
			ID:       j, 
			FilePath: fmt.Sprintf("company_policy_part_%d.pdf", j),
			Ctx:      jobCtx,
		}
	}
	close(jobs) // สำคัญมาก: ปิดช่องรับงานเพื่อบอก Worker ว่าไม่มีงานใหม่เพิ่มแล้ว ลูป for range จะได้สิ้นสุด

	// 3. รอให้พนักงานทุกตัวเคลียร์งานจนเสร็จสิ้นทั้งหมด
	wg.Wait()
	close(results) // ปิดช่องรับผลลัพธ์

	// 4. ดึงผลลัพธ์ของคิวงานทั้งหมดมาสรุปรายงาน
	fmt.Println("\n🏁 งานในคิวประมวลผลเสร็จสิ้นทั้งหมดเรียบร้อย!")
	for res := range results {
		if res.Error != nil {
			fmt.Printf("⚠️ รายงานผล: งานชิ้นที่ %d เกิดข้อผิดพลาด: %v\n", res.JobID, res.Error)
		} else {
			fmt.Printf("ℹ️ รายงานผล: งานชิ้นที่ %d -> %s\n", res.JobID, res.Data)
		}
	}
}

การต่อยอดสู่ระบบ Production ขนาดใหญ่

เมื่อระบบของคุณเติบโตไปสู่ระดับ Enterprise ที่มีผู้ใช้งานพร้อมกันหลายพันคน สิ่งที่ต้องคำนึงถึงมีดังนี้ครับ:

• In-Memory to Distributed Queues: Go Channels ทำงานได้ดีมากในหน่วยความจำ (In-Memory) ของเซิร์ฟเวอร์เครื่องเดียว แต่ถ้าคุณต้องการขยายระบบ (Scaling) ไปเป็นแบบหลายเครื่อง (Microservices) แนะนำให้เปลี่ยนจาก Channel ไปใช้คิวภายนอกอย่าง RabbitMQ หรือ Redis (แนะนำไลบรารีชื่อ Asynq ของ Go) อย่างไรก็ตาม โครงสร้างดีไซน์ภายในตัว Worker ใน Go Application ก็ยังคงใช้แนวคิด Worker Pools นี้ในการจัดการอยู่ดีครับ

• Context Propagation: การส่ง context.Context พ่วงเข้าไปใน Job (ตามตัวอย่างโค้ดด้านบน) เป็นท่ามาตรฐานที่ขาดไม่ได้ เพราะช่วยให้เราสามารถส่งต่อสัญญาณเพื่อยกเลิกการทำงาน (Abort) ฝั่ง API ได้ทันที หากระบบเครือข่ายมีปัญหา หรือโมเดล LLM เจ้านั้นค้างเป็นเวลานานเกินไป

🎯 ท้าให้ลอง (Daily Mission)

ลองนำโครงโค้ด Worker Pools ตัวนี้ไปรันในเครื่องของคุณดูครับ แล้วลองปรับแต่งตัวแปร numWorkers จาก 3 ตัว ลดเหลือ 1 ตัว หรือเพิ่มเป็น 5 ตัวดูครับ

การบ้านชวนคิด: สังเกตความเร็วในการพ่นข้อความบน Terminal ดูครับว่า ระยะเวลาในการเคลียร์งานทั้งหมด 10 ชิ้นเปลี่ยนไปอย่างไร?

และโจทย์สำคัญ: หากคุณต้องการบันทึกสถานะความคืบหน้าของงาน (Job Status เช่น Pending -> Processing -> Success) ลงฐานข้อมูล เพื่อให้ฝั่งหน้าบ้าน (Frontend) สามารถยิง API มาดึงเปอร์เซ็นต์ Progress ไปแสดงผลบน UI ได้ คุณคิดว่าควรวางฟังก์ชันสำหรับอัปเดตสถานะฐานข้อมูลไว้ที่ส่วนใดในโครงสร้างโค้ดนี้? ลองดีไซน์สถาปัตยกรรมในหัวกันดูนะครับ!

FAQ: คำถามที่พบบ่อยเกี่ยวกับ Worker Pools ในงาน AI

เราควรตั้งจำนวน Worker (numWorkers) เท่าไหร่ดีที่สุด?

ไม่มีตัวเลขตายตัวครับ ขึ้นอยู่กับข้อจำกัดของระบบ หากเป็นงานที่กิน CPU/Memory สูง เช่น การรันโมเดล AI ในเครื่องตัวเอง (Local LLM) ควรตั้งจำนวนตามจำนวน Core ของ CPU แต่หากเป็นงานประเภท Network I/O เช่น การยิง API ไปหา OpenAI หรือ Claude จำนวน Worker จะขึ้นอยู่กับข้อจำกัด Rate Limit ของ API Key และพละกำลังของเซิร์ฟเวอร์เราในการรองรับการรันขนานครับ แนะนำให้เริ่มจากตัวเลขน้อยๆ (เช่น 3-5 ตัว) แล้วทำ Load Test เพื่อดูจุดที่เหมาะสมที่สุด

ถ้ามีงานในคิวมากเกินไปจนท่อ Channel เต็ม จะส่งผลอย่างไร?

หากไม่ได้กำหนดขนาดของ Channel ไว้ (Unbuffered Channel) หรือกำหนดไว้แต่เต็มแล้ว ตัว Go Application ฝั่งที่โยนงานเข้าคิว (Publisher) จะเกิดการ Block (หยุดรอ) ตรงบรรทัดที่ส่งงานเข้า Channel ทันที ซึ่งในระบบ Production จริง เรามักจะแก้ปัญหานี้ด้วยการใช้ Non-blocking select หรือเปลี่ยนไปใช้ Distributed Queue ภายนอกแทน

สรุป

ในบทความนี้เราได้เรียนรู้เรื่องการนำสถาปัตยกรรม Worker Pools Pattern มาประยุกต์ใช้ใน Go (Golang) เพื่อจัดการกับคิวงานประมวลผล AI แบบอซิงโครนัส (Asynchronous Tasks) ซึ่งช่วยแก้ปัญหาคอขวดบนเซิร์ฟเวอร์ได้อย่างมีประสิทธิภาพ ช่วยจำกัดการใช้ทรัพยากรไม่ให้เกิดปัญหา Memory เต็ม (OOM) และยังมีกลไก Rate Limiting ในตัวเพื่อควบคุมการเรียกใช้ AI API อีกด้วย การออกแบบโครงสร้างโค้ดด้วยรูปแบบนี้จะช่วยให้ระบบฝั่ง Backend ของเรามีความเสถียรและพร้อมรองรับการขยายตัวในระดับ Enterprise ได้เป็นอย่างดีครับ

ในตอนต่อไป (EP.162): เมื่อเราจัดการจัดคิวงานเบื้องหลังได้อย่างอยู่หมัดแล้ว ปัญหาถัดมาของระบบ AI ก็คือ "เราจะเลือกใช้โมเดล AI เจ้าไหนดี?" บางเจ้าวิเคราะห์สถิติเก่ง บางเจ้าแปลภาษาหรือสรุปข้อความเก่งกว่า ตอนหน้าเราจะพาไปปลดล็อกขีดจำกัดด้วยเรื่อง "Goroutines for Multi-LLM — เรียก AI หลายเจ้าพร้อมกันเพื่อเปรียบเทียบผลลัพธ์แบบ Parallel" มารุมถาม AI พร้อมกันขนานกันเถอะ gophers

ฝากกดติดตามพวกเราได้ที่ Superdev Academy ในทุกช่องทางนะครับ!

• 🔵 Facebook: Superdev Academy Thailand (อัปเดตข่าวสารและบทความใหม่)

• 🎬 YouTube: Superdev Academy Channel (ติวเข้มแบบวิดีโอ)

• 📸 Instagram: @superdevacademy (เกร็ดความรู้สั้นๆ และเบื้องหลังการทำงาน)

• 🎬 TikTok: @superdevacademy (Tips & Tricks ฉบับย่อยง่าย)

• 🌐 Website: superdevacademy.com (คลังบทความและคอร์สเรียนฉบับเต็ม)