# [xutils/mq](https://github.com/xaces/xutils/tree/master/mq)
## 功能
- 统一接口，初始化时选择不同的协议适配器，
- 单独使用指定协议
- 创建指定个数的连接，推送、订阅时动态均衡分配
## 测试代码
```go
func TestMqtt(t *testing.T) {
c, err := NewPublish(&Options{Address: "127.0.0.1:35003", Goc: 5}, NewMqtt)
if err != nil {
c.Shutdown()
log.Fatalln(err)
}
topic := "test/mqtt"
c.Subscribe(topic, func(b []byte) error {
log.Printf("%s: %s\n", topic, b)
return nil
})
for {
time.Sleep(2 * time.Second)
c.Publish(topic, map[string]interface{}{
"device": "20198002",
"now": time.Now().Format("2006-01-02 15:04:05"),
})
}
}
func TestStomp(t *testing.T) {
c, err := NewPublish(&Options{Address: "127.0.0.1:35002", Goc: 1}, NewStomp)
if err != nil {
c.Shutdown()
log.Fatalln(err)
}
subject := "/queue/test/stomp"
c.Subscribe(subject, func(b []byte) error {
log.Printf("%s:1 %s\n", subject, b)
return nil
})
c.Subscribe(subject, func(b []byte) error {
log.Printf("%s:2 %s\n", subject, b)
return nil
})
for {
time.Sleep(2 * time.Second)
c.Publish(subject, map[string]interface{}{
"device": "20198002",
"now": time.Now().Format("2006-01-02 15:04:05"),
})
}
}
func TestNats(t *testing.T) {
c, err := NewPublish(&Options{Address: NatsURL, Goc: 1}, NewStomp)
if err != nil {
c.Shutdown()
log.Fatalln(err)
}
subject := "/queue/test/nats"
c.Subscribe(subject, func(b []byte) error {
log.Printf("%s:1 %s\n", subject, b)
return nil
})
c.Subscribe(subject, func(b []byte) error {
log.Printf("%s:2 %s\n", subject, b)
return nil
})
for {
time.Sleep(2 * time.Second)
c.Publish(subject, map[string]interface{}{
"device": "20198002",
"now": time.Now().Format("2006-01-02 15:04:05"),
})
}
}
```
有疑问加站长微信联系（非本文作者））
