Golang 微服务框架 Kratos 应用 RocketMQ 消息队列

消息队列是一种异步的服务间通信方式，适用于无服务器和微服务架构。消息在被处理和删除之前一直存储在队列上。每条消息仅可被一位用户处理一次。消息队列可被用于分离重量级处理、缓冲或批处理工作以及缓解高峰期工作负载。

消息队列是大型分布式系统不可缺少的中间件，也是高并发系统的基石中间件，所以掌握好消息队列 MQ 就变得极其重要。

在本文当中，您将了解到：什么是消息队列？什么是 RocketMQ？怎样在微服务框架 Kratos 当中应用 RocketMQ 进行业务开发。

什么是消息队列

消息队列（Message Queue，简称 MQ）指保存消息的一个容器，其实本质就是一个保存数据的队列。

消息中间件是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的构建。

消息中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性的系统架构。目前使用较多的消息队列有：ActiveMQ、RabbitMQ、ZeroMQ、RocketMQ、MetaMQ、RocketMQ、NAQ、NATS、Pulsar 等。

消息队列应用场景

消息中间件在互联网公司使用得越来越多，主要用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。以下介绍消息队列在实际应用中常用的使用场景：异步处理，应用解耦，流量削峰和消息通讯。

异步处理

通常的微服务实现的接口，都是通过 RPC 进行微服务、服务客户端之间的相互调用，这是同步阻塞执行。有一些业务，业务流程比较耗时且可以不需要立即返回结果，还有一些业务可以互不干扰的并行执行，那么我们就可以将之转为异步，并发执行。从而减少同步接口的请求响应时间，从而提高系统的吞吐量。

以下单为例，用户下单后需要实施：生成订单、赠送活动积分、赠送红包、发送下单成功通知等，一系列业务处理。假设三个业务节点每个使用 100 毫秒钟，不考虑网络等其他开销，则串行方式的时间是 400 毫秒，并行的时间只需要 200 毫秒。这样就大大提高了系统的吞吐量。

应用解耦

应用解耦，顾名思义就是解除应用系统之间的耦合依赖。通过消息队列，使得每个应用系统不必受其他系统影响，可以更独立自主。

以电商系统为例，用户下单后，订单系统需要通知积分系统。一般的做法是：订单系统直接调用积分系统的接口。这就使得应用系统间的耦合特别紧密。如果积分系统无法访问，则积分处理失败，从而导致订单失败。

加入消息队列之后，用户下单后，订单系统完成下单业务后，将消息写入消息队列，返回用户订单下单成功。积分系统通过订阅下单消息的方式获取下单通知消息，从而进行积分操作。实现订单系统与库存系统的应用解耦。如果，在下单时积分系统系统异常，也不影响用户正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作。

流量削峰

流量削峰也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛。

以秒杀活动为例，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列，秒杀业务处理系统根据消息队列中的请求信息，再做后续处理。

如上图所示，服务器接收到用户的请求后，首先写入消息队列，秒杀业务处理系统根据消息队列中的请求信息，做后续业务处理。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。

消息通讯

消息通讯是指应用间的数据通信。消息队列一般都内置了高效的通信机制，因此也可以用在单纯的消息通讯上。比如：实现点对点消息队列，或者聊天室等点对点通讯。

以上实际是消息队列的两种消息模式，点对点或发布订阅模式。

什么是 RocketMQ？

RocketMQ 是由阿里捐赠给 Apache 的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历了淘宝双十一的洗礼。RocketMQ 既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

RocketMQ 特点

• RocketMQ 是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式等特点

• Producer、Consumer、队列都可以分布式

• Producer 向一些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做广播消费，则一个 Consumer 实例消费这个 Topic 对应的所有队列，如果做集群消费，则多个 Consumer 实例平均消费这个 Topic 对应的队列集合

• 能够保证严格的消息顺序

• 支持拉（pull）和推（push）两种消息模式

• 高效的订阅者水平扩展能力

• 实时的消息订阅机制

• 亿级消息堆积能力

• 支持多种消息协议，如 JMS、OpenMessaging 等

• 较少的依赖

RocketMQ 核心概念

• Topic：消息主题，一级消息类型，生产者向其发送消息。

• Message：生产者向 Topic 发送并最终传送给消费者的数据消息的载体。

• 消息属性：生产者可以为消息定义的属性，包含 Message Key 和 Tag。

• Message Key：消息的业务标识，由消息生产者（Producer）设置，唯一标识某个业务逻辑。

• Message ID：消息的全局唯一标识，由消息队列 RocketMQ 系统自动生成，唯一标识某条消息。

• Tag：消息标签，二级消息类型，用来进一步区分某个 Topic 下的消息分类

• Producer：也称为消息发布者，负责生产并发送消息至 Topic。

