当nsq跑起来之后, 我们可能会遇到以下问题
- 分布式部署
- 处理错误(何时requeue)
- 如何使用golang lib
- 消息生命周期如何, 如重试/断线重连逻辑.
抱着不应该只停留在入门的态度, 笔者粗浅的研究了一下这几个问题, 希望也对有同样疑问的人有帮助.
分布式部署注意:
由于NSQ的分布式网络结构, 必须为每一个NSQD分配单独的地址(如IP或host)以保证消费者在lookup找到NSQD节点后能够正确的连接到对于的NSQD节点, 这也就意味着需要好好的规划NSQD的广播地址. (下文会说到在Rancher中如何配置NSQD的广播地址)
NSQ推荐的部署方式是 让NSQD与TOPIC生产者一一对应 既一个生产者一个NSQD.
register10.0.0.110.0.0.110.0.0.2详细可参考官方文档的拓扑图: topology_patterns
在Rancher中部署官方文档提供的分布式部署方式是多主机部署, 所以如何在Rancher中部署就只有自己实践了.
由于篇幅较多, 所以另起一篇.
NSQ Requeue And Backoff这两个概念与作用些许复杂, 建议结合官方文档来看
requeue(重试)
当错误发生, 需要重试时就应该使用nsq的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, 避免阻塞消息消费.
参考:
Max in Flightcnf := nsq.NewConfig()
cfg.MaxInFlight =200
MaxInFlight控制nsqd将多少消息同时发送给消费者, 默认是1, 意味着同时只有一个消息在被消费者处理, 如果你没有控制并发数量的需求, 建议设置为CPU的数量以提高性能.
golang libnsq提供golang的client lib. 支持全部特性.
本着不重复造轮子原则, 我也想尽大可能的使用nsq lib里的代码逻辑来实现需求, 但有些需求它实现不了:
- 自定义requeue的等待时间
- 判断某错误是否应该重试: 对于不应该重试的错误(如参数有误), 应该直接FIN, 而不是REQ.
我也只好自己写代码了.
先看看它原有的几个逻辑
// 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
}
消息自动重试(REQ)与完成(FIN):
func (r *Consumer) handlerLoop(handler Handler) {
r.log(LogLevelDebug, "starting Handler")
for {
message, ok := <-r.incomingMessages
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, 这种重试是很方便但是也有坏处.
使用这个重试机制的坏处是:
直接失败msg.Requeue(-1)msg.RequeueWithoutBackoff(-1)我的做法是再包裹一次Handler, 在闭包内部自定义错误逻辑. 我就不写上我乱糟糟的代码了, 您能实现得更简单.
nsqadminnsqadmin 提供一个web页面来管理nsq的消息/Topic/Channel.
Lookup
我们知道还没有生产者产生消息时(比如刚刚才部署), topic不存在, 这时如果有消费者连接上nsqlookup就会一直报错 topic not found, 为了避免这个报错, 就可以在Create Topic/Channel栏目中预先创建topic和channel.
test可以看到提示说这个topic当前不活跃, 也就是只在nsqloopup新建了topic但是没有在任何nsqd里生产. 这个提示在nsq开始发送第一个消息后消失.
如何保证消息被至少投递一次重试
在Handler中返回一个错误就会触发重试, 重试的消息被存储在nsq的Deferred队列, 一定延时后消费者会再次收到此消息.
断线恢复
FINFINFIN