nsq

nsq

为什么要使用Nsq

最近一直在寻找一个高性能,高可用的消息队列做内部服务之间的通讯。一开始想到用zeromq,但在查找资料的过程中,意外的发现了Nsq这个由golang开发的消息队列,毕竟是golang原汁原味的东西,功能齐全,关键是性能还不错。其中支持动态拓展,消除单点故障等特性, 都可以很好的满足我的需求

Nsq服务端简介

在使用Nsq服务之前,还是有必要了解一下Nsq的几个核心组件

整个Nsq服务包含三个主要部分

nsqlookupd

先看看官方的原话是怎么说:

nsqlookupd 是守护进程负责管理拓扑信息。客户端通过查询 nsqlookupd 来发现指定话题(topic)的生产者,并且 nsqd 节点广播话题(topic)和通道(channel)信息

简单的说nsqlookupd就是中心管理服务,它使用tcp(默认端口4160)管理nsqd服务,使用http(默认端口4161)管理nsqadmin服务。同时为客户端提供查询功能
总的来说,nsqlookupd具有以下功能或特性

唯一性,在一个Nsq服务中只有一个nsqlookupd服务。当然也可以在集群中部署多个nsqlookupd,但它们之间是没有关联的

去中心化,即使nsqlookupd崩溃,也会不影响正在运行的nsqd服务

充当nsqd和naqadmin信息交互的中间件

提供一个http查询服务,给客户端定时更新nsqd的地址目录

nsqadmin

官方原话:是一套 WEB UI,用来汇集集群的实时统计,并执行不同的管理任务

总的来说,nsqadmin具有以下功能或特性

提供一个对topic和channel统一管理的操作界面以及各种实时监控数据的展示,界面设计的很简洁,操作也很简单

展示所有message的数量

能够在后台创建topic和channel,这个应该不常用到

nsqadmin的所有功能都必须依赖于nsqlookupd,nsqadmin只是向nsqlookupd传递用户操作并展示来自nsqlookupd的数据

nsqadmin默认的访问地址是http://127.0.0.1:4171/

nsqd

官方原话:nsqd 是一个守护进程,负责接收,排队,投递消息给客户端

简单的说,真正干活的就是这个服务,它主要负责message的收发,队列的维护。nsqd会默认监听一个tcp端口(4150)和一个http端口(4151)以及一个可选的https端口

总的来说,nsqd 具有以下功能或特性

对订阅了同一个topic,同一个channel的消费者使用负载均衡策略(不是轮询)

只要channel存在,即使没有该channel的消费者,也会将生产者的message缓存到队列中(注意消息的过期处理)

保证队列中的message至少会被消费一次,即使nsqd退出,也会将队列中的消息暂存磁盘上(结束进程等意外情况除外)

限定内存占用,能够配置nsqd中每个channel队列在内存中缓存的message数量,一旦超出,message将被缓存到磁盘中

topic,channel一旦建立,将会一直存在,要及时在管理台或者用代码清除无效的topic和channel,避免资源的浪费

这里想到一个问题是,如果一个channel只有生产者不停的在投递message,会不会导致服务器资源被耗尽?也许nsqd内部做了相应处理,但还是要避免这种情况的出现

Nsq客户端

了解nsqlookupd,nsqd与客户端中消费者和生产者的关系

消费者

消费者有两种方式与nsqd建立连接

消费者直连nsqd,这是最简单的方式,缺点是nsqd服务无法实现动态伸缩了(当然,自己去实现一个也是可以的)

消费者通过http查询nsqlookupd获取该nsqlookupd上所有nsqd的连接地址,然后再分别和这些nsqd建立连接(官方推荐的做法),但是客户端会不停的向nsqlookupd查询最新的nsqd地址目录(不喜欢用http轮询这种方式…)

生产者

生产者必须直连nsqd去投递message(网上说,可以连接到nsqlookupd,让nsqlookupd自动选择一个nsqd去完成投递,但是我用Producer的tcp是连不上nsqlookupd的,不知道http可不可以…),

这里有一个问题就是如果生产者所连接的nsqd炸了,那么message就会投递失败,所以在客户端必须自己实现相应的备用方案