• Consumer：也称为消息订阅者，负责从 Topic 接收并消费消息。

• 分区：即 Topic Partition，物理上的概念。每个 Topic 包含一个或多个分区。

• 消费位点：每个 Topic 会有多个分区，每个分区会统计当前消息的总条数，这个称为最大位点 MaxOffset；分区的起始位置对应的位置叫做起始位点 MinOffset。

• Group：一类生产者或消费者，这类生产者或消费者通常生产或消费同一类消息，且消息发布或订阅的逻辑一致。

• Group ID：Group 的标识。

• 队列：个 Topic 下会由一到多个队列来存储消息。

• Exactly-Once 投递语义：Exactly-Once 投递语义是指发送到消息系统的消息只能被 Consumer 处理且仅处理一次，即使 Producer 重试消息发送导致某消息重复投递，该消息在 Consumer 也只被消费一次。

• 集群消费：一个 Group ID 所标识的所有 Consumer 平均分摊消费消息。例如某个 Topic 有 9 条消息，一个 Group ID 有 3 个 Consumer 实例，那么在集群消费模式下每个实例平均分摊，只消费其中的 3 条消息。

• 广播消费：一个 Group ID 所标识的所有 Consumer 都会各自消费某条消息一次。例如某个 Topic 有 9 条消息，一个 Group ID 有 3 个 Consumer 实例，那么在广播消费模式下每个实例都会各自消费 9 条消息。

• 定时消息：Producer 将消息发送到消息队列 RocketMQ 服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到 Consumer 进行消费，该消息即定时消息。

• 延时消息：Producer 将消息发送到消息队列 RocketMQ 服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到 Consumer 进行消费，该消息即延时消息。

• 事务消息：RocketMQ 提供类似 X/Open XA 的分布事务功能，通过消息队列 RocketMQ 的事务消息能达到分布式事务的最终一致。

• 顺序消息：RocketMQ 提供的一种按照顺序进行发布和消费的消息类型，分为全局顺序消息和分区顺序消息。

• 全局顺序消息：对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。

• 分区顺序消息：对于指定的一个 Topic，所有消息根据 Sharding Key 进行区块分区。同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。Sharding Key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Message Key 是完全不同的概念。

• 消息堆积：Producer 已经将消息发送到消息队列 RocketMQ 的服务端，但由于 Consumer 消费能力有限，未能在短时间内将所有消息正确消费掉，此时在消息队列 RocketMQ 的服务端保存着未被消费的消息，该状态即消息堆积。

• 消息过滤：Consumer 可以根据消息标签（Tag）对消息进行过滤，确保 Consumer 最终只接收被过滤后的消息类型。消息过滤在消息队列 RocketMQ 的服务端完成。

• 消息轨迹：在一条消息从 Producer 发出到 Consumer 消费处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，您能清晰定位消息从 Producer 发出，经由消息队列 RocketMQ 服务端，投递给 Consumer 的完整链路，方便定位排查问题。

• 重置消费位点：以时间轴为坐标，在消息持久化存储的时间范围内（默认 3 天），重新设置 Consumer 对已订阅的 Topic 的消费进度，设置完成后 Consumer 将接收设定时间点之后由 Producer 发送到消息队列 RocketMQ 服务端的消息。

• 死信队列：死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列 RocketMQ 会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明 Consumer 在正常情况下无法正确地消费该消息。此时，消息队列 RocketMQ 不会立刻将消息丢弃，而是将这条消息发送到该 Consumer 对应的特殊队列中。

消息队列 RocketMQ 将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。

RocketMQ 架构

RocketMQ 架构共有四个集群：NameServer 集群、Broker 集群、Producer 集群、Consumer 集群

NameServer 集群

提供轻量级的服务发现及路由，每个 NameServer 记录完整的路由信息，提供相应的读写服务，支持快速存储扩展。有些其它开源中间件使用 ZooKeeper 实现服务发现及路由功能，如 Apache RocketMQ。

NameServer 是一个功能齐全的服务器，主要包含两个功能：

1. Broker 管理，接收来自 Broker 集群的注册请求，提供心跳机制检测 Broker 是否存活

2. 路由管理，每个 NameServer 持有全部有关 Broker 集群和客户端请求队列的路由信息

Broker 集群

通过提供轻量级的 Topic 和 Queue 机制处理消息存储。同时支持推（Push）和拉（Pull）两种模型，包含容错机制。提供强大的峰值填充和以原始时间顺序累积数千亿条消息的能力。此外还提供灾难恢复，丰富的指标统计数据和警报机制，这些都是传统的消息系统缺乏的。

Broker 有几个重要的子模块：

1. 远程处理模块，Broker 入口，处理来自客户端的请求

2. 客户端管理，管理客户端（包括消息生产者和消费者），维护消费者的主题订阅

