golang mqtt/stomp/nats/amqp

DonWang · · 619 次点击 · · 开始浏览    

# [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"), }) } } ```

有疑问加站长微信联系(非本文作者))

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

619 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传