Chuyện mấy con Consumer

func watchConsumerCount() {
  viper.WatchConfig()
  viper.OnConfigChange(func(e fsnotify.Event) {
    fmt.Println("Config file changed:", e.Name)
    n := viper.GetInt("consumer_count")
    if n != consumer_count {
      fmt.Println("resize consumer count to => ", n)
      go ResizeConsumerCount(n)
    }
  })
}

var amql_conn *amqp.Connection
var amqp_url string
var consumer []*ConsumerRoutine
const max_count = 100
var consumer_count int
var queue_name string
​
func loadConf() {
  viper.SetConfigName("config") 
  viper.SetConfigType("yaml")  
  viper.AddConfigPath(".")      
  err := viper.ReadInConfig()   
  if err != nil {             
    panic(fmt.Errorf("Fatal error config file: %s n", err))
  }
  amqp_url = viper.GetString("amqp_url")
  consumer_count = viper.GetInt("consumer_count")
  queue_name = viper.GetString("queue_name")
}
​
func initAMQPConn() (err error) {
  amql_conn, err = amqp.Dial(amqp_url)
  if err != nil {
    panic(err)
  }
  return err
}
​
func initConsumerTask() {
  consumer = make([]*ConsumerRoutine, max_count)
}
​
func init() {
  loadConf()
  initAMQPConn()
  initConsumerTask()
  go watchConsumerCount()
}

func startNewConsumer(m *ConsumerRoutine) {
  m.Exit = make(chan bool)
  m.Done = make(chan bool)
  m.Channel, _ = amql_conn.Channel()
  m.Channel.QueueDeclare(queue_name, true, false, false, false, nil)
  m.Delivery, _ = m.Channel.Consume(queue_name, "", false, false, false, false, nil)
  fmt.Println("Consumer : ", m.Id, "started")
  for {
    select {
    case <-m.Exit:       // Lắng nghe lệnh exit từ bên ngoài
      m.Channel.Close()  // Đóng amqp channel
      m.Channel = nil    
      m.Done <- true     // Báo cáo là sẵn sàng thoát
      fmt.Println("Consumer ", m.Id, " ended")
      return
    case msg := <-m.Delivery: // Chờ message từ queue
      // Xử lí message ở đây
      fmt.Println("Consumer ", m.Id, "recived message = ", string(msg.Body))
      // Ack hoặc Reject message, ở đây để luôn có data test nên mình reject luôn :pikatroll:
      msg.Reject(true)
    }
  }
}

func ResizeConsumerCount(n int) {
  // scale up
  // Chỗ này thì thêm phần từ vô slice và sau đó thì chạy một con
  // consumer và truyền biến vào để giao tiếp
  if n >= consumer_count {
    val := consumer_count
    consumer_count = n
    for i := val; i < n; i++ {
      consumer[i] = &ConsumerRoutine{
        Id:     i,
        Exit:   make(chan bool),
        Done:   make(chan bool),
      }
      go startNewConsumer(consumer[i])
    }
    return
  }
  // scale down
  // Giả sự số lượng hiện tại là consumer_count
  // và số consumer sau khi giảm là n thì mình sẽ 
  // gửi lệnh exit cho những con consumer từ n->val-1
  // vì slice được đếm từ 0
  val := consumer_count
  consumer_count = n
  for i := n; i < val; i++ {
    fmt.Println("shuting down consumer : ", i)
    go func(m *ConsumerRoutine) {
      fmt.Println("Exit Chan Pointer = ", &m.Exit)
      m.Exit <- true
    }(consumer[i])
    <-consumer[i].Done
  }
}

func main() {
  // Mình chạy consumer_count con consumer từ file config
  for i := 0; i < consumer_count; i++ {
    consumer[i] = &ConsumerRoutine{
      Id: i,
    }
    go startNewConsumer(consumer[i])
  }
  // khởi tạo 1 channel để lắng nghe signal 
  stop := make(chan os.Signal, 1)
  signal.Notify(stop)
  for {
    v := <-stop
    fmt.Println("SIGNAL : ", v.String())
    // nếu signal là hangup thì mình sẽ tắt hết consumer và thoát app
    if v.String() == "hangup" {
      ResizeConsumerCount(0)
      os.Exit(0)
    }
  }
}

SIGNAL :  urgent I/O condition
SIGNAL :  urgent I/O condition
SIGNAL :  urgent I/O condition
SIGNAL :  urgent I/O condition
Config file changed: /Users/vimftw/Desktop/code/resizeConsumer/config.yaml
resize consumer count to =>  4
SIGNAL :  urgent I/O condition
Consumer :  2 started Exit pointer val =  0xc00005a718
Consumer :  3 started Exit pointer val =  0xc00005a758
SIGNAL :  urgent I/O condition

Config file changed: /Users/vimftw/Desktop/code/resizeConsumer/config.yaml
resize consumer count to =>  1
shuting down consumer :  1
Exit Chan Pointer =  0xc00019a058
have to exit now
Consumer  1  ended
shuting down consumer :  2
Exit Chan Pointer =  0xc00005a718
have to exit now
Consumer  2  ended
shuting down consumer :  3
Exit Chan Pointer =  0xc00005a758
have to exit now
Consumer  3  ended
SIGNAL :  urgent I/O condition
SIGNAL :  urgent I/O condition