Kratos微服务框架下的消息队列应用-阿里云开发者社区

Kratos微服务框架下的消息队列应用

2022-04-03 2521

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Kratos微服务框架下的消息队列应用，包括了：Kafka、Rabbitmq、mqtt、redis，nats，websocket等。

什么是消息队列

MQ就是消息队列，是Message Queue的缩写。消息队列是一种通信方式。消息的本质就是一种数据结构。因为MQ把项目中的消息集中式的处理和存储，所以MQ主要有解耦，并发，和削峰的功能。

为什么要使用消息队列

1. 异步

通常的微服务实现里面，都是通过RPC进行微服务之间的相互调用，这是同步的。如果消息队列的话，可以实现异步的调用。至于异步有啥好处呢，主要是为了削峰。

2. 削峰

同步的调用会带来一个问题：瞬时流量。客户的调用同步接口节奏，你是无法把控的，流量将会是忽高忽低的，猛的来一波，搞不好系统就崩了溃了。

如果消息队列的话，可以实现异步的调用，并且可以实现削峰，请求进来，我先放到消息队列里面去，慢慢消化掉，不至于猛的来一下，把系统击垮。

3. 解耦

通常的微服务实现里面，都是通过RPC进行微服务之间的相互调用，那么意味着，你要做到一件事情，你必须要知道做事情的对方是谁。在微服务的世界里面，如果设计得不好，那就是一团糟的相互调用网络，看得你晕晕的，运维会疯，后面接手的开发人员也得疯。

应用了消息队列，你就只需要跟消息队列这个代理打交道，单线联系，关系简单。我们只需要生产消息，消费消息，至于是谁消费的，谁生产的，完全不用去管它。架构上，就会清爽多了。所以，要对服务进行解耦，消息队列是一个很好的选择。

Kratos与消息队列

Kratos现在的版本（v2.2.1）中，还没有对消息队列的直接支持，但是要运用还是容易的。官方有一个空壳示例代码BeerShop，可以看到，在data层，使用Kafka的痕迹。

对于在Kratos微服务框架里面应用消息队列，我认为有两种方式可以实现：

在data层，使用消息队列，但是在这个层，你只能在那生产消息，而不好消费消息。
将消息队列的客户端实现为微服务的一个Server，然后在微服务的Service中消费消息和生产消息。

第一个方式的应用面不广，更多的时候，第二种方式的应用面会更广一些，我选择了第二种方式。但是，Kratos官方并没有支持这一种方式。故而，我只能够自己动手实现了，我从另外一个微服务Go-Micro里面提取了其Broker的实现，并且将其实现为Kratos框架里面的一个Server。事实证明，这样是可行的，并且很好使。

你可能会问，为什么我不直接使用Go-Micro呢？因为Go-Micro是一个很重的微服务框架，尽管它的功能很丰富，几乎支持了大部分的微服务需求。但是对于一个应用来说，我并不需要使用所有的技术栈、中间件，我只需要部分的技术栈。所以，我宁愿做加法，也不愿意去做减法。对于服务端来说，可控、可用、可维护是最重要的。极简，是一个很好的选择。另外，我还要腹诽一点，我从Go-Micro提取出来的Broker在测试的过程中发现，都有一些瑕疵。

我实现的代码，我放到了github：https://github.com/tx7do/kratos-transport。

它所支持的协议和消息队列有：

Kafka
RabbitMQ
NATS
Redis
MQTT
WebSocket

基本上是够用了。

kratos-transport的应用

它主要分为了3个部分：

1. Codec 编解码器

这一块和Broker都是从Go-Micro提取出来的，但是它对我的应用来说，还并没有什么用处，因为我的编解码器是很独特的，需要定制的。所以，我暂时还没用上这一块。

2. Broker 消息队列

可以直接拿来用，我拿Kafka举例：

