nsq消息队列php,NSQ学习:实现有序的消息队列
NSQ是一个内存+磁盘型的消息中间件，它使用push流的方式源源不断把消息推送给客户端，并且为了使服务端更加简单、高效，NSQ并不提供有序的消息队列。因此，如果对消息有顺序要求，只有两种解决办法：
改用类似kafka之类的有序消息队列；
生产者和消费者达成一个协议，比如增加一个序列号或者时间戳来表示顺序。
本文要介绍的是第二种方法，下面就来简单用golang实现一个有序的NSQ顺序消息队列。
首先，使用go-nsq客户端写一个main函数，并且连接上nsqlookupd：
func main() {
cfg := nsq.NewConfig()
cfg.LookupdPollInterval = time.Second
customer, err := nsq.NewConsumer("test2", "t1", cfg)
if err != nil {
log.Panic(err)
}
customer.AddHandler(&Customter{})
customer.SetLogger(nil, 0)
if err := customer.ConnectToNSQLookupd("127.0.0.1:4161"); err != nil {
log.Panic(err)
}
select {}
}
这样就表示这个消费者客户端订阅了test2这个topic的t1channel了。
接下来，我们与生产者端协商消息的格式是一个JSON文本字符串，并且这个JSON会传递一个时间戳用来表示消息的先后顺序，那么我们可以定义一个实现了sort.Interface的切片：
type MyMessage struct {
Name string `json:"name"`
CreateAt int64 `json:"create_at"`
}
type MyMessageList []MyMessage
func (list MyMessageList) Len() int {
return len(list)
}
func (list MyMessageList) Less(i, j int) bool {
return list[i].CreateAt < list[j].CreateAt
}
func (list MyMessageList) Swap(i, j int) {
list[i], list[j] = list[j], list[i]
}
sort.Interface有三个方法，分别是Len、Less、Swap。当实现了这三个方法，MyMessageList这个切片就能够把里面的MyMessage按照createAt字段的升序来进行排序。
最后，我们定义一个Customer结构体的HandleMessage方法来接收NSQ生产者的消息：
type Customter struct{}
// 存放消息结构体
var messageBuffer []MyMessage
func (c *Customter) HandleMessage(nsqMsg *nsq.Message) error {
var msg MyMessage
err := json.Unmarshal(nsqMsg.Body, &msg)
if err != nil {
log.Panicln(err)
}
messageBuffer = append(messageBuffer, msg)
if len(messageBuffer) == 3 {
sort.Sort(MyMessageList(messageBuffer))
// do something for the ordered message buffer
fmt.Println(messageBuffer)
// reset buffer
messageBuffer = messageBuffer[:0]
}
return nil
}
上面代码非常简单，每次收到一条消息，我们都会先把它放到一个buffer里面，然后当buffer的长度达到3的时候，就调用sort.Sort来进行一次排序，最后就能顺利拿到3个有序的消息，就可以做相应的操作了。
我们来简单实验一下，首先启动客户端(nsqd和nsqlookupd的启动参考官网文档，这里不介绍)：
$ ~/code/golang/src/nsq-test > go run main.go
然后在终端通过curl给nsqd的test2 topic生产3条数据：
curl -d '{"name":"aaa","create_at":1560262051}' http://127.0.0.1:4151/pub\?topic\=test2
curl -d '{"name":"bbb","create_at":1560262060}' http://127.0.0.1:4151/pub\?topic\=test2
curl -d '{"name":"ccc","create_at":1560262053}' http://127.0.0.1:4151/pub\?topic\=test2
注意，这里按照时间戳先后应该是aaa、ccc、bbb。
最后，我们回到启动客户端的终端查看，会发现打印出正确的结果：
# aaa ccc bbb 顺序是正确的
[{aaa 1560262051} {ccc 1560262053} {bbb 1560262060}]
到此就大功告成了~
完整代码在此：
package main
import (
"encoding/json"
"fmt"
"github.com/nsqio/go-nsq"
"log"
"sort"
"time"
)
type MyMessage struct {
Name string `json:"name"`
CreateAt int64 `json:"create_at"`
}
type MyMessageList []MyMessage
func (list MyMessageList) Len() int {
return len(list)
}
func (list MyMessageList) Less(i, j int) bool {
return list[i].CreateAt < list[j].CreateAt
}
func (list MyMessageList) Swap(i, j int) {
list[i], list[j] = list[j], list[i]
}
type Customter struct{}
var messageBuffer []MyMessage
func (c *Customter) HandleMessage(nsqMsg *nsq.Message) error {
var msg MyMessage
err := json.Unmarshal(nsqMsg.Body, &msg)
if err != nil {
log.Panicln(err)
}
messageBuffer = append(messageBuffer, msg)
if len(messageBuffer) == 3 {
sort.Sort(MyMessageList(messageBuffer))
// do something for the ordered message buffer
fmt.Println(messageBuffer)
// reset buffer
messageBuffer = messageBuffer[:0]
}
return nil
}
func main() {
cfg := nsq.NewConfig()
cfg.LookupdPollInterval = time.Second
customer, err := nsq.NewConsumer("test2", "t1", cfg)
if err != nil {
log.Panic(err)
}
customer.AddHandler(&Customter{})
customer.SetLogger(nil, 0)
if err := customer.ConnectToNSQLookupd("127.0.0.1:4161"); err != nil {
log.Panic(err)
}
select {}
}