golang_go-micro整合各种mq

NATS

介绍

NATS是一个开源、轻量级、高性能的分布式消息中间件,实现了高可伸缩性和优雅的Publish/Subscribe模型,使用Golang语言开发。NATS的开发哲学认为高质量的QoS应该在客户端构建,故只建立了Request-Reply,不提供 1.持久化 2.事务处理 3.增强的交付模式 4.企业级队列。

NATS消息传递模型

NATS支持各种消息传递模型,包括:

发布订阅(Publish Subscribe)

请求回复(Request Reply)

队列订阅(Queue Subscribers )

提供的功能:

纯粹的发布订阅模型(Pure pub-sub)

服务器集群(Cluster mode server)

自动精简订阅者(Auto-pruning of subscribers)

基于文本协议(Text-based protocol)

多服务质量保证(Multiple qualities of service - QoS)

发布订阅(Publish Subscribe)

NATS将publish/subscribe消息分发模型实现为一对多通信,发布者在Subject上发送消息,并且监听该Subject在任何活动的订阅者都会收到该消息

整合go-micro

consumer:

package main
import (
 "fmt"
 "github.com/micro/go-micro/broker"
 "github.com/micro/go-plugins/broker/nats"
 nat "github.com/nats-io/nats.go"
 "log"
)
var (
 topic = "go.micro.topic.foo"
)
// Example of a shared subscription which receives a subset of messages
func sharedSub() {
 _, err := broker.Subscribe(topic, func(p broker.Event) error {
 fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
 return nil
 }, broker.Queue("consumer"))
 if err != nil {
 fmt.Println(err)
 }
}
// Example of a subscription which receives all the messages
func sub() {
 _, err := broker.Subscribe(topic, func(p broker.Event) error {
 fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
 return nil
 })
 if err != nil {
 fmt.Println(err)
 }
}
func main() {
 na := nats.NewBroker(nats.Options(nat.GetDefaultOptions()))
 if err := na.Init(); err != nil {
 log.Fatalf("Broker Init error: %v", err)
 }
 if err := na.Connect(); err != nil {
 log.Fatalf("Broker Connect error: %v", err)
 }
 sub()
 select {}
}

producer:

package main
import (
 "fmt"
 "log"
 "time"
 "github.com/micro/go-micro/broker"
 "github.com/micro/go-plugins/broker/nats"
 nat "github.com/nats-io/nats.go"
)
var (
 topic = "go.micro.topic.foo"
)
func pub() {
 tick := time.NewTicker(time.Second)
 i := 0
 for _ = range tick.C {
 msg := &broker.Message{
 Header: map[string]string{
 "id": fmt.Sprintf("%d", i),
 },
 Body: []byte(fmt.Sprintf("%d: %s", i, time.Now().String())),
 }
 if err := broker.Publish(topic, msg); err != nil {
 log.Printf("[pub] failed: %v", err)
 } else {
 fmt.Println("[pub] pubbed message:", string(msg.Body))
 }
 i++
 }
}
func main() {
 na := nats.NewBroker(nats.Options(nat.GetDefaultOptions()))
 if err := na.Init(); err != nil {
 log.Fatalf("Broker Init error: %v", err)
 }
 if err := na.Connect(); err != nil {
 log.Fatalf("Broker Connect error: %v", err)
 }
 pub()
}

NATS Streaming

介绍

NATS由于不能保证消息的投递正确性和存在其他的缺点,NATS Streaming就孕育而生.他是一个由NATS提供支持的数据流系统,采用Go语言编写,NATS Streaming与核心NATS平台无缝嵌入,扩展和互操作.除了核心NATS平台的功能外,他还提供了以下功能:

NATS Streaming特征

增强消息协议(Enhanced message protocol)

消息/事件持久化(Message/event persistence)

至少一次数据传输(At-least-once-delivery)

Publisher限速(Publisher rate limiting)

Subscriber速率匹配(Rate matching/limiting per subscriber)

按主题重发消息(Historical message replay by subject)

持续订阅(Durable subscriptions)

整合go-micro

consumer:

