使用场景

我这儿用一个实例来说明nsq的使用，假设我有一个apiServer的程序，它负责面向用户。有一个dataServer的程序，负责处理数据。apiServer这个进程对用户的请求进行简单的处理，复杂的资源处理则去请求dataSever，让dataServer来处理。
dataServer部署在多台主机，apiServer需要随机的向一个dataServer发送请求，来实现负载均衡。所有我们需要在apiServer的配置文件中写入所有dataServer的服务器地址和端口。
如果这个时候有一个dataServer挂掉了怎么办呢？apiServer还是会随机的请求到它，这就会造成用户请求失败。
所以我们就需要用到nsq消息队列了，而不是将dataServer的所有地址写到配置中。每个dataServer进程都需要定时的向apiServer发送消息，告诉自身的地址，apiServer将地址进行筛选，然后就可以用这些地址来请求dataServer了。

程序演示：

dataServer（生产者）:

func HeartBeat(){
    var(
        producer *nsq.Producer
        err error
        ticker *time.Ticker
    )
    ticker = time.NewTicker(time.Duration(conf.HEART_BEAT_TIME)*time.Second)  //创建一个定时器，这里的时间我都写到配置文件里了，然后用conf包拿出来。我这里是设置的5秒
    for {
        select {
        case <-ticker.C:
            //创建一个生产者，这里的RanDomGetServer()是自定义的一个工具，用来随机获取一个nsqd地址
            if producer, err = nsq.NewProducer(utils.RadomGetNsqd(conf.NSQ_TCP_ADDRS), nsq.NewConfig()); err != nil {
                panic(err)
            }
            //pulish()接的第一个参数是topic，这个也是自己定义(值为data_server_addr），第二个参数是要发送的消息，这里是本机服务器地址
            err = producer.Publish(conf.DATA_SERVER_TOPIC, []byte(conf.SERVER_ADDR))
            if err != nil {
                panic(err)
            }
        }
    }
}

utils下的RanDomGetNsq工具：

//从nsqd集群地址中随机获取一个nsqd地址用于publish,nsqd集群地址是事先写在配置文件中的
func RadomGetNsqd([]string)string{
    rand.Seed(time.Now().UnixNano())
    return conf.NSQ_TCP_ADDRS[rand.Intn(len(conf.NSQ_TCP_ADDRS))]
}

main函数：

var confPath string

func init(){
    flag.StringVar(&confPath,"c","","config file")
    runtime.GOMAXPROCS(4)
}

func main(){
    flag.Parse()
    conf.Init(confPath)  //加载配置
    gin.SetMode(gin.DebugMode)
    r := gin.New()
    router.Load(r)  //加载路由和中间件
    server := http.Server{
        Addr:conf.SERVER_ADDR,  
        ReadTimeout:time.Duration(conf.SERVER_READ_TIME_OUT)*time.Second,
        WriteTimeout:time.Duration(conf.SERVER_READ_TIME_OUT)*time.Second,
        Handler:r,
    }
    go middleWare.HeartBeat()   //这里开goroutine来发送心跳包
    if err := server.ListenAndServe();err != nil{
        panic(err)
    }
}

这样我们就实现了每五秒发送一次本机服务器地址给apiServer进程，apiServer创建生产者就能接受到消息了。

apiServer(消费者):

消费者的创建比生产者的创建要麻烦些。
首先是自定义的工具utils.NewConsumer来创建一个消费者：

//这里第一个参数是需要绑定topic(data_server_addr),
//第二个参数传入一个string，这就是创建的ch，topic消息队列中的消息会分发到每个ch中.
//每个消费者可以创建不通的ch，也可以共用一个ch，共用一个ch, ch的消息会随机发送给其中一个消费者
//第三个参数是处理message的nsq.handler接口，需要实现一个HanddleMessage(*nsq.Message)error()方法。
func NewConsumer(topic string,chanName string,h nsq.Handler)(consumer *nsq.Consumer,err error){
    if consumer,err = nsq.NewConsumer(topic,chanName,nsq.NewConfig());err != nil{
        return nil,err
    }
    consumer.ChangeMaxInFlight(3)   //可以根据nsqds数量来配置
    consumer.AddHandler(h)
    err = consumer.ConnectToNSQLookupd(conf.NSQ_LOOKUPD_ADDR)     //todo:读取配置地址
    if err != nil{
        return nil,err
    }
    return consumer,nil
}

接下来正式处理消息了：

package consumers
import("......")

//创建一个nsq.handler接口实例
var ServerConsumerHandler = &ServerConsumer{DataServerAddrs:make(map[string]time.Time)}

//ServerConsumer实现了nsq.handler接口
type ServerConsumer struct {
    DataServerAddrs  map[string]time.Time  //保存dataServer进程发过来的服务器地址和接收时间
    rwLocker sync.RWMutex  //防止多线程同时读写
}

//HandleMessage是nsq,handler接口定义方法，必须实现，主要是用来处理传递过来的消息
func(h *ServerConsumer)HandleMessage(message *nsq.Message)error{
    if dataServer := string(message.Body);dataServer != ""{
        h.rwLocker.Lock()
        h.DataServerAddrs[dataServer] = time.Now()
        h.rwLocker.Unlock()
    }
    return nil
}

//监听服务data_server_addr这个消息队列
func MonitorDataServerAddrs(){
    consumer ,err:= utils.NewConsumer("data_server_addr","ch1",ServerConsumerHandler)
    if err != nil{
        panic(err)
    }
    //连接到NSQLookupd，它底层会创建连接到每个nsqd.这样就可以监听每个nsqd中的消息
    err = consumer.ConnectToNSQLookupd(conf.NSQ_LOOKUPD_ADDR)   //方法本身会开一个goroutine来检查消息队列
    if err != nil{
        panic(err)
    }
}

//删除过期的服务器地址
func(h *ServerConsumer)removeExpireDatasServers(){
    for  {
        h.rwLocker.Lock()
        for dataServer,t := range h.DataServerAddrs{
            //只保存10秒之内发送过来的服务器地址
            if t.Add(10*time.Second).Before(time.Now()){
                delete(h.DataServerAddrs,dataServer)
            }
        }
        h.rwLocker.Unlock()
        time.Sleep(2*time.Second)
    }
}

main函数：

flag.Parse()
    conf.Init(confPath)  //加载配置
    gin.SetMode(gin.DebugMode)
    r := gin.New()
    router.Load(r)
    server := http.Server{
        Addr:"192.168.50.250:8080",
        ReadTimeout:time.Duration(conf.SERVER_READ_TIME_OUT)*time.Second,
        WriteTimeout:time.Duration(conf.SERVER_READ_TIME_OUT)*time.Second,
        Handler:r,
    }
    consumers.MonitorDataServerAddrs()   //不用单独开goroutine
    if err := server.ListenAndServe();err != nil{
        panic(err)
    }

这样我们就实现了接收服务器地址，并且将过期的地址删掉，更大程度的保证保存的dataServer服务器地址是可用的。

通过nsqadmin来查看消息队列情况：

image.png
image.png
image.png