มาลองประยุกต์ใช้ Pattern ต่างๆ

Pubsub

Pub/sub เป็นรูปแบบการสื่อสารระหว่าง goroutines ต่าง ๆ โดย goroutine หนึ่งสามารถส่งข้อความไปยัง goroutine อื่น ๆ หลายตัวได้ ในการประยุกต์ใช้ pub/sub กับ goroutines เราสามารถใช้ channel เพื่อส่งข้อความระหว่าง goroutines

ตัวอย่าง Pattern ทั่วไป

package main

import (
  "fmt"
  "time"
)

func main() {
  // สร้าง channel เพื่อส่งข้อความ
  ch := make(chan string)

  // สร้าง goroutine เพื่อส่งข้อความไปยัง channel
  go func() {
    for i := 0; i < 10; i++ {
      ch <- fmt.Sprintf("Hello, world! %d", i)
      time.Sleep(1 * time.Second)
    }
  }()

  // สร้าง goroutine เพื่อรับข้อความจาก channel
  go func() {
    for {
      msg := <-ch
      fmt.Println(msg)
    }
  }()

  // รอให้ goroutines ทำงานเสร็จสิ้น
  time.Sleep(5 * time.Second)
}

การใช้งานร่วมกับ Fiber

Note

• ไอเดียคือ เราจะใช้ Fiber รับคำสั่งมา
• และนำคำสั่งนั้น ส่งผ่าน channel เข้าไป เพื่อให้ subscribe ข้อมูลเอาไว้ได้

ลง Package Fiber

go get github.com/gofiber/fiber/v2

main.go

package main

import (
  "fmt"
  "github.com/gofiber/fiber/v2"
)

var pubsub = &PubSub{
  Subscribers: make(map[uint]*Subscriber),
}

const (
  host     = "localhost"  // or the Docker service name if running in another container
  port     = 5432         // default PostgreSQL port
  user     = "myuser"     // as defined in docker-compose.yml
  password = "mypassword" // as defined in docker-compose.yml
  dbname   = "mydatabase" // as defined in docker-compose.yml
)

func main() {
  app := fiber.New()

  // Start a subscription routine
  go func() {
    subscriber := pubsub.Subscribe()
    defer pubsub.Unsubscribe(subscriber.ID)

    for msg := range subscriber.Channel {
      // Handle the message, e.g., log, process, etc.
      fmt.Printf("Received message: %s\n", msg.Content)
    }
  }()

  // Publish endpoint
  app.Post("/publish", func(c *fiber.Ctx) error {
    var msg Message
    if err := c.BodyParser(&msg); err != nil {
      return err
    }

    go pubsub.Publish(msg)
    return c.SendString("Message published")
  })

  app.Listen(":8888")
}

model.go

package main

import (
  "sync"
)

// Message represents a simple message structure
type Message struct {
  Content string
}

// Subscriber represents a subscriber to a topic
type Subscriber struct {
  ID      uint `gorm:"primaryKey"`
  Channel chan Message
}

// PubSub represents the publish/subscribe mechanism
type PubSub struct {
  mu           sync.RWMutex
  Subscribers  map[uint]*Subscriber
  subscriberID uint
}

func (ps *PubSub) Subscribe() *Subscriber {
  ps.mu.Lock()
  defer ps.mu.Unlock()

  ps.subscriberID++
  subscriber := &Subscriber{
    ID:      ps.subscriberID,
    Channel: make(chan Message),
  }
  ps.Subscribers[subscriber.ID] = subscriber
  return subscriber
}

func (ps *PubSub) Unsubscribe(id uint) {
  ps.mu.Lock()
  defer ps.mu.Unlock()

  close(ps.Subscribers[id].Channel)
  delete(ps.Subscribers, id)
}

func (ps *PubSub) Publish(message Message) {
  ps.mu.RLock()
  defer ps.mu.RUnlock()

  for _, subscriber := range ps.Subscribers {
    subscriber.Channel <- message
  }
}

Cronjob

