Golang The Series EP.170: สร้างระบบ AI Batch Processing ด้วย Golang Worker Pool

ยินดีต้อนรับเข้าสู่ EP.170 ครับ! ในที่สุดเราก็เดินทางมาถึง Workshop ใหญ่ปิดท้ายภาคการสร้างโครงสร้างพื้นฐาน AI API ในซีรีส์ของ Superdev Academy หลังจากที่เราได้เรียนรู้ทั้งการจัดการ Multi-LLM, Redis Cache, Rate Limiting, Load Balancing, Circuit Breaker, Prometheus Metrics ไปจนถึง Error Handling และ Concurrency กันมาอย่างครบถ้วน

ในโลกความเป็นจริงขององค์กร งาน AI ไม่ได้มีแค่การตอบแชตทีละข้อความ (Transactional Query) แต่บ่อยครั้งเราต้องรับมือกับ "Batch Workloads" ขนาดใหญ่ เช่น การสรุปคำติชมลูกค้า 10,000 รายการ, การสกัดข้อมูลสารบัญจากเอกสาร PDF นับพันฉบับ หรือการทำ Sentiment Analysis ประจำสัปดาห์

หากเราส่งข้อมูลทีละรายการในรูปแบบ Synchronous Loop ระบบจะใช้เวลานานเป็นวัน และหากเกิด Crash กลางทาง ข้อมูลทั้งหมดที่ทำมาอาจหายวับไปทันที! วันนี้เราจะใช้ภาษา Go สร้าง High-Performance AI Batch Processor ที่ประมวลผลข้อมูลนับพันรายการได้ภายในไม่กี่นาที ด้วยสถาปัตยกรรม Worker Pool ที่ปลอดภัย เสถียร และมีระบบติดตามผลครบครันครับ!

สถาปัตยกรรมระบบ AI Batch Processor

ระบบของเราจะแบ่งออกเป็น 4 ส่วนหลัก ทำงานร่วมกันผ่าน Goroutines และ Channels:

Plaintext

+------------------+      +-------------------+      +-------------------+      +------------------+
|  Job Dispatcher  | ---> |    Worker Pool    | ---> |   AI Processing   | ---> |  Result Handler  |
| (อ่านข้อมูลเตรียมใส่ |      | (รัน Worker ขนาน   |      |   (ยิง API +      |      |  (บันทึกผลลัพธ์ /  |
|   Job Channel)   |      |    ตามจำนวน n)    |      |   Circuit Breaker)|      |   ทำ Monitoring) |
+------------------+      +-------------------+      +-------------------+      +------------------+

• Job Dispatcher: ทำหน้าที่กระจายงานลงใน Buffered Channel (JobQueue)

• Worker Pool: Goroutines ตามจำนวนที่กำหนด (เช่น 50 workers) ดึงงานไปทำขนานกัน

• AI Processing: จำลอง/เรียกใช้ LLM API พร้อมมีระบบจับเวลาและ Error Tracking

• Result Handler & Collector: Goroutine แยกต่างหากที่คอยรับผลลัพธ์จาก ResultQueue เพื่อบันทึกลง Database/Log โดยไม่ขัดจังหวะ Workers

โค้ดโปรเจกต์ Workshop: AI Batch Processing System

เราจะแบ่งโค้ดออกเป็น 3 ส่วนหลักเพื่อให้ทำความเข้าใจได้ง่ายขึ้น ให้สร้างไฟล์ main.go แล้วเริ่มเขียนไปทีละส่วนครับ

ส่วนที่ 1: การกำหนดโครงสร้างข้อมูล (Structs & Initialization)

ส่วนแรกคือการเตรียมโครงสร้างสำหรับเก็บข้อมูลงาน ผลลัพธ์ และตัวคุมระบบ Worker Pool

Go

package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
)

// Job โครงสร้างข้อมูลงานแต่ละชิ้นที่ต้องให้ AI ประมวลผล
type Job struct {
	ID      int
	Payload string
}

