golang使用Nsq
为什么要使用Nsq最近一直在寻找一个高性能，高可用的消息队列做内部服务之间的通讯。一开始想到用zeromq，但在查找资料的过程中，意外的发现了Nsq这个由golang开发的消息队列，毕竟是golang原汁原味的东西，功能齐全，关键是性能还不错。其中支持动态拓展，消除单点故障等特性， 都可以很好的满足我的需求 下面上一张Nsq与其他mq的对比图，看上去的确强大。下面简单记录一下Nsq的使用方法 Nsq服务端Nsq服务端简介 在使用Nsq服务之前，还是有必要了解一下Nsq的几个核心组件 nsqlookupd 先看看官方的原话是怎么说： 简单的说nsqlookupd就是中心管理服务，它使用tcp(默认端口4160)管理nsqd服务，使用http(默认端口4161)管理nsqadmin服务。同时为客户端提供查询功能 总的来说，nsqlookupd具有以下功能或特性
nsqadmin官方原话：是一套 WEB UI，用来汇集集群的实时统计，并执行不同的管理任务 总的来说，nsqadmin具有以下功能或特性
nsqadmin默认的访问地址是http://127.0.0.1:4171/ nsqd官方原话：nsqd 是一个守护进程，负责接收，排队，投递消息给客户端 简单的说，真正干活的就是这个服务，它主要负责message的收发，队列的维护。nsqd会默认监听一个tcp端口(4150)和一个http端口(4151)以及一个可选的https端口 总的来说，nsqd具有以下功能或特性
这是官方的图，第一个channel(meteics)因为有多个消费者，所以触发了负载均衡机制。后面两个channel由于没有消费者，所有的message均会被缓存在相应的队列里，直到消费者出现 这里想到一个问题是，如果一个channel只有生产者不停的在投递message，会不会导致服务器资源被耗尽？也许nsqd内部做了相应处理，但还是要避免这种情况的出现 Nsq服务端与客户端的关系了解nsqlookupd，nsqd与客户端中消费者和生产者的关系 消费者消费者有两种方式与nsqd建立连接
生产者生产者必须直连nsqd去投递message(网上说，可以连接到nsqlookupd，让nsqlookupd自动选择一个nsqd去完成投递，但是我用Producer的tcp是连不上nsqlookupd的，不知道http可不可以...)， 这里有一个问题就是如果生产者所连接的nsqd炸了，那么message就会投递失败，所以在客户端必须自己实现相应的备用方案 安装Nsq方法一
go get github.com/tools/godep 执行完后检查godep是否已经安装在bin目录下，一般都会自动安装，如果没有，用go install手动安装下
go get github.com/bmizerany/assert
godep get github.com/bitly/nsq/... 如果安装成功，bin目录里就会出现一大堆nsq_...开头的可执行文件
方法二
运行Nsq运行单机nsqd服务nsqd是一个独立的服务，启动一个nsqd就可以完成message的收发，启动一个单机的nsqd，很简单 nsqd 客户端可以使用http，也可以使用tcp，这里我使用是官方的go-nsq包做客户端，使用tcp进行message的收发
运行Nsq服务集群
nsqlookupd
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
nsqd --lookupd-http-address=127.0.0.1:4161 基于go-nsq的客户端实现几个值得注意的地方
go_nsq-send.go//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)
}
}
//关闭
producer.Stop()
}
// 初始化生产者
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)
}
go_nsq-receive.go//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)
// }
} （编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |