备案控制台
探索云世界
开发者社区 Java技术进阶 文章 正文

RocketMq 订阅分组创建和删除

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

开篇

  • 这个系列主要用以分析mqadmin常见的比较核心的几个命令，主要包括订阅分组和topic的创建和删除、Topic的权限变更。
  • 这篇文章主要是用来分析订阅分组创建和删除。


订阅分组的作用

public class PullMessageProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {

    private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
        throws RemotingCommandException {

        RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
        final PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) response.readCustomHeader();
        final PullMessageRequestHeader requestHeader =
            (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
        response.setOpaque(request.getOpaque());

        // 查找SubscriptionGroupConfig，如果不存在就会直接返回错误
        SubscriptionGroupConfig subscriptionGroupConfig =
            this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup());
        if (null == subscriptionGroupConfig) {
            response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
            response.setRemark(String.format("subscription group [%s] does not exist, %s", requestHeader.getConsumerGroup(), FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST)));
            return response;
        }
}
  • SubscriptionGroupConfig可以理解为消费分组和broker之间的契约，如果broker不存在消费分组的订阅关系，那么该消费分组就无法从该broker消费任何消息。


创建订阅分组

updateSubGroup

usage: mqadmin updateSubGroup [-a <arg>] [-b <arg>] [-c <arg>] [-d <arg>] -g <arg> [-h] [-i <arg>] [-m <arg>]
       [-n <arg>] [-q <arg>] [-r <arg>] [-s <arg>] [-w <arg>]
 -a,--notifyConsumerIdsChanged <arg>       notify consumerId changed
 -b,--brokerAddr <arg>                     create subscription group to which broker
 -c,--clusterName <arg>                    create subscription group to which cluster
 -d,--consumeBroadcastEnable <arg>         broadcast
 -g,--groupName <arg>                      consumer group name
 -h,--help                                 Print help
 -i,--brokerId <arg>                       consumer from which broker id
 -m,--consumeFromMinEnable <arg>           from min offset
 -n,--namesrvAddr <arg>                    Name server address list, eg: 192.168.0.1:9876;192.168.0.2:9876
 -q,--retryQueueNums <arg>                 retry queue nums
 -r,--retryMaxTimes <arg>                  retry max times
 -s,--consumeEnable <arg>                  consume enable
 -w,--whichBrokerWhenConsumeSlowly <arg>   which broker id when consume slowly
  • updateSubGroup负责创建订阅分组。
  • updateSubGroup的命令参数中通过--groupName来指定consumer group name。


UpdateSubGroupSubCommand

public class UpdateSubGroupSubCommand implements SubCommand {

    @Override
    public String commandName() {
        return "updateSubGroup";
    }

    @Override
    public String commandDesc() {
        return "Update or create subscription group";
    }

    @Override
    public Options buildCommandlineOptions(Options options) {
        Option opt = new Option("b", "brokerAddr", true, "create subscription group to which broker");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("c", "clusterName", true, "create subscription group to which cluster");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("g", "groupName", true, "consumer group name");
        opt.setRequired(true);
        options.addOption(opt);

        opt = new Option("s", "consumeEnable", true, "consume enable");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("m", "consumeFromMinEnable", true, "from min offset");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("d", "consumeBroadcastEnable", true, "broadcast");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("q", "retryQueueNums", true, "retry queue nums");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("r", "retryMaxTimes", true, "retry max times");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("i", "brokerId", true, "consumer from which broker id");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("w", "whichBrokerWhenConsumeSlowly", true, "which broker id when consume slowly");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("a", "notifyConsumerIdsChanged", true, "notify consumerId changed");
        opt.setRequired(false);
        options.addOption(opt);

        return options;
    }

