การดู : 107

02/06/2026 01:24น.

การใช้ Go Channels และ Goroutines ในการจัดการ Stream Chunks จาก AI API

Golang The Series EP.148: Handling Streams - สร้างระบบ Chat Real-time ด้วย Go Channels

#Go Channels

#Go

#Golang

#Concurrency

#AI Streaming

#Goroutines

#Backend Latency

ยินดีต้อนรับเข้าสู่ EP.148 ครับ! ในตอนที่แล้วเราได้เรียนรู้วิธีรับข้อมูลแบบ Structured JSON ที่แม่นยำ 100% ไปแล้ว แต่ในการใช้งานจริงเรามักจะเจอกับปัญหาใหญ่คือ "ความช้า" (Latency) เพราะยิ่งเราสั่งให้ AI วิเคราะห์เยอะ การรอให้มันประมวลผลจนจบก้อน JSON อาจใช้เวลานานหลายวินาที จนผู้ใช้งานนึกว่าระบบค้างหรือแอปพังไปแล้ว

วันนี้เราจะมาแก้ปัญหานี้ด้วยการดึงจุดแข็งที่สุดของภาษา Go นั่นคือ Channels และ Concurrency มาจัดการกับระบบ Streaming เพื่อส่งข้อมูลแบบชิ้นเล็กๆ (Chunks) ทยอยแสดงผลออกมาแบบ Real-time ทันทีที่ AI เริ่มคิดคำแรกได้ ให้ความรู้สึกลื่นไหลเหมือนหน้าจอ ChatGPT ที่เราคุ้นเคยนั่นเองครับ

ทำไมต้องใช้ Go Channels ในการจัดการ Stream?

เมื่อเราเปิดโหมด Stream กับ AI API (ไม่ว่าจะเป็น OpenAI หรือ Ollama) พฤติกรรมการส่งข้อมูลจะเปลี่ยนไป จากเดิมที่มาเป็นก้อนใหญ่ก้อนเดียว จะเปลี่ยนเป็นการส่ง Chunks หรือเศษเสี้ยวของข้อความทยอยไหลมาตาม Network Connection ครับ

การใช้ Go Channels เข้ามาเป็นท่อกลางในการรับ-ส่งข้อมูลชิ้นเล็กๆ เหล่านี้ มีข้อดีที่เหนือกว่าการเขียนลูปแบบปกติ 3 ด้าน:

  • Decoupling (การแยกส่วน): ช่วยให้เราแยกหน้าที่กันชัดเจน (Separation of Concerns) ฝั่งหนึ่งมีหน้าที่แค่ดึงข้อมูลจาก AI ส่วนอีกฝั่งมีหน้าที่แค่แสดงผลให้ User ทั้งสองฝั่งไม่ต้องรู้จักกัน แค่คุยกันผ่าน Channel ก็พอ

  • Non-blocking (ทำงานไม่ขัดจังหวะ): Go สามารถประมวลผลข้อมูลชิ้นที่ 1 (เช่น ส่งไปแสดงที่หน้าจอ) ไปพร้อมๆ กับการรอรับข้อมูลชิ้นที่ 2 จาก Network ได้ทันทีผ่านการทำงานแบบ Concurrency ทำให้ไม่มีช่วงเวลาที่ระบบต้อง "หยุดรอ" โดยเปล่าประโยชน์

  • Type Safety (ความปลอดภัยของข้อมูล): เราสามารถกำหนดได้ว่า Channel นี้จะรับข้อมูลประเภทไหน (เช่น chan string หรือ chan MyStruct) ทำให้มั่นใจได้ว่าข้อมูลที่ไหลอยู่ในท่อจะถูกต้องตาม Schema ที่เราออกแบบไว้เสมอ ลดโอกาสเกิด Runtime Error

💡 Insight สำหรับ Gopher:

