不忘初心，砥砺前行
作者 | 陌无崖
转载请联系授权
导语
API模式笔记告一段了，这篇文章总结了Micro中的消息订阅模式，在之前的系列文章中讲解了RabbitMQ消息队列，使用的是原生代码进行编写，在Micro中，其实为我们封装了RabbitMQ消息队列，不仅仅是RabbitMQ还有NATs作为消息中间件，称之为broker模式，在这篇文章中我们进行一个简单的发布与订阅模式的介绍。
入门案例介绍
首先建立如下目录
编写我们的main.go,首先，定义一个待发布的主题和一个broker，使用broker需要导入包。
代码语言：javascript复制import "github.com/micro/go-micro/broker"
var (
// 定义一个待发布的主题
topic = "mu.micro.book.topic.payment.done"
b broker.Broker
)
编写一个发布消息的函数如下，在代码中broker.Message是一个结构体，我们将要发送的消息需要存入该结构体中，最后使用broker.Publish发布消息。
代码语言：javascript复制func pub() {
// 该Ticker包含一个通道字段，并会每隔时间段d就向该通道发送当时的时间。
// 它会调整时间间隔或者丢弃tick信息以适应反应慢的接收者。如果d<=0会panic。关闭该Ticker可以释放相关资源。
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())),
}
log.Infof(broker.String())
// 发布消息
if err := broker.Publish(topic, msg); err != nil {
log.Infof("[pub] Message publication failed: %v", err)
} else {
fmt.Println("[pub] Message published: ", string(msg.Body))
}
i
}
}
有发布消息的函数，肯定就有接收消息的函数如下,通过broker.Subscribe完成主题的订阅。其他的调用见名知义，不再详细介绍。
代码语言：javascript复制func sub() {
// 订阅消息并接收
_, err := broker.Subscribe(topic, func(p broker.Event) error {
log.Info("[sub] Received Body: %s,Header: %sn", string(p.Message().Body), p.Message().Header)
return nil
})
if err != nil {
fmt.Println(err)
}
}
编写我们的main函数,这里需要注意导入相关的包
代码语言：javascript复制"github.com/micro/go-micro/config/cmd"
"github.com/micro/go-micro/util/log"
使用Init进行初始化然后开始连接，开启两个协程进行收发消息，使用time.After()设置过期时间，当时间到时，并向当前通道发送事件，结束主线程。防止主线程提前结束。如果这一点不懂请移步我的博客看关于进程线程和Go协程总结这篇文章。
代码语言：javascript复制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 * 20)
}
看到这里是不是觉得很简单，测试一番,结果如下
代码语言：javascript复制PS F:micolearnday02mico-brokerbasic> go run main.go
2019/08/12 16:28:37 http
[pub] Message published: 0:2019-08-12 16:28:37.388162 0800 CST m= 1.024290401
2019/08/12 16:28:37 [sub] Received Body: %s,Header: %s
0:2019-08-12 16:28:37.388162 0800 CST m= 1.024290401map[id:0]
2019/08/12 16:28:38 http
[pub] Message published: 1:2019-08-12 16:28:38.3884518 0800 CST m= 2.024580201
2019/08/12 16:28:38 [sub] Received Body: %s,Header: %s
1:2019-08-12 16:28:38.3884518 0800 CST m= 2.024580201map[id:1]
2019/08/12 16:28:39 http
[pub] Message published: 2:2019-08-12 16:28:39.3878092 0800 CST m= 3.023937601
2019/08/12 16:28:39 [sub] Received Body: %s,Header: %s
2:2019-08-12 16:28:39.3878092 0800 CST m= 3.023937601map[id:2]
2019/08/12 16:28:40 http
[pub] Message published: 3:2019-08-12 16:28:40.3881043 0800 CST m= 4.024232701
2019/08/12 16:28:40 [sub] Received Body: %s,Header: %s
3:2019-08-12 16:28:40.3881043 0800 CST m= 4.024232701map[id:3]
2019/08/12 16:28:41 http
[pub] Message published: 4:2019-08-12 16:28:41.3884312 0800 CST m= 5.024559601
2019/08/12 16:28:41 [sub] Received Body: %s,Header: %s
4:2019-08-12 16:28:41.3884312 0800 CST m= 5.024559601map[id:4]
2019/08/12 16:28:42 http
[pub] Message published: 5:2019-08-12 16:28:42.3887561 0800 CST m= 6.024884501
2019/08/12 16:28:42 [sub] Received Body: %s,Header: %s
5:2019-08-12 16:28:42.3887561 0800 CST m= 6.024884501map[id:5]
2019/08/12 16:28:43 http
[pub] Message published: 6:2019-08-12 16:28:43.3870869 0800 CST m= 7.023215301
2019/08/12 16:28:43 [sub] Received Body: %s,Header: %s
6:2019-08-12 16:28:43.3870869 0800 CST m= 7.023215301map[id:6]
2019/08/12 16:28:44 http
[pub] Message published: 7:2019-08-12 16:28:44.3885307 0800 CST m= 8.024659101
2019/08/12 16:28:44 [sub] Received Body: %s,Header: %s
7:2019-08-12 16:28:44.3885307 0800 CST m= 8.024659101map[id:7]
2019/08/12 16:28:45 http
[pub] Message published: 8:2019-08-12 16:28:45.3887328 0800 CST m= 9.024861201
2019/08/12 16:28:45 [sub] Received Body: %s,Header: %s
8:2019-08-12 16:28:45.3887328 0800 CST m= 9.024861201map[id:8]
2019/08/12 16:28:46 http
[pub] Message published: 9:2019-08-12 16:28:46.3870944 0800 CST m= 10.023222801
2019/08/12 16:28:46 [sub] Received Body: %s,Header: %s
9:2019-08-12 16:28:46.3870944 0800 CST m= 10.023222801map[id:9]
2019/08/12 16:28:47 http
[pub] Message published: 10:2019-08-12 16:28:47.3873887 0800 CST m= 11.023517101
2019/08/12 16:28:47 [sub] Received Body: %s,Header: %s
10:2019-08-12 16:28:47.3873887 0800 CST m= 11.023517101map[id:10]
2019/08/12 16:28:48 http
[pub] Message published: 11:2019-08-12 16:28:48.3887128 0800 CST m= 12.024841201
2019/08/12 16:28:48 [sub] Received Body: %s,Header: %s
11:2019-08-12 16:28:48.3887128 0800 CST m= 12.024841201map[id:11]
2019/08/12 16:28:49 http
[pub] Message published: 12:2019-08-12 16:28:49.3870651 0800 CST m= 13.023193501
2019/08/12 16:28:49 [sub] Received Body: %s,Header: %s
12:2019-08-12 16:28:49.3870651 0800 CST m= 13.023193501map[id:12]
2019/08/12 16:28:50 http
[pub] Message published: 13:2019-08-12 16:28:50.3873677 0800 CST m= 14.023496101
2019/08/12 16:28:50 [sub] Received Body: %s,Header: %s
13:2019-08-12 16:28:50.3873677 0800 CST m= 14.023496101map[id:13]
2019/08/12 16:28:51 http
[pub] Message published: 14:2019-08-12 16:28:51.3886879 0800 CST m= 15.024816301
2019/08/12 16:28:51 [sub] Received Body: %s,Header: %s
14:2019-08-12 16:28:51.3886879 0800 CST m= 15.024816301map[id:14]
2019/08/12 16:28:52 http
[pub] Message published: 15:2019-08-12 16:28:52.3870503 0800 CST m= 16.023178701
2019/08/12 16:28:52 [sub] Received Body: %s,Header: %s
15:2019-08-12 16:28:52.3870503 0800 CST m= 16.023178701map[id:15]
2019/08/12 16:28:53 http
[pub] Message published: 16:2019-08-12 16:28:53.3875243 0800 CST m= 17.023652701
2019/08/12 16:28:53 [sub] Received Body: %s,Header: %s
16:2019-08-12 16:28:53.3875243 0800 CST m= 17.023652701map[id:16]
2019/08/12 16:28:54 http
[pub] Message published: 17:2019-08-12 16:28:54.3885713 0800 CST m= 18.024699701
2019/08/12 16:28:54 [sub] Received Body: %s,Header: %s
17:2019-08-12 16:28:54.3885713 0800 CST m= 18.024699701map[id:17]
2019/08/12 16:28:55 http
[pub] Message published: 18:2019-08-12 16:28:55.3872265 0800 CST m= 19.023354901
2019/08/12 16:28:55 [sub] Received Body: %s,Header: %s
18:2019-08-12 16:28:55.3872265 0800 CST m= 19.023354901map[id:18]
END
今日推荐阅读
RabbitMQ系列笔记广播模式和路由模式 RabbitMQ系列笔记入门篇
RabbitMQ系列笔记work模式
RabbitMQ系列笔记work模式
protoc语法详解及结合grpc定义服务
Golang中Model的使用
基于Nginx和Consul构建高可用及自动发现的Docker服务架构