    @Override
    public void execute(final CommandLine commandLine, final Options options,
        RPCHook rpcHook) throws SubCommandException {
        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);

        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));

        try {
            SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
            subscriptionGroupConfig.setConsumeBroadcastEnable(false);
            subscriptionGroupConfig.setConsumeFromMinEnable(false);

            // groupName
            subscriptionGroupConfig.setGroupName(commandLine.getOptionValue('g').trim());

            // consumeEnable
            if (commandLine.hasOption('s')) {
                subscriptionGroupConfig.setConsumeEnable(Boolean.parseBoolean(commandLine.getOptionValue('s')
                    .trim()));
            }

            // consumeFromMinEnable
            if (commandLine.hasOption('m')) {
                subscriptionGroupConfig.setConsumeFromMinEnable(Boolean.parseBoolean(commandLine
                    .getOptionValue('m').trim()));
            }

            // consumeBroadcastEnable
            if (commandLine.hasOption('d')) {
                subscriptionGroupConfig.setConsumeBroadcastEnable(Boolean.parseBoolean(commandLine
                    .getOptionValue('d').trim()));
            }

            // retryQueueNums
            if (commandLine.hasOption('q')) {
                subscriptionGroupConfig.setRetryQueueNums(Integer.parseInt(commandLine.getOptionValue('q')
                    .trim()));
            }

            // retryMaxTimes
            if (commandLine.hasOption('r')) {
                subscriptionGroupConfig.setRetryMaxTimes(Integer.parseInt(commandLine.getOptionValue('r')
                    .trim()));
            }

            // brokerId
            if (commandLine.hasOption('i')) {
                subscriptionGroupConfig.setBrokerId(Long.parseLong(commandLine.getOptionValue('i').trim()));
            }

            // whichBrokerWhenConsumeSlowly
            if (commandLine.hasOption('w')) {
                subscriptionGroupConfig.setWhichBrokerWhenConsumeSlowly(Long.parseLong(commandLine
                    .getOptionValue('w').trim()));
            }

            // notifyConsumerIdsChanged
            if (commandLine.hasOption('a')) {
                subscriptionGroupConfig.setNotifyConsumerIdsChangedEnable(Boolean.parseBoolean(commandLine
                    .getOptionValue('a').trim()));
            }

            if (commandLine.hasOption('b')) {
                String addr = commandLine.getOptionValue('b').trim();

                defaultMQAdminExt.start();

                defaultMQAdminExt.createAndUpdateSubscriptionGroupConfig(addr, subscriptionGroupConfig);
                System.out.printf("create subscription group to %s success.%n", addr);
                System.out.printf("%s", subscriptionGroupConfig);
                return;

            } else if (commandLine.hasOption('c')) {
                String clusterName = commandLine.getOptionValue('c').trim();

                defaultMQAdminExt.start();
                Set<String> masterSet =
                    CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
                for (String addr : masterSet) {
                    try {
                        defaultMQAdminExt.createAndUpdateSubscriptionGroupConfig(addr, subscriptionGroupConfig);
                        System.out.printf("create subscription group to %s success.%n", addr);
                    } catch (Exception e) {
                        e.printStackTrace();
                        Thread.sleep(1000 * 1);
                    }
                }
                System.out.printf("%s", subscriptionGroupConfig);
                return;
            }

            ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
        } catch (Exception e) {
            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
        } finally {
            defaultMQAdminExt.shutdown();
        }
    }
}
  • UpdateSubGroupSubCommand是处理updateSubGroup命令的函数。
  • 核心逻辑在于组装subscriptionGroupConfig对象，subscriptionGroupConfig就是订阅分组的变量。
  • UpdateSubGroupSubCommand可以指定broker或者指定cluster，指定broker就在指定的broker上创建订阅分组，指定cluster就在cluster下的所有broker上创建订阅分组。


MQClientAPIImpl

public class MQClientAPIImpl {

    public void createSubscriptionGroup(final String addr, final SubscriptionGroupConfig config,
        final long timeoutMillis)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP, null);

        byte[] body = RemotingSerializable.encode(config);
        request.setBody(body);

        RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
            request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                return;
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());

    }
}
  • 创建订阅分组的RequestCode为UPDATE_AND_CREATE_SUBSCRIPTIONGROUP。


AdminBrokerProcessor

public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {

    private RemotingCommand updateAndCreateSubscriptionGroup(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);

        log.info("updateAndCreateSubscriptionGroup called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        SubscriptionGroupConfig config = RemotingSerializable.decode(request.getBody(), SubscriptionGroupConfig.class);
        if (config != null) {
            this.brokerController.getSubscriptionGroupManager().updateSubscriptionGroupConfig(config);
        }

        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
  • updateAndCreateSubscriptionGroup在broker负责处理消费分组的创建逻辑。
  • SubscriptionGroupManager负责保存订阅分组。


SubscriptionGroupManager

public class SubscriptionGroupManager extends ConfigManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    private final ConcurrentMap<String, SubscriptionGroupConfig> subscriptionGroupTable =
        new ConcurrentHashMap<String, SubscriptionGroupConfig>(1024);
    private final DataVersion dataVersion = new DataVersion();

    public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config) {
        SubscriptionGroupConfig old = this.subscriptionGroupTable.put(config.getGroupName(), config);
        if (old != null) {
            log.info("update subscription group config, old: {} new: {}", old, config);
        } else {
            log.info("create new subscription group, {}", config);
        }

        this.dataVersion.nextVersion();

        this.persist();
    }
}
  • SubscriptionGroupManager的subscriptionGroupTable负责保存订阅分组。