3. 存储服务，提供在物理硬盘上存储和查询消息的简单 API

4. HA 服务，提供主从 Broker 间数据同步

5. 索引服务，通过指定键为消息建立索引并提供快速消息查询

Producer 集群

消息生产者支持分布式部署，分布式生产者通过多种负载均衡模式向 Broker 集群发送消息。

Consumer 集群

消息消费者也支持 Push 和 Pull 模型的分布式部署，还支持集群消费和消息广播。提供了实时的消息订阅机制，可以满足大多数消费者的需求。

RocketMQ 消息收发模型

消息队列 RocketMQ 支持发布和订阅模型，消息生产者应用创建 Topic 并将消息发送到 Topic。消费者应用创建对 Topic 的订阅以便从其接收消息。通信可以是一对多（扇出）、多对一（扇入）和多对多。具体通信如下图所示。

• 生产者集群：用来表示发送消息应用，一个生产者集群下包含多个生产者实例，可以是多台机器，也可以是一台机器的多个进程，或者一个进程的多个生产者对象。

• 消费者集群：用来表示消费消息应用，一个消费者集群下包含多个消费者实例，可以是多台机器，也可以是多个进程，或者是一个进程的多个消费者对象。

一个生产者集群可以发送多个 Topic 消息。发送分布式事务消息时，如果生产者中途意外宕机，消息队列 RocketMQ 服务端会主动回调生产者集群的任意一台机器来确认事务状态。

一个消费者集群下的多个消费者以均摊方式消费消息。如果设置的是广播方式，那么这个消费者集群下的每个实例都消费全量数据。

一个消费者集群对应一个 Group ID，一个 Group ID 可以订阅多个 Topic，如上图中的 Group 2 所示。Group 和 Topic 的订阅关系可以通过直接在程序中设置即可。

Docker 部署开发环境

必须要至少启动一个 NameServer，一个 Broker。

docker pull apache/rocketmq:latest
# NameServerdocker run -d \      --name rmqnamesrv \      -e "JAVA_OPT_EXT=-Xms512M -Xmx512M -Xmn128m" \      -p 9876:9876 \      apache/rocketmq:latest \      sh mqnamesrv
# Brokerdocker run -d \      --name rmqbroker \      -p 10911:10911 \      -p 10909:10909 \      -p 10912:10912 \      --link rmqnamesrv \      -e "JAVA_OPT_EXT=-Xms512M -Xmx512M -Xmn128m" \      -e "NAMESRV_ADDR=rmqnamesrv:9876" \      apache/rocketmq:latest \      sh mqbroker -c /home/rocketmq/rocketmq-4.9.2/conf/broker.conf

docker pull styletang/rocketmq-console-ng:latest
docker run -d \    --name rmqconsole \    -p 9800:8080 \    --link rmqnamesrv \    -e "JAVA_OPTS=-Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" \    -t styletang/rocketmq-console-ng:latest

控制台访问地址： http://localhost:9800/#/

另外，NameServer 下发的是 Docker 容器的内网 IP 地址，从宿主机的外网访问是访问不了的，需要进行配置：

vi /home/rocketmq/rocketmq-4.9.2/conf/broker.conf

添加如下配置，brokerIP1 可以是 ip 也可以是 dns，hostname：

brokerIP1 = host.docker.internal

Kratos 下如何应用 RocketMQ？

我对 RocketMQ 做了一个封装，要在 Kratos 下面使用 RocketMQ，首先需要在项目中引用我封装的两个库：

第一个库可以视之为 RocketMQ 客户端的一个封装：

go get -u github.com/tx7do/kratos-transport/broker/rocketmq

这一个库是讲 ROcketMQ 的客户端封装成一个 Kratos 的transport.Server，该库依赖上面的库：

go get -u github.com/tx7do/kratos-transport/transport/rocketmq

想要在 Kratos 里面应用 RocketMQ，有两条途径可以达成：

1. 在Data层引用 RocketMQ 的Broker，仅用于发布(Publish)消息之用，换言之，就是只发送不接收的单向通讯；

2. 在Server层引用 RocketMQ 的Server，可以发布(Publish)消息，也可以订阅(Subscribe)消息，换言之，就是既发送又接收的全双工通讯。

接下来我就详细的讲解应用方法：

在Data层引用 RocketMQ 的Broker

首先创建 RocketMQ 的Broker:

import (    "github.com/tx7do/kratos-transport/broker"    "github.com/tx7do/kratos-transport/broker/rocketmq")
func NewRocketMQBroker(cfg *conf.Bootstrap) broker.Broker {    b := rocketmq.NewBroker(        broker.WithNameServer(cfg.Data.RocketMQ.Addrs),        broker.WithCodec(cfg.Data.RocketMQ.Codec),    )    if b == nil {        return nil    }
    _ = b.Init()
    if err := b.Connect(); err != nil {        return nil    }
    return b}