การใช้ Channel ในบริบทนี้เปรียบเสมือนการสร้างสายพานในโรงงานครับ API คือเครื่องจักรต้นทางที่วางของลงมาเรื่อยๆ และ UI คือพนักงานปลายทางที่คอยหยิบของไปโชว์ ถ้าไม่มีสายพาน (Channel) นี้ เราก็ต้องรอให้เครื่องจักรผลิตเสร็จทั้งหมดก่อนถึงจะเดินไปหยิบมาทีเดียว ซึ่งนั่นคือสาเหตุของความช้านั่นเอง

โครงสร้างการเขียน Stream ด้วย Go

เราจะสร้างฟังก์ชันที่ทำหน้าที่เป็นผู้ผลิต (Producer) โดยจะส่งค่ากลับไปเป็น Receive-only Channel (<-chan) ซึ่งเปรียบเสมือนการส่งมอบท่อที่ปลายทางเอาไว้รอรับข้อมูลได้อย่างเดียว ป้องกันการส่งข้อมูลย้อนกลับผิดฝั่งครับ

ตัวอย่างการ Implement:

Go

// ส่งคืนค่าเป็น <-chan เพื่อบอกว่า "ท่อนี้เอาไว้รับข้อมูลออกไปใช้อย่างเดียว"
func StreamAIResponse(ctx context.Context, client *openai.Client, prompt string) <-chan string {
    out := make(chan string)

    // รัน Goroutine แยกออกไปเพื่อไม่ให้ Block การทำงานของฟังก์ชันหลัก
    go func() {
        // สำคัญมาก: ต้องปิด Channel เมื่อจบงานเสมอ เพื่อบอกปลายทางว่า "ข้อมูลหมดแล้ว"
        defer close(out) 
        
        req := openai.ChatCompletionRequest{
            Model:  openai.GPT4o,
            Stream: true, // จุดเปลี่ยนสำคัญ: เปิดโหมด Streaming ให้ API ค่อยๆ พ่นข้อมูลออกมา
            Messages: []openai.ChatCompletionMessage{
                {Role: "user", Content: prompt},
            },
        }

        stream, err := client.CreateChatCompletionStream(ctx, req)
        if err != nil {
            // ในระดับโปรดักชัน ควรส่ง Error ออกไปทาง Channel หรือใช้ Logging
            return 
        }
        defer stream.Close()

        for {
            response, err := stream.Recv()
            // io.EOF คือสัญญาณบอกว่า AI พูดจบประโยคแล้ว
            if errors.Is(err, io.EOF) {
                return 
            }
            if err != nil {
                return
            }
            
            // ส่งข้อมูล "ส่วนต่าง" (Delta) ที่ได้เพิ่มมาเข้าท่อไปทันที
            if len(response.Choices) > 0 {
                out <- response.Choices[0].Delta.Content
            }
        }
    }()

    return out
}

🔍 เจาะลึกสิ่งที่เกิดขึ้นใน Code:

  1. go func() { ... }(): เราแยกการรอรับข้อมูลจาก API ไปไว้ใน Background (Goroutine) เพื่อให้ฟังก์ชัน StreamAIResponse คืนค่า Channel กลับไปให้ผู้ใช้ได้ทันทีโดยไม่ต้องรอ AI ตอบเสร็จ

  2. defer close(out): นี่คือมารยาทที่ดีของการใช้ Channel หากเราไม่ปิดท่อ ปลายทางที่รอรับข้อมูลด้วย range จะค้างอยู่แบบนั้นตลอดไป (Deadlock)

  3. stream.Recv(): ฟังก์ชันนี้จะ "Block" การทำงานภายใน Goroutine เพื่อรอรับข้อมูลชิ้นถัดไปจาก AI เมื่อได้มาแล้วเราก็รีบส่งเข้าท่อ out <- ทันที

การนำไปใช้งาน (Consumption)

