在使用nsq消費(fèi)時(shí),發(fā)現(xiàn)如果HandleMessage方法執(zhí)行時(shí)間超過2分鐘,在此方法運(yùn)行完后,就會(huì) 拋出
IO error - EOF錯(cuò)誤。測試發(fā)現(xiàn),當(dāng)超過2分鐘時(shí),nsqadmin 后臺(tái)顯示的節(jié)點(diǎn)未連接。在查看github Issues時(shí),有人提出過,于是做了測試,發(fā)現(xiàn)一切正常。(在實(shí)際開發(fā)中,不建議把慢執(zhí)行放在消費(fèi)服務(wù)里,因?yàn)镠andleMessage方法不執(zhí)行完,是不會(huì)消費(fèi)下一個(gè)隊(duì)列消息的)
func (m *MyHandler) HandleMessage(msg *nsq.Message) (err error) {
// msg.NSQDAddress的地址是[--broadcast-address所指的ip] --broadcast-address=nsqd
fmt.Printf("%s recv from %v, msg:%v\n", m.Title, msg.NSQDAddress, string(msg.Body))
done := make(chan int)
go func() {
t := time.Tick(time.Second * 10)
for {
select {
case <-t:
fmt.Println("touch.")
msg.Touch()
case <-done:
return
}
}
}()
// 測試sleep500秒
time.Sleep(time.Second * 500)
done <- 1
/*//msg.DisableAutoResponse() // 禁用自動(dòng)提交
// err := 執(zhí)行業(yè)務(wù)邏輯
if err != nil {
//msg.Requeue(time.Second * 5) // 重新放入隊(duì)列消費(fèi)
} else {
//msg.Finish() // 手動(dòng)提交消費(fèi)完成,移出隊(duì)列
}*/
return
}