然后，注入到Wire的ProviderSet：

package data
import "github.com/google/wire"
// ProviderSet is data providers.var ProviderSet = wire.NewSet(    ...    NewRocketMQBroker,)

最后，我们就可以在Service里面调用了：

package service
type ReportService struct {    v1.ReportServiceHTTPServer
    rocketmqBroker broker.Broker    log         *log.Helper}
func NewReportService(logger log.Logger, rocketmqBroker broker.Broker) *ReportService {    l := log.NewHelper(log.With(logger, "module", "report/service/agent-service"))    return &ReportService{        log:         l,        rocketmqBroker: rocketmqBroker,    }}
func (s *ReportService) PostReport(_ context.Context, req *v1.PostReportRequest) (*v1.PostReportResponse, error) {    _ = s.rocketmqBroker.Publish(topic.EventReportData, reportV1.RealTimeWarehousingData{        EventName:  &req.EventName,        ReportData: &req.Content,        CreateTime: util.UnixMilliToStringPtr(trans.Int64(time.Now().UnixMilli())),    })    return &v1.PostReportResponse{        Code: 0,        Msg:  "success",    }, nil}

需要注意的是，添加了以上代码之后，需要使用命令生成 Wire 的胶水代码：

go run -mod=mod github.com/google/wire/cmd/wire ./cmd/server

在Server层引用 RocketMQ 的Server

首先要创建Server：

package server
import (    ...    "github.com/tx7do/kratos-transport/transport/rocketmq")
// NewRocketMQServer create a rocketmq server.func NewRocketMQServer(cfg *conf.Bootstrap, _ log.Logger, svc *service.SaverService) *rocketmq.Server {    ctx := context.Background()
    srv := rocketmq.NewServer(        rocketmq.WithNameServer(cfg.Server.RocketMQ.Addrs),        rocketmq.WithGlobalTracerProvider(),        rocketmq.WithGlobalPropagator(),        rocketmq.WithCodec("json"),    )
    registerRocketMQSubscribers(ctx, srv, svc)
    return srv}
func registerRocketMQSubscribers(ctx context.Context, srv *rocketmq.Server, svc *service.SaverService) {    _ = rocketmq.RegisterSubscriber(srv, ctx,        topic.UserReportData, topic.LoggerSaverQueue, false,        svc.SaveUserReport,    )
    _ = rocketmq.RegisterSubscriber(srv, ctx,        topic.EventReportData, topic.LoggerSaverQueue, false,        svc.SaveEventReport,    )}

接着，调用kratos.Server把 RocketMQ 的服务器注册到 Kratos 里去：

func newApp(ll log.Logger, rr registry.Registrar, ks *rocketmq.Server) *kratos.App {    return kratos.New(        kratos.ID(Service.GetInstanceId()),        kratos.Name(Service.Name),        kratos.Version(Service.Version),        kratos.Metadata(Service.Metadata),        kratos.Logger(ll),        kratos.Server(            ks,        ),        kratos.Registrar(rr),    )}

最后，我们就可以在Service里愉快的玩耍了，在这里，我只演示收到 RocketMQ 消息之后立即写入数据库的操作：

package service
type SaverService struct {    log          *log.Helper    statusRepo   *data.AcceptStatusRepo    realtimeRepo *data.RealtimeWarehousingRepo}
func NewSaverService(    logger log.Logger,    statusRepo *data.AcceptStatusRepo,    realtimeRepo *data.RealtimeWarehousingRepo,) *SaverService {    l := log.NewHelper(log.With(logger, "module", "saver/service/logger-service"))    return &SaverService{        log:          l,        statusRepo:   statusRepo,        realtimeRepo: realtimeRepo,    }}
func (s *SaverService) SaveUserReport(_ context.Context, _ string, _ broker.Headers, msg *v1.AcceptStatusReportData) error {    return s.statusRepo.Create(msg)}
func (s *SaverService) SaveEventReport(_ context.Context, _ string, _ broker.Headers, msg *v1.RealTimeWarehousingData) error {    return s.realtimeRepo.Create(msg)}

实例代码

以上代码以及接口定义，可以在我的另外一个开源项目里面找到：

• https://github.com/tx7do/kratos-uba
  

• https://gitee.com/tx7do/kratos-uba
  

需要注意的是，这个项目使用的是 Kafka，但是差异不大，因为接口是一致的，只是注入的参数有差异罢了。

参考资料

• 什么是消息队列？
  

• 秒懂消息队列MQ，万字总结带你全面了解消息队列MQ
  

• 什么是消息队列？
  

• 什么是消息队列RocketMQ版？
  

• RocketMQ 介绍及核心概念
  

• RocketMQ 简介
  

• Apache RocketMQ 官网