一 环境依赖:
golang 开发环境(version >= 1.2) 下源码,配置环境变量,执行安装脚本
gpm 依赖包管理器 ubantu: sudo apt-get intall gpm
二 NSQ安装:
- git获取源码: mkdir -p $GOPATH/src/github.com/nsqio;cd $GOPATH/src/github.com/nsqio;git clone https://github.com/nsqio/nsq.git;cd nsq
- 安装依赖包: gpm install
- 安装NSQ: go install ./...
三 开启NSQ:
- nsqd节点维护进程:nsqlookupd &
- nsqd节点进程:nsqd --lookupd-tcp-address=127.0.0.1:4160 &
- 消息产看进程:nsqadmin --lookupd-http-address=127.0.0.1:4161 &
ps:nsqlookupd与nsqadmin为辅助进程,可不使用直接用nsqd也可正常工作.
这里开启的进程均用默认的端口
四 工具测试:
- curl -d ‘hello world 1‘ ‘http://127.0.0.1:4151/put?topic=test‘ // 产生一个topic为“test” 消息内容为“hello world”的消息
- nsq_to_file --topic=test --output-dir=./tmp --lookupd-http-address=127.0.0.1:4161 // 将topic为“test”的消息写到./tmp目录下的一个文件中
五 代码测试:
1 package main 2 3 import ( 4 "fmt" 5 "time" 6 7 "github.com/nsqio/go-nsq" 8 ) 9 10 // nsq发布消息 11 func Producer() { 12 p, err := nsq.NewProducer("127.0.0.1:4150", nsq.NewConfig()) // 新建生产者 13 if err != nil { 14 panic(err) 15 } 16 17 if err := p.Publish("test", []byte("hello NSQ!!!")); err != nil { // 发布消息 18 panic(err) 19 } 20 } 21 22 // nsq订阅消息 23 type ConsumerT struct{} 24 25 func (*ConsumerT) HandleMessage(msg *nsq.Message) error { 26 fmt.Println(string(msg.Body)) 27 return nil 28 } 29 30 func Consumer() { 31 c, err := nsq.NewConsumer("test", "test-channel", nsq.NewConfig()) // 新建一个消费者 32 if err != nil { 33 panic(err) 34 } 35 c.AddHandler(&ConsumerT{}) // 添加消息处理 36 if err := c.ConnectToNSQD("127.0.0.1:4150"); err != nil { // 建立连接 37 panic(err) 38 } 39 } 40 // 主函数 41 func main() { 42 Producer() 43 Consumer() 44 time.Sleep(time.Second * 3) 45 } 46 // 运行将会打印: hello NSQ!!!
六 使用总结:
单机使用条件,同步发布消息速度也非常快(10w/s),发布消息端基本无需再做缓存封装。接收端的消息处理应耗时尽量的短,避免消息积累,当消息积累到NSQ的缓存的数量会将多余的消息写到文件,此时也会减缓发送端的发送速度,
因此,对接收端可使用channel和go routine做简单封装处理。若某topic消息产生太快太多也可将其单独使用一个nsqd处理,避免消息积累影响其它消息投递与接收。
时间: 2024-10-12 17:51:56