Go กับ WebSocket Cluster - เพิ่มพลังการรองรับผู้ใช้แบบกระจายตัว!

ใน EP นี้ เราจะพูดถึงการสร้าง WebSocket Server ที่สามารถทำงานในรูปแบบ Cluster เพื่อรองรับการใช้งานแบบกระจายตัว เพิ่มความสามารถในการรองรับผู้ใช้งานจำนวนมาก และช่วยเพิ่มความเสถียรของระบบครับ!

Cluster คืออะไร?

Cluster คือการรวมเซิร์ฟเวอร์หลายตัวให้ทำงานร่วมกันเสมือนเป็นระบบเดียว โดยมี Load Balancer คอยกระจายงาน (เช่น การเชื่อมต่อ WebSocket) ไปยังเซิร์ฟเวอร์แต่ละตัวใน Cluster

ข้อดีของการทำงานแบบ Cluster :

1. รองรับผู้ใช้จำนวนมาก : กระจายการเชื่อมต่อไปยังเซิร์ฟเวอร์หลายตัว

2. เพิ่มความเสถียร : หากเซิร์ฟเวอร์บางตัวล่ม ระบบยังคงทำงานได้

3. ปรับขนาดได้ง่าย : เพิ่มหรือลดจำนวนเซิร์ฟเวอร์ตามความต้องการ

การจัดการ WebSocket แบบ Cluster

ปัญหาหลักที่ต้องแก้ไขใน Cluster

1. Session Stickiness : ทำให้การเชื่อมต่อของผู้ใช้เดิมส่งไปยังเซิร์ฟเวอร์ตัวเดิม

2. การกระจายข้อความ (Message Broadcasting) : ให้ข้อความที่ส่งจากเซิร์ฟเวอร์หนึ่งกระจายไปยังผู้ใช้ทุกคนใน Cluster

3. การซิงโครไนซ์สถานะ (State Synchronization) : รักษาสถานะของผู้ใช้ในเซิร์ฟเวอร์หลายตัว

การสร้าง WebSocket Cluster ด้วย Redis Pub/Sub

Redis Pub/Sub ช่วยให้เซิร์ฟเวอร์หลายตัวสามารถสื่อสารกันได้ โดยเซิร์ฟเวอร์ทุกตัวจะ Subscribe (ติดตาม) ข้อความจากช่องเดียวกัน (Channel)

ตัวอย่างการใช้ Redis Pub/Sub สำหรับ WebSocket Cluster

1. ติดตั้ง Redis และ Go Redis Library

ติดตั้ง Redis บนเซิร์ฟเวอร์ของคุณ

sudo apt install redis

เพิ่มไลบรารี Go Redis:

go get github.com/redis/go-redis/v9

2. ตัวอย่างโค้ด WebSocket Cluster

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"

    "github.com/gorilla/websocket"
    "github.com/redis/go-redis/v9"
)

var (
    upgrader = websocket.Upgrader{
        CheckOrigin: func(r *http.Request) bool {
            return true
        },
    }
    clients  = make(map[*websocket.Conn]bool)
    broadcast = make(chan []byte)
    rdb      = redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    ctx = context.Background()
)

func handleConnections(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("Error upgrading connection:", err)
        return
    }
    defer conn.Close()

    clients[conn] = true
    defer delete(clients, conn)

    for {
        _, msg, err := conn.ReadMessage()
        if err != nil {
            log.Println("Error reading message:", err)
            break
        }

        // ส่งข้อความไปยัง Redis Channel
        if err := rdb.Publish(ctx, "websocketChannel", msg).Err(); err != nil {
            log.Println("Error publishing to Redis:", err)
        }
    }
}

func handleRedisMessages() {
    sub := rdb.Subscribe(ctx, "websocketChannel")
    ch := sub.Channel()

    for msg := range ch {
        log.Printf("Received message from Redis: %s", msg.Payload)
        broadcast <- []byte(msg.Payload)
    }
}

func handleBroadcasts() {
    for {
        msg := <-broadcast
        for client := range clients {
            if err := client.WriteMessage(websocket.TextMessage, msg); err != nil {
                log.Println("Error writing message:", err)
                client.Close()
                delete(clients, client)
            }
        }
    }
}

func main() {
    http.HandleFunc("/ws", handleConnections)

    go handleBroadcasts()
    go handleRedisMessages()

    log.Println("WebSocket server started at :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

อธิบายโค้ด

1. เชื่อมต่อ Redis : เซิร์ฟเวอร์ทุกตัว Subscribe ไปยัง websocketChannel

2. ส่งข้อความไปยัง Redis : เมื่อมีผู้ใช้ส่งข้อความ ข้อความจะถูก Publish ไปยัง Redis

3. กระจายข้อความจาก Redis : ข้อความที่ได้รับจาก Redis จะถูกส่งไปยังผู้ใช้ทุกคนที่เชื่อมต่อกับเซิร์ฟเวอร์

การทดสอบระบบ Cluster

1. รัน WebSocket Server หลายตัว (พอร์ตต่างกัน เช่น :8080, :8081)

2. ใช้ NGINX หรือ Load Balancer กระจายคำขอไปยังเซิร์ฟเวอร์แต่ละตัว

3. เชื่อมต่อไคลเอนต์หลายตัว และส่งข้อความเพื่อดูผลลัพธ์

ปรับปรุงเพิ่มเติม

1. เพิ่ม Sticky Sessions : ใช้ Load Balancer เช่น NGINX เพื่อติดตามเซสชันผู้ใช้

2. Monitor Redis : ใช้เครื่องมือเช่น RedisInsight เพื่อตรวจสอบข้อความและช่อง Redis

3. รองรับการ Scaling อัตโนมัติ : ใช้ Kubernetes สำหรับจัดการจำนวนเซิร์ฟเวอร์

สรุปง่ายๆ

ใช้ Redis Pub/Sub สำหรับการกระจายข้อความใน WebSocket Cluster
ใช้ NGINX หรือ Load Balancer เพื่อกระจายการเชื่อมต่อ
ปรับปรุงการซิงโครไนซ์สถานะและเพิ่ม Sticky Sessions เพื่อเพิ่มประสิทธิภาพ