Cronjob เป็นเครื่องมือที่ใช้ในการเรียกใช้งานงานซ้ำ ๆ ในช่วงเวลาที่กำหนด ในการประยุกต์ใช้ cronjob กับ goroutines เราสามารถใช้ goroutine เพื่อเรียกใช้งานงานซ้ำ ๆ ในช่วงเวลาที่กำหนด โดยเราสามารถกำหนดช่วงเวลาที่ต้องการเรียกใช้งานงานโดยใช้ channel

ตัวอย่าง Pattern ทั่วไป

package main

import (
  "fmt"
  "time"
)

func main() {
  // สร้าง channel เพื่อกำหนดช่วงเวลาที่ต้องการเรียกใช้งานงาน
  ch := time.Tick(1 * time.Second)

  // สร้าง goroutine เพื่อเรียกใช้งานงานซ้ำ ๆ ในช่วงเวลาที่กำหนด
  go func() {
    for range ch {
      fmt.Println("Hello, world!")
    }
  }()

  // รอให้ goroutine ทำงานเสร็จสิ้น
  time.Sleep(5 * time.Second)
}

การใช้งานร่วมกับ GORM

go get -u gorm.io/gorm
go get -u gorm.io/driver/postgres

main.go

package main

import (
  "fmt"
  "github.com/robfig/cron/v3"
  "gorm.io/driver/postgres"
  "gorm.io/gorm"
  "log"
)

const (
  host     = "localhost"  // or the Docker service name if running in another container
  port     = 5432         // default PostgreSQL port
  user     = "myuser"     // as defined in docker-compose.yml
  password = "mypassword" // as defined in docker-compose.yml
  dbname   = "mydatabase" // as defined in docker-compose.yml
)

func main() {
  // Configure your PostgreSQL database details here
  dsn := fmt.Sprintf("host=%s port=%d user=%s "+
    "password=%s dbname=%s sslmode=disable",
    host, port, user, password, dbname)
  db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
  if err != nil {
    panic("failed to connect to database")
  }
  // Migrate the schema
  db.AutoMigrate(&ExampleModel{})
  fmt.Println("Database migration completed!")

  c := cron.New()
  _, err = c.AddFunc("@every 1m", func() {
    go task(db)
  })

  if err != nil {
    log.Fatal("Error scheduling a task:", err)
  }

  c.Start()

  // Block the main thread as the cron job runs in the background
  select {}
}

model.go

package main

import (
  "fmt"
  "gorm.io/gorm"
  "log"
  "time"
)

// ExampleModel represents a database model
type ExampleModel struct {
  gorm.Model
  Name string
}

// task is the function to be executed on a schedule
func task(db *gorm.DB) {
  now := time.Now()
  log.Println("Task is being run...", now)

  // Example database operation
  newRecord := ExampleModel{Name: fmt.Sprintf("Record at %s", now.Format(time.RFC3339))}
  db.Create(&newRecord)
}

Pattern อื่นๆ

• Distributed systems: Pattern สำหรับการกระจายข้อมูลผ่าน goroutines
• Pipeline: Pattern นี้เชื่อมต่อ goroutine หลายตัวเข้าด้วยกัน โดยผลลัพธ์จาก goroutine หนึ่งเป็นอินพุตสำหรับ goroutine ถัดไป เหมาะกับงานที่เป็นลำดับ เช่น ประมวลผลข้อมูลเป็นขั้นตอน ทำได้โดยใช้ channel เชื่อมต่อ goroutine ต่างๆ
• Worker pool: Pattern นี้ใช้ pool ของ goroutines ที่ทำงานเสร็จแล้วรอรับงานใหม่แทนการสร้าง goroutine ใหม่ตลอดเวลา เหมาะกับงานจำนวนมากที่ไม่ขึ้นต่อกัน เช่น ประมวลผลรูปภาพ ทำได้โดยใช้ queue หรือ channel เพื่อเก็บงาน และใช้ goroutine ใน pool รับงานจาก queue/channel ไปประมวลผล
• Rate limiting: Pattern นี้ควบคุมจำนวนการทำงานของ goroutine ต่อช่วงเวลา เหมาะกับป้องกัน server โอเวอร์โหลด ทำได้โดยใช้ package sync/atomic หรือ channel ร่วมกับ timer

