转载

小鼹鼠（Go）的新朋友 —— Apache RocketMQ

本文作者王文锋

Horizon.ai研发工程师，
Apache RocketMQ Committer
Apache Go Client项目负责人

今天和大家来谈一个有关消息队列的话题，消息队列作为最常见的互联网中间件，广泛的应用于业务之间异步解耦、分布式事务的数据一致性同步、流量削峰填谷、分布式缓存更新等场景中。在目前Go的生态里面，大家最常用的消息队列之一就是NSQ了，NSQ最大的优势就是简单易用，但在一些对稳定性和可靠性要求比较苛刻的场景中，或者一些业务更复杂的场景中，NSQ会略显不足。

在如今的2020年，一个Gopher做MQ的技术选型的时候，还有没有更好的选择？这时候很多同学可能会想到Kafka，不过今天给大家介绍的并不是Kafka，而是另一款优秀的消息队列产品——Apache RocketMQ。在文章的后半部分，会有一个RocketMQ和Kafka的对比，希望能对大家有所启发。鉴于篇幅有限，今天主要侧重于讲解应用面，具体的技术细节不会过多展开，将来有机会再和大家进行分享。

本文字数： 3441字

精读时间： 10分钟

也可在5分钟内完成速读

Apache RocketMQ

小鼹鼠（Go）的新朋友 —— Apache RocketMQ

Apache RocketMQ 在2016年被阿里巴巴捐献给了 Apache 基金会，经过孵化后成为了 Apache 的顶级项目。2011年阿里巴巴中间件团队自主研发了RocketMQ 消息中间件，具有单机亿级消息堆积能力，且能支持严格的消息顺序。在2016年的时候，阿里巴巴线上所有消息全部通过RocketMQ来转发投递，2016年（暂时没有最新数据）双十一当天达到了万亿级消息量，峰值几千万TPS，创造了当时国内乃至世界上最大的消息流转记录。在2017年的时候，RocketMQ的单机TPS就成功突破了50w。RocketMQ开源多年以来，已广泛的应用于国内各种企业的各种场景中。在最新的v4.6版本中，主要特性主要有:

基于Topic模型的消息发布/订阅
支持集群消费/广播消费
支持顺序消息
支持定时/延时消息
支持事务消息
支持消息重试和消费过滤
高可靠性和稳定性
多种Connector
支持k8s Operator
支持多租户和ACL
支持Prometheus

不过，由于RocketMQ是基于Java实现，长期以来在Go语言社区并未广泛使用，很多社区的小伙伴也对其知之甚少。而在这几天，RocketMQ Go Client 2.0.0-rc1正式发布，正好借此机会，向社区的小伙伴进行下相关的介绍。

Go Client

背景

小鼹鼠（Go）的新朋友 —— Apache RocketMQ

如上图所示，在1.7号的时候，RocketMQ社区正式发布了2.0-rc1版本，在这之前，已经发布了3个alpha版本，目前最新版本的代码已经在头条和斗鱼线上有了较大规模的应用。因此虽然Go Client项目是一个完全社区驱动的项目，但还是请大家放心使用（PS: 大部分的坑我们都已经踩完啦），社区对issue的响应也是比较快速的。

在2.0之前，我们基于cgo开发了RocketMQ 1.2.x版本（图中1.2.x的小版本没有体现），由于Apache不允许发行版携带任何的二进制文件，所以需要使用1.2.x版本的同学自行编译RocketMQ CPP项目，而CPP项目依赖了不少c++库，编译起来实在是酸爽，以及很多Gopher并没有c++背景，出了问题，没办法自行排查，只能反馈给CPP的同学，导致沟通成本高、排查周期长、使用体验差等问题，因此我决定参考Java Client使用Go开发一个native版本的Client，解决以上问题。

从实现上来说，RocketMQ Java Client是一个典型的富客户端，应用了较多的设计模式和开发范式，存在整体逻辑较复杂，代码耦合度过高等问题，为了避免在Go项目里面引入这些问题，易于社区参与，我们将简单和可维护性放到了首位，基于Go的特性对Client进行了全新的设计。不过在具体开发过程中，对个别地方的把控还是做的不够好，略显粗糙和带着些许Java的味道。这块后续还有不少工作需要做，期待社区能有感兴趣的同学一起参与进来。

使用

<!-- go.mod -->
require (
    github.com/apache/rocketmq-client-go v2.0.0-rc1
)

生产消息

p, _ := rocketmq.NewProducer(
        producer.WithNameServer([]string{"127.0.0.1:9876"}),
        producer.WithRetry(2),
    )
    err := p.Start()
    // error handler
    msg := primitive.NewMessage("test-topic", []byte("Hello Golang!"))
    res, err := p.SendSync(context.Background(), msg)
    // response and error handler
    err = p.Shutdown()
    if err != nil {
        fmt.Printf("shutdown producer error: %s", err.Error())
    }

