AI 助理
备案控制台
开发者社区 Java技术进阶 文章 正文

RocketMq 订阅分组创建和删除

2020-07-09 5487
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
数据可视化DataV，5个大屏 1个月
可视分析地图（DataV-Atlas），3 个项目，100M 存储空间
简介： 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版分组
云消息队列 MQ订阅分组
消息队列RocketMQ版订阅
相关实践学习
消息队列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天前
|
消息中间件 测试技术
通过轻量消息队列（原MNS）主题HTTP订阅+ARMS实现自定义数据多渠道告警
轻量消息队列（原MNS）以其简单队列模型、轻量化协议及按量后付费模式，成为阿里云产品间消息传输首选。本文通过创建主题、订阅、配置告警集成等步骤，展示了该产品在实际应用中的部分功能，确保消息的可靠传输。
刘天棋
27 2
勇哥Java实战
|
消息中间件 存储 负载均衡
两个实验让我彻底弄懂了「订阅关系一致」
这篇文章，笔者想聊聊 RocketMQ 最佳实践之一：**保证订阅关系一致**。 订阅关系一致指的是同一个消费者 Group ID 下所有 Consumer 实例所订阅的 Topic 、Tag 必须完全一致。 如果订阅关系不一致，消息消费的逻辑就会混乱，甚至导致消息丢失。
勇哥Java实战
865 1
两个实验让我彻底弄懂了「订阅关系一致」
wljslmz
|
传感器 负载均衡 物联网
MQTT v5共享订阅是怎么回事？如何使用共享订阅提高消息订阅的灵活性和可伸缩性？
MQTT v5共享订阅是怎么回事？如何使用共享订阅提高消息订阅的灵活性和可伸缩性？
wljslmz
477 1
真的很搞笑
|
6月前
|
消息中间件 Java RocketMQ
MQ产品使用合集之在同一个 Java 进程内建立三个消费对象并设置三个消费者组订阅同一主题和标签的情况下，是否会发生其中一个消费者组无法接收到消息的现象
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
真的很搞笑
121 1
小Lee
|
消息中间件 安全 Go
动态订阅时 rocketmq-client-go 代码有map并发bug
动态订阅时 rocketmq-client-go 代码有map并发bug
小Lee
65 2
wljslmz
|
6月前
|
传感器 监控 网络协议
MQTT 发布、订阅模式介绍
【2月更文挑战第17天】
wljslmz
429 6
MQTT 发布、订阅模式介绍
三分钟热度的鱼
|
6月前
|
存储 负载均衡 安全
MQTT常见问题之MQTT使用共享订阅失败如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
464 2
极客李华
|
6月前
|
消息中间件 Java
RabbitMQ中的消息发布-订阅模式是什么？如何实现？
RabbitMQ中的消息发布-订阅模式是什么？如何实现？
极客李华
170 0
EMQ
|
存储 缓存 网络性能优化
MQTT 订阅选项的使用
在本文中，我们将重点介绍在 MQTT 中哪些订阅选项可供我们使用，以及它们的使用方法。
EMQ
247 0
MQTT 订阅选项的使用
EMQ
|
存储 Python
MQTT 订阅标识符详解
本文介绍了 MQTT 5.0 中的订阅标识符，并通过一个实例演示了如何使用订阅标识符实现更高效的消息处理。
EMQ
199 0
MQTT 订阅标识符详解

Java技术进阶

热门文章

最新文章

  • 1
    2018最新蚂蚁金服面试题：Docker+秒杀设计+RocketMQ+亿级数据设计
  • 2
    RocketMQ与MYSQL事务消息整合
  • 3
    RocketMQ：一个纯java的开源消息中间件--开发测试环境搭建
  • 4
    喜讯！RocketMQ成为Apache软件基金会顶级开源项目
  • 5
    自顶向下学习 RocketMQ（九）：回溯消费
  • 6
    开篇：在Spring生态中玩转RocketMQ
  • 7
    Docker 安装 RocketMQ 并结合 SpringBoot 使用实例
  • 8
    RocketMQ 客户端负载均衡机制详解及最佳实践
  • 9
    在华为鲲鹏OpenEuler20.03系统上安装RocketMQ
  • 10
    消息队列RocketMQ版：消费异常运维排查体验
  • 1
    win10 安装RabbitMQ的步骤--和报错解决
    87
  • 2
    在 Windows 平台搭建 MQTT 服务
    115
  • 3
    mq报错abbit@syld36: * connected to epmd (port 4369) on syld36 * epmd reports node ‘rabbit‘ uses po
    73
  • 4
    Spring Boot与RabbitMQ的集成
    88
  • 5
    Docker部署RabbitMQ消息中间件
    277
  • 6
    JAVA面试之MQ
    70
  • 7
    EMQX 与 MQTT: AI 大模型时代的分布式数据中枢
    239
  • 8
    如何在Java中使用RabbitMQ
    248
  • 9
    【RocketMQ系列十四】RocketMQ中消息堆积如何处理
    1374
  • 10
    【RocketMQ系列十三】RocketMQ的集群核心概念之消费重试&死信队列&幂等消息的出现以及处理
    131

    • 相关课程

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

    • 相关电子书

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

    • 相关实验场景

    更多
  • 通过轻量消息队列（原MNS）主题HTTP订阅+ARMS实现自定义数据多渠道告警
  • 快速体验云消息队列RocketMQ版
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
    • 下一篇
    无影云桌面