删除订阅分组

deleteSubGroup

usage: mqadmin deleteSubGroup [-b <arg>] [-c <arg>] -g <arg> [-h] [-n <arg>]
 -b,--brokerAddr <arg>    delete subscription group from which broker
 -c,--clusterName <arg>   delete subscription group from which cluster
 -g,--groupName <arg>     subscription group name
 -h,--help                Print help
 -n,--namesrvAddr <arg>   Name server address list, eg: 192.168.0.1:9876;192.168.0.2:9876
  • deleteSubGroup负责删除订阅分组。
  • deleteSubGroup命令参数中通过--groupName来指定consumer group name。


DeleteSubscriptionGroupCommand

public class DeleteSubscriptionGroupCommand implements SubCommand {
    @Override
    public String commandName() {
        return "deleteSubGroup";
    }

    @Override
    public String commandDesc() {
        return "Delete subscription group from broker.";
    }

    @Override
    public Options buildCommandlineOptions(Options options) {
        Option opt = new Option("b", "brokerAddr", true, "delete subscription group from which broker");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("c", "clusterName", true, "delete subscription group from which cluster");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("g", "groupName", true, "subscription group name");
        opt.setRequired(true);
        options.addOption(opt);

        return options;
    }

    @Override
    public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
        DefaultMQAdminExt adminExt = new DefaultMQAdminExt(rpcHook);
        adminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
        try {
            // groupName
            String groupName = commandLine.getOptionValue('g').trim();

            if (commandLine.hasOption('b')) {
                String addr = commandLine.getOptionValue('b').trim();
                adminExt.start();

                adminExt.deleteSubscriptionGroup(addr, groupName);
                System.out.printf("delete subscription group [%s] from broker [%s] success.%n", groupName,
                    addr);

                return;
            } else if (commandLine.hasOption('c')) {
                String clusterName = commandLine.getOptionValue('c').trim();
                adminExt.start();

                Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
                for (String master : masterSet) {
                    adminExt.deleteSubscriptionGroup(master, groupName);
                    System.out.printf(
                        "delete subscription group [%s] from broker [%s] in cluster [%s] success.%n",
                        groupName, master, clusterName);
                }

                try {
                    DeleteTopicSubCommand.deleteTopic(adminExt, clusterName, MixAll.RETRY_GROUP_TOPIC_PREFIX
                        + groupName);
                    DeleteTopicSubCommand.deleteTopic(adminExt, clusterName, MixAll.DLQ_GROUP_TOPIC_PREFIX
                        + groupName);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return;
            }

            ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
        } catch (Exception e) {
            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
        } finally {
            adminExt.shutdown();
        }
    }
}
  • DeleteSubscriptionGroupCommand是处理deleteSubGroup命令的函数。
  • 核心逻辑在于往指定的broker删除指定的consumer groupName。
  • DeleteSubscriptionGroupCommand可以指定broker或者指定cluster，指定broker就在指定的broker上删除订阅分组，指定cluster就在cluster下的所有broker上删除订阅分组。
  • 删除集群下订阅分组即消费分组，同时需要删除消费分组的重试Topic和死信Topic。


deleteSubscriptionGroup

public class MQClientAPIImpl {

    public void deleteSubscriptionGroup(final String addr, final String groupName, final long timeoutMillis)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        DeleteSubscriptionGroupRequestHeader requestHeader = new DeleteSubscriptionGroupRequestHeader();
        requestHeader.setGroupName(groupName);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, requestHeader);

        RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
            request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                return;
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }
}
  • 删除订阅分组的RequestCode为DELETE_SUBSCRIPTIONGROUP。


AdminBrokerProcessor