packagekafkaimport (
"context""fmt""github.com/tx7do/kratos-transport/broker""os""os/signal""syscall""testing")
funcTestSubscribe(t*testing.T) {
interrupt :=make(chanos.Signal, 1)
signal.Notify(interrupt, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
ctx :=context.Background()
b :=NewBroker(
broker.Addrs("127.0.0.1:9092"),
broker.OptionContext(ctx),
    )
_, _=b.Subscribe("logger.sensor.ts", receive,
broker.SubscribeContext(ctx),
broker.Queue("fx-group"),
    )
<-interrupt}
funcreceive(eventbroker.Event) error {
fmt.Println("Topic: ", event.Topic(), " Payload: ", string(event.Message().Body))
//_ = event.Ack()returnnil}
funcTestPublish(t*testing.T) {
ctx :=context.Background()
b :=NewBroker(
broker.Addrs("127.0.0.1:9092"),
broker.OptionContext(ctx),
    )
varmsgbroker.Messagemsg.Body= []byte(`{"Humidity":60, "Temperature":25}`)
_=b.Publish("logger.sensor.ts", &msg)
}

3. Server 封装给Kratos的Server实现

还是拿Kafka举例：

packagemainimport (
"fmt""github.com/go-kratos/kratos/v2""log""github.com/tx7do/kratos-transport/broker""github.com/tx7do/kratos-transport/transport/kafka")
funcmain() {
//ctx := context.Background()kafkaSrv :=kafka.NewServer(
kafka.Address("127.0.0.1:9092"),
kafka.Subscribe("test_topic", "a-group", receive),
    )
app :=kratos.New(
kratos.Name("kafka"),
kratos.Server(
kafkaSrv,
        ),
    )
iferr :=app.Run(); err!=nil {
log.Println(err)
    }
}
funcreceive(eventbroker.Event) error {
fmt.Println("Topic: ", event.Topic(), " Payload: ", string(event.Message().Body))
returnnil}
funcsendData(sendData []byte) error {
varmsgbroker.Messagemsg.Body=sendDatakafkaSrv.Publish(topic, &msg)
}

另外再看一个例子，是Websocket的，它的应用其实也是很广的：

packagemainimport (
"fmt""log""github.com/go-kratos/kratos/v2""github.com/tx7do/kratos-transport/transport/websocket")
funcmain() {
//ctx := context.Background()wsSrv :=websocket.NewServer(
websocket.Address(":8800"),
websocket.ReadHandle("/ws", handleMessage),
websocket.ConnectHandle(handleConnect),
    )
app :=kratos.New(
kratos.Name("websocket"),
kratos.Server(
wsSrv,
        ),
    )
iferr :=app.Run(); err!=nil {
log.Println(err)
    }
}
funchandleConnect(connectionIdstring, registerbool) {
ifregister {
fmt.Printf("%s connected\n", connectionId)
    } else {
fmt.Printf("%s disconnect\n", connectionId)
    }
}
funchandleMessage(connectionIdstring, message*websocket.Message) (*websocket.Message, error) {
fmt.Printf("[%s] Payload: %s\n", connectionId, string(message.Body))
varrelyMsgwebsocket.MessagerelyMsg.Body= []byte("hello")
return&relyMsg, nil}

具体的应用实例

我写了两个实例代码，并且都已经提交到了Kratos的examples代码仓库中去了。这两个例子都是物联网方面的应用。

kratos-cqrs

这是一个简单的CQRS的实现，主要就是拿了Kafka来消费来自于传感器的遥感数据，然后把数据存储到数据库中去。

需要注意的是，这个实例并不够完整，我并没有实现MQTT的消费，没有实现前端页面等等。只实现了对Kafka的消费。

kratos-realtimemap

这是一个完整的例子，有前端，有后端，可以完整的跑起来看。

通过MQTT接收一个开放的公交遥测数据源，然后通过REST和Websocket向前端发送数据，在地图上展现出来车辆的轨迹、车辆的位置、车辆的速度、开关门状态等等。