ที่ฝั่ง Main หรือ Controller (ปลายทางของท่อ) เราไม่จำเป็นต้องรู้เรื่องความซับซ้อนของ API หรือการจัดการ Goroutine เลยครับ หน้าที่เดียวของเราคือการรอรับข้อมูลที่ไหลออกมาจาก Channel โดยใช้ keyword for range

ตัวอย่างการเรียกใช้งาน:

Go

// 1. เรียกใช้งานฟังก์ชันและรับ Channel กลับมา
responseChan := StreamAIResponse(ctx, client, "ช่วยอธิบายเรื่อง Go Channels หน่อย")

// 2. ใช้ range ในการดึงข้อมูลออกจากท่อจนกว่า Channel จะถูกสั่ง close()
for msg := range responseChan {
    // 3. แสดงผลทีละตัวอักษร/คำ ทันทีที่ได้รับข้อมูลมา
    fmt.Print(msg) 
}

fmt.Println("\n--- จบการทำ Streaming ---")

ทำไมวิธีนี้ถึงยอดเยี่ยม?

  • ความลื่นไหล (Smooth UX): ข้อความจะไม่กระตุกหรือรอมาทีเดียวเป็นก้อน แต่จะค่อยๆ ปรากฏบนจอ (Typing Effect) เหมือนมีคนกำลังพิมพ์อยู่จริงๆ ซึ่งช่วยลดความรู้สึกหงุดหงิดของผู้ใช้งานเวลาต้องรอ AI คิดนานๆ

  • จัดการหน่วยความจำได้ดี: เราไม่ต้องเก็บ String ยาวๆ ไว้ใน Memory จนจบการทำงาน แต่เราประมวลผล (ในที่นี้คือการพิมพ์ออกจอ) ไปได้เลยทีละนิด

  • Loop จะหยุดเองอัตโนมัติ: เมื่อ Goroutine ฝั่งต้นทางทำงานเสร็จและสั่ง close(out) ลูป for range นี้จะหลุดออกมาเองอย่างปลอดภัย (Graceful Exit) โดยที่เราไม่ต้องเขียนเงื่อนไขหยุดให้วุ่นวายครับ

ข้อควรระวังในการทำ Stream

ถึงแม้ Go จะจัดการ Concurrency ได้ดีมาก แต่เราในฐานะ Developer ต้องระวัง 2 เรื่องหลักเพื่อความปลอดภัยของระบบ:

1. Context Cancellation (การยกเลิกคำสั่ง)

นี่คือจุดที่พลาดกันบ่อยที่สุดครับ! หาก User กดหยุด (Stop) หรือปิดหน้า Browser ทิ้งไปในขณะที่ AI กำลังพ่นคำออกมา ถ้าเราไม่จัดการส่ง ctx (Context) เข้าไปในฟังก์ชัน CreateChatCompletionStream ตัว Backend ของเราจะยังคงรันลูปและรับข้อมูลจาก AI ต่อไปเรื่อยๆ จนจบ

  • ผลเสีย: เสีย Token ฟรี (เสียเงิน!), Server ทำงานหนักโดยไม่มีใครดูผลลัพธ์

  • วิธีแก้: ตรวจสอบเสมอว่า ctx ยังใช้งานได้อยู่หรือไม่ และต้องมั่นใจว่า stream.Close() จะถูกเรียกทันทีเมื่อ Context ถูกยกเลิก

2. Buffer Size & Backpressure (การจัดการแรงดันข้อมูล)

โดยปกติการทำ Chat เรามักใช้ Unbuffered Channel (ขนาด 0) เพื่อความสดใหม่ของข้อมูล เพราะเราอยากให้ข้อมูลไหลจาก API ถึงหน้าจอ User ทันทีโดยไม่ค้างอยู่ในท่อ

  • เมื่อไหร่ควรใช้ Buffered Channel?: หากฝั่ง "ผู้รับ" (Consumer) ทำงานช้ากว่าฝั่งผู้ส่ง (เช่น ต้องเอาข้อมูลไปผ่านฟิลเตอร์บางอย่างก่อนโชว์) การใช้ Buffered Channel (เช่น make(chan string, 10)) จะช่วยให้ฝั่งส่งไม่ต้องหยุดรอฝั่งรับจนเกินไป ช่วยลดอาการกระตุกของ Stream ได้ครับ

💡 สรุป Rule of Thumb สำหรับ Gopher:

"เปิดได้ ต้องปิดเป็น" — ทุกครั้งที่ใช้ go func และ channel ต้องตอบคำถามให้ได้เสมอว่า Goroutine นี้จะจบการทำงานเมื่อไหร่ และ Channel นี้จะถูกปิดตอนไหน ถ้าหาคำตอบไม่ได้ มีโอกาสสูงที่จะเกิด Goroutine Leak ซึ่งจะกัดกิน Memory ของ Server คุณไปเรื่อยๆ ครับ

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

เพื่อให้คุณเข้าใจการไหลของข้อมูลได้ลึกซึ้งขึ้น ผมอยากให้ลองอัปเกรดฟังก์ชัน Stream ให้มีความเป็นมืออาชีพมากขึ้น โดยไม่ได้ส่งแค่ข้อความดิบๆ แต่ให้ส่งข้อมูลที่มีบริบทครบถ้วนผ่าน Channel ครับ

โจทย์: ลองสร้าง Struct สำหรับส่งข้อมูล และปรับปรุงฟังก์ชันให้ส่งค่าผ่าน Channel ดังนี้:

Go

type StreamResponse struct {
    Content      string
    FinishReason string
    Err          error
}

// การบ้าน: ลองปรับ StreamAIResponse ให้คืนค่าเป็น <-chan StreamResponse

คำถามชวนคิด:

ลองจับเวลา (Benchmark) ในใจดูครับว่า ระหว่าง:

  1. Standard Mode: รอ 5 วินาที แล้วข้อความยาวๆ โผล่มาทีเดียว

  2. Streaming Mode: รอ 0.5 วินาที แล้วคำแรกเริ่มพ่นออกมาเรื่อยๆ จนจบในวินาทีที่ 5

คุณคิดว่าผู้ใช้งานจะกด Refresh หรือปิดแอปทิ้งในกรณีไหนมากกว่ากัน? นี่คือสิ่งที่เรียกว่า Perceived Performance หรือความเร็วที่ใจของคนใช้รู้สึก ซึ่งสำคัญพอๆ กับความเร็วของ Code เลยครับ

สรุป

การทำ Streaming ด้วย Go Channels เปลี่ยนประสบการณ์จากแอปที่ดูนิ่งค้างให้กลายเป็นแอปที่มีชีวิต การจัดการ Concurrency ที่สะอาดตาของ Go ช่วยให้เราเขียนระบบที่ดูเหมือนจะซับซ้อนให้กลายเป็นเรื่องง่ายและปลอดภัย

แต่ความเร็วที่เพิ่มขึ้นย่อมมาพร้อมกับความรับผิดชอบที่ใหญ่ยิ่ง...

ในตอนต่อไป (EP.149):

เมื่อระบบไหลลื่น ผู้ใช้แฮปปี้ แต่สิ่งที่อาจทำให้คุณไม่แฮปปี้คือบิลค่า API ตอนสิ้นเดือนครับ! เราจะมาดูวิธีควบคุมงบประมาณแบบเข้มข้นในตอน "Token Management: วิธีนับ Token และคำนวณต้นทุน API ในฝั่ง Backend"

เราจะมาเจาะลึกกันว่า AI มองตัวอักษรของเราเป็นเลขอะไร? และเราจะเดาใจค่าใช้จ่ายก่อนส่งไปหา API ได้อย่างไร? ใครไม่อยากให้บิลปลายเดือนช็อก... ห้ามพลาดครับ!

เจอกันตอนหน้าครับ Gophers!

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

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

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

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

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

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