public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {

    private RemotingCommand deleteSubscriptionGroup(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        DeleteSubscriptionGroupRequestHeader requestHeader =
            (DeleteSubscriptionGroupRequestHeader) request.decodeCommandCustomHeader(DeleteSubscriptionGroupRequestHeader.class);

        log.info("deleteSubscriptionGroup called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        this.brokerController.getSubscriptionGroupManager().deleteSubscriptionGroupConfig(requestHeader.getGroupName());

        if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats()) {
            this.brokerController.getBrokerStatsManager().onGroupDeleted(requestHeader.getGroupName());
        }
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
}

deleteSubscriptionGroup在broker负责处理订阅分组即消费分组的删除逻辑。
SubscriptionGroupManager负责删除订阅分组。


SubscriptionGroupManager

public class SubscriptionGroupManager extends ConfigManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    private final ConcurrentMap<String, SubscriptionGroupConfig> subscriptionGroupTable =
        new ConcurrentHashMap<String, SubscriptionGroupConfig>(1024);
    private final DataVersion dataVersion = new DataVersion();
    private transient BrokerController brokerController;

    public void deleteSubscriptionGroupConfig(final String groupName) {
        SubscriptionGroupConfig old = this.subscriptionGroupTable.remove(groupName);
        if (old != null) {
            log.info("delete subscription group OK, subscription group:{}", old);
            this.dataVersion.nextVersion();
            this.persist();
        } else {
            log.warn("delete subscription group failed, subscription groupName: {} not exist", groupName);
        }
    }
}
  • 从SubscriptionGroupManager的subscriptionGroupTable删除订阅分组。
文章标签：
云消息队列 MQ
DataV数据可视化
消息中间件
RocketMQ
关键词：
云消息队列 MQ订阅
消息队列RocketMQ版分组
消息队列RocketMQ版订阅
相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv，使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
晴天哥
目录
相关文章
章为忠学架构
|
消息中间件 Java 物联网
一文搞懂MQTT，如何在SpringBoot中使用MQTT实现消息的订阅和发布
之前介绍了RabbitMQ以及如何在SpringBoot项目中整合使用RabbitMQ，看过的朋友都说写的比较详细，希望再总结一下目前比较流行的MQTT。所以接下来，就来介绍什么MQTT？它在IoT中有着怎样的作用？如何在项目中使用MQTT？
章为忠学架构
8518 5
一文搞懂MQTT，如何在SpringBoot中使用MQTT实现消息的订阅和发布
Wesky
|
监控 物联网 API
【.NET+MQTT】.NET6 环境下实现MQTT通信，以及服务端、客户端的双边消息订阅与发布的代码演示
MQTT广泛应用于工业物联网、智能家居、各类智能制造或各类自动化场景等。MQTT是一个基于客户端-服务器的消息发布/订阅传输协议，在很多受限的环境下，比如说机器与机器通信、机器与物联网通信等。好了，科普的废话不多说，下面直接通过.NET环境来实现一套MQTT通信demo，实现服务端与客户端的双边消息发布与订阅的功能和演示。
Wesky
940 0
【.NET+MQTT】.NET6 环境下实现MQTT通信，以及服务端、客户端的双边消息订阅与发布的代码演示
wljslmz
|
5月前
|
传感器 负载均衡 物联网
MQTT v5共享订阅是怎么回事？如何使用共享订阅提高消息订阅的灵活性和可伸缩性？
MQTT v5共享订阅是怎么回事？如何使用共享订阅提高消息订阅的灵活性和可伸缩性？
wljslmz
114 1
wljslmz
|
1月前
|
传感器 监控 网络协议
MQTT 发布、订阅模式介绍
【2月更文挑战第17天】
wljslmz
56 6
MQTT 发布、订阅模式介绍
小Lee
|
6月前
|
消息中间件 安全 Go
动态订阅时 rocketmq-client-go 代码有map并发bug
动态订阅时 rocketmq-client-go 代码有map并发bug
小Lee
36 2
勇哥Java实战
|
6月前
|
消息中间件 存储 负载均衡
两个实验让我彻底弄懂了「订阅关系一致」
这篇文章，笔者想聊聊 RocketMQ 最佳实践之一：**保证订阅关系一致**。 订阅关系一致指的是同一个消费者 Group ID 下所有 Consumer 实例所订阅的 Topic 、Tag 必须完全一致。 如果订阅关系不一致，消息消费的逻辑就会混乱，甚至导致消息丢失。
勇哥Java实战
586 1
两个实验让我彻底弄懂了「订阅关系一致」
极客李华
|
3月前
|
消息中间件 Java
RabbitMQ中的消息发布-订阅模式是什么？如何实现？
RabbitMQ中的消息发布-订阅模式是什么？如何实现？
极客李华
46 0
EMQ
|
9月前
|
存储 缓存 网络性能优化
MQTT 订阅选项的使用
在本文中，我们将重点介绍在 MQTT 中哪些订阅选项可供我们使用，以及它们的使用方法。
EMQ
169 0
MQTT 订阅选项的使用
EMQ
|
9月前
|
存储 Python
MQTT 订阅标识符详解
本文介绍了 MQTT 5.0 中的订阅标识符，并通过一个实例演示了如何使用订阅标识符实现更高效的消息处理。
EMQ
131 0
MQTT 订阅标识符详解
EMQ
|
消息中间件 存储 网络性能优化
MQTT 客户端出现连接订阅等问题时如何排查？
这是一期EMQX社区专题FAQ，我们整理了近期社区中关注度较高的问题，在这里进行统一汇总解答。
EMQ
337 0

