golang微服务框架go-micro 入门笔记2.3 micro工具之消息接收和发布

本章节阐述micro消息订阅和发布相关内容阅读本文前你可能需要进行如下知识储备golang分布式微服务框架go-micro 入门笔记1:搭建go-micro环境,golang微服务框架go-micro 入门笔记2.1 micro工具之micro apigolang微服务框架go-micro 入门笔记2.2 micro工具之micro webbroker代理微服务之间需要通过broker来传递消息,go-micro支持http/nats/memory三种broker,其中http是默认的broker。同时,go-micro以强大的插件形式,提供如下几种常见的broker。$ls gocloud/ googlepubsub/ grpc/ kafka/ mqtt/ nats/ nsq/ proxy/ rabbitmq/ redis/ snssqs/ sqs/ stan/ stomp/httpHTTP Broker 是基于HTTP的异步broker,源代码在github.commicrogo-microv1.9.1brokerbroker.go中,默认DefaultBroker为httpvar ( DefaultBroker Broker newHttpBroker() )httpbroker实际上就是一个结构体type httpBroker struct { id string //微服务ID address string //主机地址 opts Options //一些配置 mux *http.ServeMux //通过这个监听其他端发送的http请求 c *http.Client //通过这个发送请求到其他端 r registry.Registry sync.RWMutex subscribers map[string][]*httpSubscriber //订阅 running bool exit chan chan error // offline message inbox mtx sync.RWMutex inbox map[string][][]byte //数据缓存 }通过http.Client发送请求,通过http.ServeMux实现请求监听,通过inbox存储数据redisredis初始化代码如下//main.go //初始化URL格式redis://密码主机:端口/ b : redis.NewBroker( broker.Addrs(redis://user:secretlocalhost:6379/), ) //初始化 b.Init() //连接 b.Connect() // 新建service service : grpc.NewService( micro.Name(go.micro.web.config), micro.Version(latest), micro.Broker(b), ) //初始化service service.Init() //启动,运行,监听 service.Run()启动应用程序需要指定broker为redisgo run main.go --brokerredisgrpc 初始化初始化过程如下//main.go import ( github.com/micro/go-plugins/broker/grpc ) // 建立连接 b : grpc.NewBroker() b.Init() b.Connect() // 订阅事件 sub, _ : b.Subscribe(events) defer sub.Unsubscribe() // 发布事件 b.Publish(events, broker.Message{ Headers: map[string]string{type: event}, Body: []byte(an event), })启动应用程序需要指定broker为grpcgo run main.go --brokergrpcrabbitmq 初始化初始化过程如下//main.go import ( github.com/micro/go-plugins/broker/grpc ) b : rabbitmq.NewBroker( broker.Addrs(amqp://用户名:密码主机host:端口port), ) b.Init() b.Connect()启动应用程序需要指定broker为rabbitmqgo run main.go plugin.go --brokerrabbitmqmqtt初始化过程如下//main.go import ( github.com/micro/go-micro github.com/micro/go-plugins/broker/mqtt ) func main() { service : micro.NewService( micro.Name(my.service), micro.Broker(mqtt.NewBroker()), ) //... }启动应用程序需要指定broker为mqttgo run main.go plugin.go --brokermqtt其他其他可以阅读代码$GOPATH/src/github.com/micro/go-plugins/broker消息订阅和发布通过micro.RegisterSubscriber实现消息订阅消息订阅主要API接口如下,第一个参数标识消息主题,第二个参数表示服务实例。// Register Struct as Subscriber micro.RegisterSubscriber(go.micro.srv.testsrv, service.Server(), new(subscriber.Testsrv)) // Register Function as Subscriber micro.RegisterSubscriber(go.micro.srv.testsrv, service.Server(), subscriber.Handler)重点注意第三个参数,第三个参数是处理函数,可以是函数,也可以是实现了func Handler(ctx context.Context, msg *testsrv.Message) error方法的结构体,micro内部会根据参数类型自动适配。结构体中可以实现多个func Handler(ctx context.Context, msg *testsrv.Message) error类型方法通过broker.Subscribe实现订阅Broker提供如下接口type Broker interface { Init(...Option) error Options() Options Address() string Connect() error Disconnect() error Publish(topic string, m *Message, opts ...PublishOption) error Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error) String() string }Subscribe 订阅事件,topic代表主题,h事件处理函数Publish 发布事件消息处理函数Handler 定义在上述涉及到处理函数handler,具体含义如下type Handler func(Event) error // Event is given to a subscription handler for processing type Event interface { Topic() string Message() *Message Ack() error } type Message struct { Header map[string]string Body []byte }通过broker.Publish实现发布举例如下// 建立连接 b : grpc.NewBroker() b.Init() b.Connect() // 订阅事件 sub, _ : b.Subscribe(events) defer sub.Unsubscribe() // 发布事件 b.Publish(events, broker.Message{ Headers: map[string]string{type: event}, Body: []byte(an event), })通过micro publish实现发布举例如下micro publish go.micro.web.config hello实战和代码效果下载代码broker.zip解压到techidea8.com/microapp/broker下运行,效果图忑发布消息需要注意json格式字符串micro publish go.micro.srv.broker {\say\:\这是测试消息\}获得代码关注公众号回复micro-broker即可获得