// Result โครงสร้างผลลัพธ์หลังประมวลผลสำเร็จหรือล้มเหลว
type Result struct {
	JobID  int
	Output string
	Err    error
}

// BatchProcessor ตัวควบคุมระบบ Worker Pool
type BatchProcessor struct {
	WorkerCount  int
	JobQueue     chan Job
	ResultQueue  chan Result
	SuccessCount uint64
	FailureCount uint64
}

// NewBatchProcessor สร้าง Instance พร้อมกำหนดขนาด Queue
func NewBatchProcessor(workerCount int, queueSize int) *BatchProcessor {
	return &BatchProcessor{
		WorkerCount: workerCount,
		JobQueue:    make(chan Job, queueSize),
		ResultQueue: make(chan Result, queueSize),
	}
}

ส่วนที่ 2: ฟังก์ชันจำลอง AI และการรัน Worker Pool

ฟังก์ชัน callAIModel จะจำลองการเรียกใช้งาน AI API (มีการดีเลย์และสุ่มให้เกิด Error) ส่วน StartWorkers จะทำการสร้าง Goroutine ตามจำนวนที่เรากำหนดเพื่อรอรับงานจาก JobQueue

Go

// callAIModel จำลองการส่งข้อมูลไปประมวลผลที่ AI API
func (bp *BatchProcessor) callAIModel(ctx context.Context, job Job) (string, error) {
	// จำลองเวลาประมวลผลของ AI (200ms - 600ms)
	processTime := time.Duration(200+rand.Intn(400)) * time.Millisecond

	select {
	case <-time.After(processTime):
		// จำลองโอกาสเกิดความผิดพลาด random 5%
		if rand.Float32() < 0.05 {
			return "", fmt.Errorf("AI Provider Error on Job #%d", job.ID)
		}
		return fmt.Sprintf("สรุปข้อมูลรายการที่ %d: '%s' [สำเร็จ]", job.ID, job.Payload), nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

// StartWorkers เริ่มต้นการทำงานของ Goroutine Workers ตามจำนวนที่กำหนด
func (bp *BatchProcessor) StartWorkers(ctx context.Context, wg *sync.WaitGroup) {
	for i := 1; i <= bp.WorkerCount; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			for job := range bp.JobQueue {
				// ยิงประมวลผล AI
				output, err := bp.callAIModel(ctx, job)
				
				// ใช้ atomic เพื่อความปลอดภัยเวลาหลาย Worker อัปเดตตัวแปรพร้อมกัน (ป้องกัน Race Condition)
				if err != nil {
					atomic.AddUint64(&bp.FailureCount, 1)
				} else {
					atomic.AddUint64(&bp.SuccessCount, 1)
				}
				
				// ส่งผลลัพธ์เข้าคิวเพื่อนำไปจัดการต่อ
				bp.ResultQueue <- Result{JobID: job.ID, Output: output, Err: err}
			}
		}(i)
	}
}

ส่วนที่ 3: ฟังก์ชัน Main (ประกอบร่างและสั่งการ)

ในส่วนสุดท้าย เราจะจำลองการป้อนงาน 1,000 รายการ และรวบรวมผลลัพธ์ผ่านตัว Collector เพื่อสรุปผลการทำงาน

