當nsq跑起來之后, 我們可能會遇到以下問題

分布式部署

處理錯誤(何時requeue)

如何使用golang lib

抱著不應該只停留在入門的態度, 筆者粗淺的研究了一下這幾個問題, 希望也對有同樣疑問的人有幫助.

部署

由于NSQ的分布式網絡結構, NSQD必須廣播(到lookup)自己的地址并讓消費者連接, 那么多個NSQD無法做透明負載均衡.

就必須為每一個NSQD分配單獨的IP(或host)以保證消費者(在lookup找到NSQ節點)能夠正確的連接. 這樣部署起來可能會麻煩一些 但問題不大, 注意一下就好.

ps: 如果有更好的方法請告訴我, 小弟感激不盡.

NSQ Requeue And Backoff

建議結合官方文檔來看

requeue(重試)

用于當錯誤發生, 需要重試時.

backoff(避退)

backoff能降低消費者吞吐量以讓消費者從錯誤中恢復.

當消費者在backoff狀態時, 這個消費者將不再處理任何消息, 直到backoff超時

當觸發backoff時控制臺將打印:

// 進入backoff狀態, RDY設置為0代表準備接收0條消息(不接收消息) (協議詳情看 https://nsq.io/clients/tcp_protocol_spec.html)

WRN 1 [test/test] backing off for 1m4s (backoff level 6), setting all to RDY 0

// 時間到了將設置RDY為1接收1條消息以測試狀態, 官方將這個狀態稱為`tests the waters`

WRN 1 [test/test] (DESKTOP-HELJ7V4:4150) backoff timeout expired, sending RDY 1

當有多個消費者競爭時, 出錯的消費者應當主動backoff不再處理消息(以讓出更多的機會給其他消費者).

如果只有一個消費者, 則消費者會等到backoff超時后才開始處理消息(空出時間讓消費者恢復).

避退是存在于整個消費者上的, 所以消費者每當一個消息處理失敗了之后都會增加這個消費者的backoff level. 這會影響這個消費者的處理能力.

到底需不需要用backoff, 就要看業務了:

消息是用來更新數據庫訂單狀態的, 這是一個不容易出錯的邏輯, 如果需要requeue則需要backoff讓出優先級, 讓其他消費者來做, 盡量以挽救這個訂單.

消息是用來通知第三方(如支付寶支付成功的http回調)的, 一般requeue是發生在第三方端響應不滿足預期的響應, 這不是我方消費者的錯誤, 應當不使用backoff, 避免阻塞消息消費.

參考:

golang lib

nsq提供golang的client lib. 支持全部特性.

本著不重復造輪子原則, 我也想盡大可能的使用nsq lib里的代碼邏輯來實現需求, 但有些需求它實現不了, 我也只好自己寫代碼了.

先看看它原有的幾個邏輯

消息自動重試

// Handler is the message processing interface for Consumer

//

// Implement this interface for handlers that return whether or not message

// processing completed successfully.

//

// When the return value is nil Consumer will automatically handle FINishing.

//

// When the returned value is non-nil Consumer will automatically handle REQueing.

type Handler interface {

HandleMessage(message *Message) error

}

消息自動重試與判斷失敗

func (r *Consumer) handlerLoop(handler Handler) {

r.log(LogLevelDebug, "starting Handler")

for {

message, ok :=

if !ok {

goto exit

}

if r.shouldFailMessage(message, handler) {

message.Finish()

continue

}

err := handler.HandleMessage(message)

if err != nil {

r.log(LogLevelError, "Handler returned error (%s) for msg %s", err, message.ID)

if !message.IsAutoResponseDisabled() {

message.Requeue(-1)

}

continue

}

if !message.IsAutoResponseDisabled() {

message.Finish()

}

}

exit:

r.log(LogLevelDebug, "stopping Handler")

if atomic.AddInt32(&r.runningHandlers, -1) == 0 {

r.exit()

}

}

判斷失敗

func (r *Consumer) shouldFailMessage(message *Message, handler interface{}) bool {

// message passed the max number of attempts

if r.config.MaxAttempts > 0 && message.Attempts > r.config.MaxAttempts {

r.log(LogLevelWarning, "msg %s attempted %d times, giving up",

message.ID, message.Attempts)

logger, ok := handler.(FailedMessageLogger)

if ok {

logger.LogFailedMessage(message)

}

return true

}

return false

}

requeue

可以看到當handler返回的error不為空時, nsq將自動requeue, 這種重試是很方便但是

使用這個重試機制的壞處是:

不能自定義requeue的等待時間(默認等待時間=config.DefaultRequeueDelay*Attempts)

會在控制臺打印一個ERR(不能自定義格式, 而且有一些err不應該打印到控制臺), 這點可能有潔癖的開發者受不了.

一些錯誤不應該重試, 如入參不合法, 再怎么重試也是徒勞. 這時候應該直接失敗.

所以我建議不要使用這個err機制, 而應當手動使用msg.Requeue(-1)或者msg.RequeueWithoutBackoff(-1) 來顯示指定requeue.

shouldFailMessage

我們可以使用 FailedMessageLogger interface自定義當消息失敗時的處理方式.

但它的shouldFailMessage又有什么需求滿足不了呢?

在失敗的時候拿到最后一次錯誤信息

shouldFailMessage只能判斷處理重試次數過多的失敗, 不能處理直接失敗的消息.

所以又只有自己實現啦:

我們直接在Handler中判斷Attempts來實現錯誤處理.

但為了保證我們的消息不被shouldFailMessage處理, 需要配置MaxAttempts為0或者一個比較大的數.