คำแนะนำในการใช้ goroutine

Go routine ของ Go เป็นเครื่องมือที่มีประสิทธิภาพในการทำให้โปรแกรมทำงานได้เร็วขึ้นโดยการใช้ทรัพยากร CPU ร่วมกัน โดยทั่วไปแล้ว เราควรใช้ go routine ใน use case ต่อไปนี้

• งานที่ต้องใช้เวลานาน เช่น การประมวลผลข้อมูลขนาดใหญ่ การดาวน์โหลดไฟล์ขนาดใหญ่ หรือการวิเคราะห์ข้อมูล
• งานที่ต้องทำงานซ้ำๆ เช่น การวนลูปเพื่อประมวลผลข้อมูลจำนวนมาก การอ่านข้อมูลจากอุปกรณ์ หรือการเชื่อมต่อกับ API
• งานที่ต้องทำงานพร้อมกัน เช่น การประมวลผลคำขอจากผู้ใช้หลายราย การดาวน์โหลดไฟล์หลายไฟล์พร้อมกัน หรือการวิเคราะห์ข้อมูลหลายชุดพร้อมกัน

ตัวอย่างของ use case ที่ควรใช้ go routine ได้แก่

• โปรแกรมที่ประมวลผลข้อมูลขนาดใหญ่ เช่น โปรแกรมประมวลผลภาพหรือเสียง โปรแกรมประมวลผลข้อมูลทางการแพทย์หรือทางการเงิน
• โปรแกรมที่ดาวน์โหลดไฟล์ขนาดใหญ่ เช่น โปรแกรมดาวน์โหลดเพลงหรือภาพยนตร์ โปรแกรมดาวน์โหลดไฟล์จากอินเทอร์เน็ต
• โปรแกรมที่วิเคราะห์ข้อมูลขนาดใหญ่ เช่น โปรแกรมวิเคราะห์ข้อมูลพฤติกรรมผู้บริโภค โปรแกรมวิเคราะห์ข้อมูลทางธุรกิจ
• โปรแกรมที่ให้บริการผู้ใช้จำนวนมาก เช่น โปรแกรมเว็บเซิร์ฟเวอร์ โปรแกรมแชทบอท โปรแกรมเกม

อย่างไรก็ตาม เราควรระมัดระวังในการเลือกใช้ go routine เนื่องจาก go routine แต่ละตัวจะใช้ทรัพยากร CPU อยู่ตลอดเวลา ดังนั้นหากเราใช้ go routine มากเกินไปก็อาจทำให้โปรแกรมทำงานช้าลงหรือทำให้เครื่องร้อนเกินไปได้

คำแนะนำในการใช้ go routine อย่างมีประสิทธิภาพ ได้แก่

• กำหนดจำนวน go routine ให้เหมาะสม โดยทั่วไปแล้ว เราควรใช้ go routine ไม่เกินจำนวน CPU ของเครื่อง หากเราใช้ go routine มากเกินไปก็อาจทำให้โปรแกรมทำงานช้าลงหรือทำให้เครื่องร้อนเกินไปได้
• ใช้ mutex หรือวิธีอื่นๆ เพื่อป้องกัน race condition Race condition คือ ปัญหาที่อาจเกิดขึ้นเมื่อหลาย goroutine พยายามเข้าถึงและแก้ไขทรัพยากรหรือข้อมูลร่วมกันในเวลาเดียวกัน การใช้ mutex หรือวิธีอื่นๆ จะช่วยป้องกัน race condition และทำให้โปรแกรมทำงานได้อย่างถูกต้อง
• ตรวจสอบสถานะ go routine อย่างสม่ำเสมอ หาก go routine ใดๆ ทำงานผิดปกติหรือค้างก็อาจทำให้โปรแกรมทำงานผิดพลาดได้ เราควรตรวจสอบสถานะ go routine อย่างสม่ำเสมอเพื่อหาปัญหาที่อาจเกิดขึ้น