Go

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	totalJobs := 1000  // จำนวนงานทั้งหมด 1,000 รายการ
	workerCount := 50  // รัน Worker ขนานกัน 50 ตัว

	processor := NewBatchProcessor(workerCount, totalJobs)

	var workerWg sync.WaitGroup
	var resultWg sync.WaitGroup

	startTime := time.Now()

	// 1. เริ่มต้น Worker Pool
	processor.StartWorkers(ctx, &workerWg)

	// 2. Goroutine รับผลลัพธ์ (Result Collector)
	resultWg.Add(1)
	go func() {
		defer resultWg.Done()
		for result := range processor.ResultQueue {
			if result.Err != nil {
				// ในระบบจริง: บันทึกลง Error Log หรือส่งไป retry queue
			} else {
				// ในระบบจริง: บันทึกลง Database หรือส่งออกเป็น CSV/JSON
			}
		}
	}()

	// 3. Dispatcher: ป้อนงาน 1,000 รายการลง Queue
	fmt.Printf("🚀 เริ่มต้นส่งงาน %d รายการเข้าสู่ Worker Pool (%d Workers)...\n", totalJobs, workerCount)
	for i := 1; i <= totalJobs; i++ {
		processor.JobQueue <- Job{
			ID:      i,
			Payload: fmt.Sprintf("ความคิดเห็นจากลูกค้าคนที่ %d", i),
		}
	}
	close(processor.JobQueue) // ปิด Job Channel เพื่อแจ้ง Worker ว่าหมดงานแล้ว

	// 4. รอให้ Workers ทำงานครบทั้งหมด แล้วปิด Result Queue
	workerWg.Wait()
	close(processor.ResultQueue)

	// 5. รอให้ Collector เก็บรวบรวมผลลัพธ์เสร็จสิ้น
	resultWg.Wait()

	totalDuration := time.Since(startTime)

	// 6. สรุปรายงานการประมวลผล
	fmt.Println("\n==============================================")
	fmt.Println("📊 สรุปผลการประมวลผล AI Batch Processing")
	fmt.Println("==============================================")
	fmt.Printf("⏱️ เวลารวมที่ใช้      : %v\n", totalDuration)
	fmt.Printf("✅ ประมวลผลสำเร็จ     : %d รายการ\n", processor.SuccessCount)
	fmt.Printf("❌ ประมวลผลล้มเหลว    : %d รายการ\n", processor.FailureCount)
	fmt.Printf("⚡ อัตราความเร็ว (RPS) : %.2f Requests/sec\n", float64(totalJobs)/totalDuration.Seconds())
	fmt.Println("==============================================")
}

ผลลัพธ์และการวิเคราะห์ประสิทธิภาพ

เมื่อรันโค้ดด้านบนด้วย go run main.go คุณจะเห็นผลลัพธ์สถิติการประมวลผลดังนี้:

Plaintext

🚀 เริ่มต้นส่งงาน 1000 รายการเข้าสู่ Worker Pool (50 Workers)...

==============================================
📊 สรุปผลการประมวลผล AI Batch Processing
==============================================
⏱️ เวลารวมที่ใช้      : 8.12s
✅ ประมวลผลสำเร็จ     : 952 รายการ
❌ ประมวลผลล้มเหลว    : 48 รายการ
⚡ อัตราความเร็ว (RPS) : 123.15 Requests/sec
==============================================

ทำไมวิธีนี้ถึงทรงพลัง?

• ลดเวลาลงมากกว่า 50 เท่า: หากประมวลผลทีละรายการ (Sequential) 1,000 รายการ × 400ms จะใช้เวลานานถึง 400 วินาที (~6.6 นาที) แต่ด้วย Go Worker Pool 50 ตัว เราประมวลผลเสร็จในเพียง 8 วินาที เท่านั้น!

• Memory Constrained: การใช้ Bounded Channel และ Atomic Counters ช่วยควบคุมการใช้ RAM ไม่ให้พุ่งสูงเกินไป แม้จะมีงานเข้ามารอนับแสนรายการก็ตาม

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

ลองนำโค้ด Workshop นี้ไปทดสอบรันในเครื่องของคุณ แล้วลองปรับเปลี่ยนจำนวน workerCount (เช่น จาก 50 เป็น 10 หรือ 100) เพื่อเปรียบเทียบผลลัพธ์ความเร็ว

การบ้านชวนคิด: หากในระหว่างทำ Batch Processing เกิดมี Rate Limit จากผู้ให้บริการ AI สั่งตัดไฟลงมา (เกิด 429 Too Many Requests) คุณจะนำแพ็กเกจ golang.org/x/time/rate ที่เราเรียนใน EP.164 มาครอบจังหวะการหยิบงานออกจาก JobQueue ของ Workers เพื่อไม่ให้ยิงเกินโควตาได้อย่างไร? ลองปรับแต่งโค้ดกันดูนะครับ!