安装Nsq

运行Nsq

运行Nsq服务集群

首先启动nsqlookud

nsqlookupd

启动nsqd,并接入刚刚启动的nsqlookud。这里为了方便接下来的测试,启动了两个nsqd

nsqd --lookupd-tcp-address=127.0.0.1:4160
nsqd --lookupd-tcp-address=127.0.0.1:4160 -tcp-address=0.0.0.0:4152 -http-address=0.0.0.0:4153

启动nqsadmin

nsqd --lookupd-http-address=127.0.0.1:4161

基于go-nsq的客户端实现

注意

  • Producer断线后不会重连,需要自己手动重连,Consumer断线后会自动重连
  • Consumer的重连时间配置项有两个功能
    • Consumer检测到与nsqd的连接断开后,每隔x秒向nsqd请求重连
    • Consumer每隔x秒,向nsqlookud进行http轮询,用来更新自己的nsqd地址目录
    • Consumer的重连时间默认是60s
  • Consumer可以同时接收不同nsqd node的同名topic数据,为了避免混淆,就必须在客户端进行处理
  • 在AddConurrentHandlers和 AddHandler中设置的接口回调是在另外的goroutine中执行的
  • Producer不能发布(Publish)空message,否则会导致panic
//Nsq发送测试
package main

import (
    "bufio"
    "fmt"
    "github.com/nsqio/go-nsq"
    "os"
)

var producer *nsq.Producer

// 主函数
func main() {
    strIP1 := "127.0.0.1:4150"
    strIP2 := "127.0.0.1:4152"
    InitProducer(strIP1)

    running := true

    //读取控制台输入
    reader := bufio.NewReader(os.Stdin)
    for running {
        data, _, _ := reader.ReadLine()
        command := string(data)
        if command == "stop" {
            running = false
        }

        for err := Publish("test", command); err != nil; err = Publish("test", command) {
            //切换IP重连
            strIP1, strIP2 = strIP2, strIP1
            InitProducer(strIP1)
        }
    }
}

// 初始化生产者
func InitProducer(str string) {
    var err error
    fmt.Println("address: ", str)
    producer, err = nsq.NewProducer(str, nsq.NewConfig())
    if err != nil {
        panic(err)
    }
}

//发布消息
func Publish(topic string, message string) error {
    var err error
    if producer != nil {
        if message == "" { //不能发布空串,否则会导致error
            return nil
        }

        err = producer.Publish(topic, []byte(message)) // 发布消息
        return err
    }
    return fmt.Errorf("producer is nil", err)
}
//Nsq接收测试
package main

import (
    "fmt"
    "time"
    "github.com/nsqio/go-nsq"
)

// 消费者
type ConsumerT struct{}

// 主函数
func main() {
    InitConsumer("test", "test-channel", "127.0.0.1:4161")
    for {
        time.Sleep(time.Second * 10)
    }
}

//处理消息
func (*ConsumerT) HandleMessage(msg *nsq.Message) error {
    fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
    return nil
}

//初始化消费者
func InitConsumer(topic string, channel string, address string) {
    cfg := nsq.NewConfig()
    cfg.LookupdPollInterval = time.Second          //设置重连时间
    c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一个消费者
    if err != nil {
        panic(err)
    }

    c.SetLogger(nil, 0)        //屏蔽系统日志
    c.AddHandler(&ConsumerT{}) // 添加消费者接口

    //建立NSQLookupd连接
    if err := c.ConnectToNSQLookupd(address); err != nil {
        panic(err)
    }

    //建立多个nsqd连接
    // if err := c.ConnectToNSQDs([]string{"127.0.0.1:4150", "127.0.0.1:4152"}); err != nil {
    //  panic(err)
    // }

    // 建立一个nsqd连接
    // if err := c.ConnectToNSQD("127.0.0.1:4150"); err != nil {
    //  panic(err)
    // }
}

参考资料

https://www.cnblogs.com/swarmbees/p/6635467.html
http://feixiao.github.io/2015/09/10/nsq_design/

发表评论