如何实现一个简单易用的 RocketMQ SDK

简介: 2018 年,做为架构负责人,接到一个架构需求:实现一个简单易用的 RocketMQ SDK 。因为各个团队 RocketMQ 原生客户端配置起来千奇百怪,有的配置存在风险,各团队负责人都需要一个简洁易用的 RocketMQ SDK 。我立马调研相关开源的方案,当时 RocketMQ-Spring 项目并没有开源,而阿里云的 ONS SDK 是开源的,我只能讲目标转向 阿里云 ONS 。通过学习 ONS 的设计方式,我对于 RocketMQ 的客户端原理有了进一步了解,同时参考 ONS 的设计,也实现了公司内部使用的 RocketMQ SDK 。

2018 年,做为架构负责人,接到一个架构需求:实现一个简单易用的 RocketMQ SDK 。

因为各个团队 RocketMQ 原生客户端配置起来千奇百怪,有的配置存在风险,各团队负责人都需要一个简洁易用的 RocketMQ SDK 。

我立马调研相关开源的方案,当时 RocketMQ-Spring 项目并没有开源,而阿里云的 ONS SDK 是开源的,我只能讲目标转向 阿里云 ONS 。

通过学习 ONS 的设计方式,我对于 RocketMQ 的客户端原理有了进一步了解,同时参考 ONS 的设计,也实现了公司内部使用的 RocketMQ SDK 。

项目地址:https://github.com/makemyownlife/platform-rocketmq

之所以说简单,就是让用户(开发者)使用 SDK 时,减少心智负担

举三个例子:

1 发送顺序消息

使用原生代码发送消息时,会使用如下的代码:

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
   
   
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
   
   
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    }
}, orderId);

我们可以将 SDK API 简化为:

SendResult send(final ProducerMessage message, final String shardingKey);

开发者不需要定义队列选择器,只需要传递分片键 orderId 即可。

2 单条消息消费

使用原来代码定义消费监听器时,使用如下的代码:

consumer.registerMessageListener(new MessageListenerConcurrently() {
   
   
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
   
   
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        // 返回消息消费状态,ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

监听器内部,对于开发者操作的对象是消息列表 msgs ,很多开发同学想只操作一条消息。

于是,我们可以将 SDK API 简化为:

consumer.subscribe("mytest", new ConsumerListener() {
    @Override
    public ConsumerAction consumer(ConsumerMessage msg) {
        byte[] body = msg.getBody();
        System.out.println("msg:" + new String(body));
        return ConsumerAction.CommitMessage;
    }
});

开发者在消费时,可以一条一条操作,代码简洁了不少。

同时,很多开发者在使用普通消费、顺序消费时,需要返回延时消费的状态码时,两种消费模式定义的枚举也不相同。我们将枚举做了统一:

/**
 * 消费消息的返回结果
 */
public enum ConsumerAction {
   
   

    /**
     * 消费成功,继续消费下一条消息
     */
    CommitMessage,

    /**
     * 消费失败,告知服务器稍后再投递这条消息,继续消费其他消息
     */
    ReconsumeLater;
}

3 订阅关系一致

实际场景里,订阅关系不一致是极容易发生的事情,就算是高级别的架构师也会翻车,每次翻车现场都是惨不忍睹。

正确的订阅关系见下图:

正确的订阅关系

代码逻辑角度来看,每个消费者实例内订阅方法的主题、 TAG、监听逻辑都需要保持一致

当订阅关系不一致时,在 Broker 端同一个消费组内的各个消费者客户端的订阅信息相互被覆盖,从而导致某个消费者客户端无法拉取到新的消息。

怎么解决呢 ?

我当时想起了阿里技术专家沈询的一句话:

世界上解决一个计算机问题最简单的方法:“恰好”不需要解决它 !

公司内部出现订阅关系一致99%的问题是:消费者组一致的前提下,主题相同,但 TAG 不相同。

基于此,我的设计思路就明确了:不开放订阅 TAG 的权限!

没想到吧,我就是这么粗暴。

按照这种设计思路,虽然开始有的程序员会有质疑,但你和他梳理好消费者组的定义,以及做好领域划分,对业务来讲,反而清晰了。

4 写到最后

我并不认为我们写得多么的好,只是想让同学们理解:

1、成长的第一步就是模仿

2、把开发者当用户,以用户的体验为先

我经常去阅读阿里云产品的 SDK , 去思考他们为什么这么设计, 为什么和开源的有所不同,有的时候开始没想明白,但模仿得多了,好像也慢慢懂了。

要紧的是果敢地迈出第一步,对与错先都不管,自古就没有把一切都设计好再开步的事。

别想把一切都弄清楚,再去走路。鲁莽者要学会思考,善思者要克服的是犹豫。

——史铁生

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
7月前
|
监控 物联网 开发工具
MQTT常见问题之MQTT云端sdk消费者 出现重复消费如何解决
MQTT(Message Queuing Telemetry Transport)是一个轻量级的、基于发布/订阅模式的消息协议,广泛用于物联网(IoT)中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总:
|
7月前
|
消息中间件 物联网 网络安全
MQTT常见问题之调用.net sdk 报“不知道这样的主机”如何解决
MQTT(Message Queuing Telemetry Transport)是一个轻量级的、基于发布/订阅模式的消息协议,广泛用于物联网(IoT)中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总:
|
6月前
|
消息中间件 负载均衡 网络性能优化
消息队列 MQ产品使用合集之 终端sdk和云端sdk的区别是什么
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
7月前
|
消息中间件 网络协议 开发工具
MQ产品使用合集之rocketmq5.x只有tcp接入点吗,python sdk需要http接入点,请问怎么使用
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
224 2
|
消息中间件 Arthas 监控
一次RocketMQ ons SDK Bug导致消息不断堆积到重试队列的案例分析
一次RocketMQ ons SDK Bug导致消息不断堆积到重试队列的案例分析
474 1
|
消息中间件 自然语言处理 Cloud Native
RocketMQ 多语言 SDK 开源贡献召集令
我们欢迎任何形式的贡献,包括且不限于新 feature、bugfix、代码优化、生态集成、测试工作、文档撰写。更加欢迎能够认领一个完整的特定语言实现的同学!不要犹豫,欢迎大家以 issue/pull request 的形式将你的想法反馈到社区,一起来建设更好的 RocketMQ!
310 0
RocketMQ 多语言 SDK 开源贡献召集令
|
消息中间件 运维 Cloud Native
RocketMQ 5.0 API 与 SDK 的演进
RocketMQ 5.0 SDK 采用了全新的 API,使用 gRPC 作为通信层的实现,并在可观测性上做了很大幅度的提升。
466 0
RocketMQ 5.0 API 与  SDK 的演进
|
消息中间件 存储 缓存
全面升级 —— Apache RocketMQ 5.0 SDK 的新面貌
长久以来,RocketMQ 易于部署、高性能、高可用的架构,支撑了数十年来集团内外海量的业务场景。时至今日,为了迎接如今云原生时代的新挑战,我们重磅推出了 RocketMQ 5.0 新架构。
306 0
全面升级 —— Apache RocketMQ 5.0 SDK 的新面貌
|
2月前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
90 5
|
2月前
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
85 7

热门文章

最新文章