Java技术进阶

热门文章

最新文章

  • 1
    RocketMQ 流数据库解析：如何实现一体化流处理？
  • 2
    小白版的springboot中集成mqtt服务（超级无敌详细），实现不了掐我头！！！
  • 3
    RocketMQ 面试题及答案整理，最新面试题
  • 4
    RabbitMQ 面试题及答案整理，最新面试题
  • 5
    MQTT常见问题之mqtt 连接一直显示 Not authorized to connect如何解决
  • 6
    使用阿里云 mqtt serverless 版本超低成本快速实现 webscoket 长链接服务器
  • 7
    深入解析：选择最佳C++ MQTT库的综合指南
  • 8
    MQTT常见问题之发布MQTT主题消息失败如何解决
  • 9
    Docker安装部署RabbitMQ & 密码修改 &创建用户及角色
  • 10
    RocketMQ的下载与安装（全网最细保姆级别教学）
  • 1
    消息队列RocketMQ版：基础消息收发功能体验
    77
  • 2
    消息中间件第八讲：消息队列 RocketMQ 版实战、集群及原理
    96
  • 3
    消息队列RocketMQ版：消费异常运维排查体验
    821
  • 4
    消息队列RocketMQ版：定时消息通知功能体验
    564
  • 5
    消息队列RocketMQ版：基础消息收发功能体验
    392
  • 6
    【名师代练】带你玩转 RocketMQ，角逐「RocketMQ 首席评测官」
    333
  • 7
    聊聊RocketMQ名字服务
    81
  • 8
    微服务 RocketMQ-延时消息 消息过滤 管控台搜索问题
    85
  • 9
    RocketMQ-消息消费模式 顺序消费
    156
  • 10
    消息中间件-RocketMQ入门 消息发送的三种方式
    111

    • 相关课程

    更多
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • 基于 RocketMQ+ Knative 驱动云原生 Ser
  • 基于 RocketMQ + Knative 驱动云原生 Serverless 应用
  • RocketMQ Operator-K8s平台自动化部署工具

    • 相关实验场景

    更多
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 基于ROS的消息发布与订阅
  • 部署RabbitMQ
  • RocketMQ一站式入门使用
    • 下一篇
    阿里云oss简介和使用流程
    为什么选择阿里云什么是云计算全球基础设施技术领先稳定可靠安全合规分析师报告
    产品和定价全部产品免费试用产品动态产品定价价格计算器云上成本管理
    解决方案技术解决方案
    文档与社区文档开发者社区天池大赛培训与认证
    权益中心免费试用高校计划企业扶持计划推荐返现计划
    支持与服务基础服务企业增值服务迁云服务官网公告健康看板信任中心
    关注阿里云
    关注阿里云公众号或下载阿里云APP，关注云资讯，随时随地运维管控云服务
    阿里云APP阿里云微信
    联系我们：4008013260
    法律声明Cookies政策廉正举报安全举报联系我们加入我们
    阿里巴巴集团淘宝网天猫全球速卖通阿里巴巴国际交易市场1688阿里妈妈飞猪阿里云计算AliOS万网高德UC友盟优酷钉钉支付宝达摩院淘宝海外阿里云盘饿了么
    © 2009-2024 Aliyun.com 版权所有 增值电信业务经营许可证： 浙B2-20080101 域名注册服务机构许可： 浙D3-20210002 京D3-20220015
    浙公网安备 33010602009975号浙B2-20080101-4