package main
import (
 "fmt"
 "github.com/micro/go-micro/broker"
 "github.com/micro/go-plugins/broker/stan"
 stancf "github.com/nats-io/stan.go"
 "log"
)
var (
 topic = "go.micro.topic.foo"
)
// Example of a shared subscription which receives a subset of messages
func sharedSub() {
 _, err := broker.Subscribe(topic, func(p broker.Event) error {
 fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
 return nil
 }, broker.Queue("consumer"))
 if err != nil {
 fmt.Println(err)
 }
}
// Example of a subscription which receives all the messages
func sub() {
 _, err := broker.Subscribe(topic, func(p broker.Event) error {
 fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
 return nil
 })
 if err != nil {
 fmt.Println(err)
 }
}
func main() {
 na := stan.NewBroker(stan.ClusterID("test-cluster"), stan.Options(stancf.GetDefaultOptions()))
 if err := na.Init(); err != nil {
 log.Fatalf("Broker Init error: %v", err)
 }
 if err := na.Connect(); err != nil {
 log.Fatalf("Broker Connect error: %v", err)
 }
 sub()
 select {}
}

producer:

package main
import (
 "fmt"
 "log"
 "time"
 "github.com/micro/go-micro/broker"
 "github.com/micro/go-plugins/broker/stan"
 stancf "github.com/nats-io/stan.go"
)
var (
 topic = "go.micro.topic.foo"
)
func pub() {
 tick := time.NewTicker(time.Second)
 i := 0
 for _ = range tick.C {
 msg := &broker.Message{
 Header: map[string]string{
 "id": fmt.Sprintf("%d", i),
 },
 Body: []byte(fmt.Sprintf("%d: %s", i, time.Now().String())),
 }
 if err := broker.Publish(topic, msg); err != nil {
 log.Printf("[pub] failed: %v", err)
 } else {
 fmt.Println("[pub] pubbed message:", string(msg.Body))
 }
 i++
 }
}
func main() {
 na := stan.NewBroker(stan.ClusterID("test-cluster"), stan.Options(stancf.GetDefaultOptions()))
 if err := na.Init(); err != nil {
 log.Fatalf("Broker Init error: %v", err)
 }
 if err := na.Connect(); err != nil {
 log.Fatalf("Broker Connect error: %v", err)
 }
 pub()
}

命令行选择mq

package main
import (
 "fmt"
 "log"
 "time"
 "github.com/micro/go-micro/broker"
 "github.com/micro/go-micro/config/cmd"
 _ "github.com/micro/go-plugins/broker/nats" //导入该包,
)
var (
 topic = "go.micro.topic.foo"
)
func pub() {
 tick := time.NewTicker(time.Second)
 i := 0
 for _ = range tick.C {
 msg := &broker.Message{
 Header: map[string]string{
 "id": fmt.Sprintf("%d", i),
 },
 Body: []byte(fmt.Sprintf("%d: %s", i, time.Now().String())),
 }
 if err := broker.Publish(topic, msg); err != nil {
 log.Printf("[pub] failed: %v", err)
 } else {
 fmt.Println("[pub] pubbed message:", string(msg.Body))
 }
 i++
 }
}
func sub() {
 _, err := broker.Subscribe(topic, func(p broker.Event) error {
 fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
 return nil
 })
 if err != nil {
 fmt.Println(err)
 }
}
func main() {
 cmd.Init()
 if err := broker.Init(); err != nil {
 log.Fatalf("Broker Init error: %v", err)
 }
 if err := broker.Connect(); err != nil {
 log.Fatalf("Broker Connect error: %v", err)
 }
 go pub()
 go sub()
 <-time.After(time.Second * 10)
}

启动时添加:--broker=启动的mq名称 --broker_address=ip地址

总结:

基本上只需要更改插件既可切换 mq,可以在github.com/micro/go-plugins/broker包下找到自己需要的mq,导入使用既可

"github.com/micro/go-plugins/broker/stan"

本文来自:Segmentfault

感谢作者:yylany

查看原文:golang_go-micro整合各种mq

• 编辑
• 预览