🙋‍♂️ คำถามที่พบบ่อย (FAQ)

ทำไมถึงใช้ Channel กระจายงาน แทนที่จะสร้าง slice แล้วใช้ go func() วนลูปยิงตรงๆ เลย?

การสร้าง Goroutine แบบไม่จำกัดจำนวน (Unbounded) เช่นการสั่ง go func() 1,000 ครั้งพร้อมกัน อาจทำให้เกิดปัญหา Out of Memory หรือทำให้การเชื่อมต่อเครือข่ายพังได้ (Connection Exhaustion) การใช้ Worker Pool + Channel ช่วยให้เราควบคุม Concurrency ไว้ที่ระดับที่เหมาะสม (เช่น 50 ตัว) ทำให้ระบบนิ่งและเสถียรกว่ามาก

หากต้องการให้ระบบทำการ "ทำซ้ำ (Retry)" งานที่ Error โดยอัตโนมัติ ควรทำอย่างไร?

คุณสามารถปรับปรุงระบบ Result Collector ได้ครับ หากเช็กแล้วพบว่า result.Err != nil แทนที่จะบันทึกลง Error Log ทันที ก็ให้ส่ง result.JobID กลับเข้าไปใน JobQueue อีกครั้ง (หรือสร้าง RetryQueue แยกต่างหาก) แต่ระวังอย่าลืมจำกัดจำนวนครั้งที่ให้ Retry ด้วยเพื่อป้องกันการเกิด Infinite Loop

ควรกำหนด workerCount (จำนวน Worker) เท่าไหร่ถึงจะดีที่สุด?

สำหรับงานประเภท I/O Bound เช่นการยิง API หรือเรียก Database (เหมือนใน Workshop นี้) จำนวน Worker สามารถเซตได้สูง (เช่น 50, 100, หรือ 200) ทั้งนี้ขึ้นอยู่กับ Rate Limit ของผู้ให้บริการ API แต่ถ้าเป็นงานประเภท CPU Bound (การคำนวณหนักๆ ในเครื่องตัวเอง) ควรกำหนดให้ใกล้เคียงกับจำนวน Core ของ CPU ในเครื่องเซิร์ฟเวอร์ครับ

สรุป

ในบทความนี้เราได้เรียนรู้วิธีการออกแบบและสร้างระบบ AI Batch Processing ระดับ Enterprise ด้วยสถาปัตยกรรม Worker Pool ในภาษา Go ซึ่งช่วยให้เราสามารถจัดการกับภาระงานขนาดใหญ่ได้อย่างมีประสิทธิภาพ เราได้เห็นพลังของการทำงานแบบ Concurrency ผ่าน Goroutines การส่งต่อข้อมูลอย่างปลอดภัยผ่าน Channels ตลอดจนการจัดการผลลัพธ์แบบ Asynchronous สิ่งเหล่านี้คือหัวใจสำคัญที่ทำให้ Go กลายเป็นภาษายอดนิยมในการสร้างระบบ Backend ครับ

ในตอนต่อไป (EP.171): ยินดีด้วยครับ! คุณได้ผ่านหลักสูตรโครงสร้างพื้นฐาน AI Backend ทั้งหมดเรียบร้อยแล้ว ในตอนหน้าเราจะก้าวเข้าสู่ "ซีรีส์ใหม่ / หัวข้อใหม่ขั้นสูง" นั่นคือโลกของ AI Autonomous Agents ในบทความ "EP.171: Intro to AI Agents เมื่อ AI สามารถตัดสินใจใช้เครื่องมือเองได้" เราจะมาดูกันว่าจากการเป็นแค่ LLM ตอบคำถาม AI จะกลายร่างมาเป็นระบบที่คิด วิเคราะห์ และรันโค้ด หรือเรียกใช้ API แก้ปัญหาให้เราได้อย่างไร ห้ามพลาดเด็ดขาดครับ Gophers!

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

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

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

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

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

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