完整的代码请参考:

https://github.com/apache/rocketmq-client-go/blob/native/examples/producer/simple/main.go, 这是一个最简单的例子。

批量、异步等更多场景的例子请参考:

https://github.com/apache/rocketmq-client-go/blob/native/examples/producer

消费消息

 c, _ := rocketmq.NewPushConsumer(
        consumer.WithGroupName("testGroup"),
        consumer.WithNameServer([]string{"127.0.0.1:9876"}),
    )
    selector := consumer.MessageSelector{}
    msgHandler := func(ctx context.Context, 
                       msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
        for i := range msgs {
            fmt.Printf("recevied message: %v /n", msgs[i])
        }
        return consumer.ConsumeSuccess, nil
    }
    err := c.Subscribe("test-topic", selector, )
    // error handler
    // 注意: consumer实例启动前至少需要订阅1个topic，因为consumer启动后立马开始执行
    // 负载均衡逻辑，没有Topic订阅信息会退出
    err = c.Start()
    // error handler & gracefully shutdown

完整的代码请参考:

https://github.com/apache/rocketmq-client-go/blob/native/examples/consumer/simple/main.go, 这是一个最简单的例子。

顺序、广播等更多场景的例子请参考:

https://github.com/apache/rocketmq-client-go/blob/native/examples/consumer

高级用法

Interceptor

在Go Client，我们实现了Interceptor接口

type Invoker func(ctx context.Context, req, reply interface{}) error
type Interceptor func(ctx context.Context, req, reply interface{}, next Invoker) error

熟悉gRPC的同学是不是比较眼熟？我们使用了gRPC风格的Interceptor替换了Java中的RPCHook，在体验上变的更加友好。在使用方面，也基本类似，一般都是下面这个样子的:

func DemoInterceptor() primitive.Interceptor {
    return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
        // do before
        if msgs, ok := req.([]*primitive.Message); ok {
            // do something
        }
        err := next(ctx, req, reply)
        // do after
        switch reply.(type) {
        case *primitive.SendResult:
            // do something
        case *consumer.ConsumeResult:
            // do something
        default:
        }
        return err
    }
}

ACL

RocketMQ支持对不同的`topic`设置ACL，在Go Client中，producer和consumer都支持Credential参数，实现topic的私有访问。

type Credentials struct {
    AccessKey     string
    SecretKey     string
    SecurityToken string
}
func WithCredentials(c primitive.Credentials) Option {
    return func(options *consumerOptions) {
        options.ClientOptions.Credentials = c
    }
}
// consumer也一样
c, err := rocketmq.NewPushConsumer(
        ...,
        consumer.WithCredentials(primitive.Credentials{
            AccessKey: "abcdef",
            SecretKey: "123456",
        }),
    )

ACL在服务端具体的配置方法可以参考文末的文章链接。

结构化日志

Go Client项目默认使用了`logrus`作为默认的日志框架，不过为了满足大家个性化的需求，我们抽象出了日志接口，方便大家定制化进行开发:

type Logger interface {
    Debug(msg string, fields map[string]interface{})
    Info(msg string, fields map[string]interface{})
    Warning(msg string, fields map[string]interface{})
    Error(msg string, fields map[string]interface{})
    Fatal(msg string, fields map[string]interface{})
}

内部的默认实现:

https://github.com/apache/rocketmq-client-go/blob/native/rlog/log.go#L66。

大家有需要的话可以进行参考。

上述几种高级用法，可能是企业中最高频的几个场景，剩下的大家若感兴趣的话可以通过前文的`examples`链接进行探索。

对比Kafka

RocketMQ和Kafka都是非常优秀的开源消息队列产品，其应用场景也比较类似，性能方面二者均能达到一个极高的吞吐量，因此孰强孰弱更多的是一个仁者见仁智者见智的问题。就我个人认为，RocketMQ相对Kafka，最关键的两点在于:

Kafka依赖ZooKeeper作为HA方案，引入了外部依赖，提高了运维成本。而RocketMQ不依赖任何的第三方软件
对于国内用户而言，由于RocketMQ社区由阿里巴巴主导，也投入了很多专职的同学在做开源工作，加上社区参与的同学也大多都是中国人，因此在沟通和问题响应上是有优势的

下面是我列一个功能方面的对比表格，方便大家参考

功能	RocketMQ	Kafka
可靠性	同步/异步落盘	异步刷盘
HA	master-slave	依赖zk
单机TPS	百万级	百万级
投递实时性	长轮询, 平均~5ms	取决于轮询频率
消息失败重试
严格顺序消息
定时消息
事务消息
消息轨迹
消息堆积	上亿	上亿
消息过滤	服务端+客户端	客户端
消息查询	支持	不支持
消息回溯	直接支持	间接支持