全局堆积 在使用RocketMQ时，可能会遇到消息堆积的情况，即未能及时处理和消费的消息大量积压在消息队列中。 造成消息堆积的原因可能是消费者处理消息的速度不够快，或者消费者出现故障导致无法正常消费消息。当消息堆积发生时，可能会导致系统的稳定性下降，甚至影响整个应用的正常运行。 分布式消息服务RocketMQ标识该集群下，所有topic的消费堆积情况，列表按倒序排列了堆积量最大的前20个topic的堆积情况。 全局消息 分布式消息服务RocketMQ标识该集群，所有topic的生产情况，按现有消息量倒序排列。

本章节介绍了分布式消息服务RocketMQ使用过程中的约束和限制。 本章主要为您介绍分布式消息服务RocketMQ使用过程中的一些限制。 表1 分布式消息服务RocketMQ使用限制明细 限制项 约束与限制 创建Topic的数量 Topic的数量根据实例规格确定，不支持修改。 当Topic的数量达到上限后，您无法继续创建Topic。 创建消费组的数量 消费组的数量根据实例规格确定，不支持修改。 当消费组的数量达到上限后，您无法继续创建消费组。 消息大小 生产消息的最大长度为4MB，其中，消息属性大小均不能超过16KB。消息大小不支持修改。 消息大小超过限制会导致消息发送失败。 消息存储时长 消息默认保留时间为48小时，支持修改，最大存储时长为720小时，超过保留时间会被自动删除。 消费位点重置 支持重置消费2天内任意时间点的消息。 定时消息的延时时长 最大延时时长为1年，不支持修改。 支持1年内任意时间的定时消息。 RequestReply机制 不支持此机制 修改配置参数 不支持调用开源接口修改配置参数

本章节介绍了分布式消息服务RocketMQ常见的应用场景，以及应用架构图。 电商应用 电商应用存在诸多难题： 电商场景中通常会涉及到订单、支付和通知等等场景的业务处理。业务链通常都是多个系统相互协作完成一次作业，上层服务强依赖于下层服务，上层服务的性能会强依赖于下层服务，当业务链过深，则会严重影响外层服务的性能和用户体验。 在电商促销活动中，需要用户拥有订阅通知的能力。 分布式消息服务RocketMQ为搭建电商系统提供了更多的选择。 分布式消息服务RocketMQ可以解除多个业务系统之间的耦合度，提升各系统的处理能力和响应速度。 分布式消息服务RocketMQ提供的定时、延迟等能力，满足需要订阅通知的电商场景。 金融场景 相较于传统金融场景，互联网金融需要能及时响应互联网的快速变化。传统金融场景中共性的部分需要被抽取出来，进行细化为各个不同的微服务模块，构成基础的业务中台，基于业务中台提供的基本能力，用户可以在上层快速的构建多样的业务能力。而分布式消息服务RocketMQ因其优秀的解耦链接能力，增强了各微服务模块的处理能力和响应速度，在业务中台内扮演着不可或缺的角色。

本文介绍如何在事件总线EventBridge管理控制台中添加分布式消息服务RocketMQ类型的自定义事件源。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 创建自定义总线。 分布式消息服务RocketMQ 开通分布式消息服务RocketMQ。 创建实例，并创建对应主题与消费组。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件总线。 3. 在事件总线页面，单击目标总线名称。 4. 在左侧导航栏，单击事件源。 5. 在事件源 页面，单击添加事件源。 6. 在添加自定义事件源 面板，输入名称 和描述 ，事件提供方 选择分布式消息服务RocketMQ，选择实例、主题等配置然后单击确认，如图1所示。 图1 创建事件源时事件提供方选择分布式消息服务RocketMQ 参数说明 参数 说明 示例 名称 事件源名。 source1 实例名称 选择前提条件中已创建的分布式消息服务RocketMQ版实例。 xxx Topic 选择当前实例中的Topic。 topic1 Group 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。仅支持从最早位点进行消费。 最早位点：从最初位点开始消费。 最早位点

分类 服务 命名空间 维度 监控指标参考文档 企业中间件 分布式消息服务 SYS.DMS 请参考右侧相关文档链接页面。

发送消息 参考如下示例代码。 import time from rocketmq.client import TransactionMQProducer, Message, TransactionStatus endpoint "${ENDPOINT}" 填写分布式消息服务RocketMQ控制台Namesrv接入点 accesskey "${ACCESSKEY}" 填写AccessKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID accesssecret "${SECRETKEY}" 填写SecretKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户密钥 topic "${TOPIC}" 填写Topic，在管理控制台创建 producergroup "${GROUP}" 生产者组group def transactioncheckercallback(msg, userargs): return TransactionStatus.COMMIT def transactionlocalexecute(msg, userargs): return TransactionStatus.UNKNOWN producer TransactionMQProducer(producergroup, transactioncheckercallback) producer.setnameserveraddress(endpoint) producer.setsessioncredentials(accesskey, accesssecret, "") producer.start() msg Message(topic) msg.setbody("Hello RocketMQ") ret producer.sendmessageintransaction(msg, transactionlocalexecute, None) print(ret.status, ret.msgid, ret.offset) while True: time.sleep(3600) 订阅消息 参考如下示例代码。 import time from rocketmq.client import PushConsumer, ConsumeStatus endpoint "${ENDPOINT}" 填写分布式消息服务RocketMQ控制台Namesrv接入点 accesskey "${ACCESSKEY}" 填写AccessKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID accesssecret "${SECRETKEY}" 填写SecretKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户密钥 topic "${TOPIC}" 填写Topic，在管理控制台创建 group "${GROUP}" 填写订阅组group，在管理控制台创建 def callback(msg): print(msg.id, msg.body) return ConsumeStatus.CONSUMESUCCESS consumer PushConsumer(group) consumer.setnameserveraddress(endpoint) consumer.setsessioncredentials(accesskey, accesssecret, "") consumer.subscribe(topic, callback) consumer.start() while True: time.sleep(3600) consumer.shutdown()

本章节主要介绍创建用户并授权使用分布式消息服务RabbitMQ。 如果您需要对您所拥有的分布式消息服务RabbitMQ服务进行精细的权限管理，您可以使用统一身份认证服务（Identity and Access Management，简称IAM），通过IAM，您可以： 根据企业的业务组织，在您的帐号中，给企业中不同职能部门的员工创建IAM用户，让员工拥有唯一安全凭证，并使用分布式消息服务RabbitMQ资源。 根据企业用户的职能，设置不同的访问权限，以达到用户之间的权限隔离。 将分布式消息服务RabbitMQ资源委托给更专业、高效的其他帐号或者云服务，这些帐号或者云服务可以根据权限进行代运维。 如果帐号已经能满足您的要求，不需要创建独立的IAM用户，您可以跳过本章节，不影响您使用分布式消息服务RabbitMQ服务的其它功能。 本章节为您介绍对用户授权的方法，操作流程如图1所示。 说明 分布式消息服务RabbitMQ服务的权限与策略基于分布式消息服务DMS，因此在IAM服务中为RabbitMQ分配用户与权限时，请选择并使用“DMS”的权限与策略。 前提条件 给用户组授权之前，请您了解用户组可以添加的分布式消息服务RabbitMQ系统策略，并结合实际需求进行选择，分布式消息服务RabbitMQ支持的系统策略及策略间的对比，请参见：权限管理。 示例流程 图1 给用户授权RabbitMQ权限流程 1. 创建用户组并授权 在IAM控制台创建用户组，并授予RabbitMQ的只读权限“DMS ReadOnlyAccess”。 2. 创建用户并加入用户组 在IAM控制台创建用户，并将其加入1中创建的用户组。 3. 用户登录并验证权限 新创建的用户登录控制台，切换至授权区域，验证权限： 在“服务列表”中选择分布式消息服务RabbitMQ，进入RabbitMQ实例主界面，单击右上角“购买RabbitMQ实例”，尝试购买RabbitMQ实例，如果无法购买RabbitMQ实例（假设当前权限仅包含DMS ReadOnlyAccess），表示“DMS ReadOnlyAccess”已生效。 在“服务列表”中选择云硬盘（假设当前策略仅包含DMS ReadOnlyAccess），若提示权限不足，表示“DMS ReadOnlyAccess”已生效。

本章节介绍分布式消息服务RabbitMQ的产品价格。 分布式消息服务RabbitMQ的产品价格，分为老规格和新规格两个类型，老规格是指“2vCPU4GB”格式的规格，新规格是指“rabbitmq.2u4g.cluster”格式的规格。自新规格正式上线之日起，用户新订购实例只能订购新规格、无法新订购老规格实例，但用户已订购的老规格实例，可正常使用，也可以进行规格变更。 RabbitMQ支持ARM架构的鲲鹏实例，其规格、价格与X86架构相同。 详细的产品价格如下： 一、老规格 分布式消息服务RabbitMQ老规格仅支持包年/包月计费模式，价格如下： 实例类型 实例规格 数量 包月价格（元/月） :::: 单机版 2C4G 1 278 单机版 4C8G 1 541 单机版 8C16G 1 1068 单机版 16C32G 1 1939 集群版 4C8G 3 3158 集群版 4C8G 5 4421 集群版 4C8G 7 5684 集群版 8C16G 3 6231 集群版 8C16G 5 8723 集群版 8C16G 7 11215 集群版 16C32G 3 12926 集群版 16C32G 5 18096 集群版 16C32G 7 23267 分布式消息服务RabbitMQ老规格的存储空间价格，同云硬盘单价一致。 分布式消息服务RabbitMQ老规格包周期订购，无包年优惠。

分布式消息服务RocketMQ版支持任意时间的定时消息，最大推迟时间可达到40天。 定时消息即生产者生产消息到分布式消息服务RocketMQ版后，消息不会立即被消费，而是延迟到设定的时间点后才会发送给消费者进行消费。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 1. 在命令行输入python，检查是否已安装Python。得到如下回显，说明Python已安装。 PS C:> python Python 3.9.9 (tags/v3.9.9:ccb0e6a, Nov 15 2021, 18:08:50) [MSC v.1929 64 bit (AMD64)] on win32 Type "help", "copyright", "credits" or "license" for more information. 如果未安装Python，请使用以下命令安装： pip install rocketmqclientpython 2. 安装librocketmq库和rocketmqclientpython。 说明 建议下载rocketmqclientcpp2.2.0，获取librocketmq库。 3. 将librocketmq.so添加到系统动态库搜索路径。 1. 查找librocketmq.so的路径。 find / name librocketmq.so 2. 将librocketmq.so添加到系统动态库搜索路径。 ln s /查找到的librocketmq.so路径/librocketmq.so /usr/lib sudo ldconfig 以下示例代码中的参数说明如下，请参考收集连接信息获取参数值。 GROUP：表示消费组名称。 ENDPOINT：表示实例连接地址和端口。 TOPIC：表示Topic名称。 发送消息 参考如下示例代码。 import datetime from rocketmq.client import Producer, Message from rocketmq.exceptions import RocketMQException endpoint "${ENDPOINT}" 填写分布式消息服务RocketMQ控制台Namesrv接入点 accesskey "${ACCESSKEY}" 填写AccessKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID accesssecret "${SECRETKEY}" 填写SecretKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户密钥 topic "${TOPIC}" 填写Topic，在管理控制台创建 producergroup "${GROUP}" 生产者组group 初始化生产者 producer Producer(producergroup) producer.setnamesrvaddr(endpoint) producer.setsessioncredentials(accesskey, accesssecret, "") 启动生产者 try: producer.start() except RocketMQException as e: print('start producer error:', e) exit(1) msg Message(topic) msg.setbody("Hello RocketMQ") delaytime 10 发送任意延迟消息，时间单位为毫秒，如下所示：消息将在10s后投递 delaytimestamp int((datetime.datetime.now() + datetime.timedelta(secondsdelaytime)).timestamp() 1000) msg.setproperty('STARTDELIVERTIME', str(delaytimestamp)) 发送消息 try: result producer.sendsync(msg) print('send result:', result) except RocketMQException as e: print('send message error:', e) producer.shutdown() exit(1) 关闭生产者实例，释放资源 producer.shutdown()

本节介绍了RabbitMQ元数据迁移的实践。 RabbitMQ元数据迁移指将用户线下实例的rabbitmq实例元数据迁移到线上实例。 背景信息 RabbitMQ集群元数据是指RabbitMQ集群的信息，包括User、Vhost、Queue、Exchange、Binding Key、Permission、Parameter等信息。RabbitMQ集群元数据存储于RabbitMQ集群的内部数据库，在集群的各个节点之间自动复制。集群中的每个节点都有自己的元数据副本。当某个节点的元数据变更时，所有节点的元数据都会同步更新。因此，集群的各个节点的元数据被导出时都是相同的。RabbitMQ集群元数据可以被导出成一份JSON文件，然后被导入另一个RabbitMQ集群，实现RabbitMQ集群元数据备份。 迁移元数据上云是指将开源RabbitMQ集群的元数据迁移到天翼云分布式消息服务RabbitMQ实例。分布式消息服务RabbitMQ是天翼云提供的全托管消息队列服务，兼容开源RabbitMQ。您可以将RabbitMQ集群元数据导出，然后导入分布式消息服务RabbitMQ实例，分布式消息服务RabbitMQ会根据成功导入的元数据在目标分布式消息服务RabbitMQ实例中创建对应的Vhost、Queue、Exchange、Binding，实现RabbitMQ集群元数据迁移上云。您可以将全部Vhost信息导入分布式消息服务RabbitMQ实例，也可以根据需要将某个Vhost信息导入分布式消息服务RabbitMQ实例中的Vhost。 迁移元数据 （1）在RabbitMQ WebUI页面查看，如图所示。 Overview视图中，点击“Download broker definitions”按钮下载集群元数据。得到rabbitmq元数据的json文件。 （2）上线rabbitmq实例导入元数据。 如上图所示，先点击选择文件，选择上一步骤导出的json文件，再点击导入配置。 注意：在线下的Rabbitmq集群中不能含有用户名rabbitmq，否则会被覆盖。用户名rabbitmq是天翼云Rabbitmq内部使用的管理账号。

本节介绍了分布式消息服务RabbitMQ退订操作 针对分布式消息服务RabbitMQ退订操作，具体请参见文档费用中心退订退订流程。

参数 说明 示例 实例 选择分布式消息服务Kafka实例。 instancexxx Topic 选择分布式消息服务Kafka实例的Topic。 topic1 消息体 选择消息体（Body）的内容，更多信息请参考 完整事件 消息键值 选择消息键值（Key）的内容，更多信息请参考 空

本节介绍分布式消息服务RocketMQ相关的产品规格,以便您正确理解和使用。 （1）以下适用于华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 节点。 注意 通用型规格已调整为白名单特性，如需了解该规格参数请联系技术支持。 ctgmq引擎已调整为白名单特性，如需了解该引擎请联系技术支持。 单机版实例面向用户体验和业务测试场景，无法保证性能和高可用。如果需要在生产环境使用RocketMQ实例，建议购买集群版实例。 天翼云分布式消息服务RocketMQ版产品规格由以下五个维度定义： 资源规格：定义使用的弹性云服务器的规格类型。 代理个数：即Broker数量，定义实例的规模，天翼云分布式消息RocketMQ每个Broker由一个Master节点（主节点）和一个Slave节点（备节点）组成，详细见产品架构。 存储容量：定义单个代理可以保存的存储容量。 单个代理TPS：定义单个代理的TPS性能。 单个代理消费组数上限：定义单个代理可以创建的消费组数量。 天翼云分布式消息服务RocketMQ支持的产品规格如下所示： 说明 TPS（Transaction per second）是指每秒可以生产消息和消费消息的总次数，可以理解为对应规格每秒生产消息和消费消息的总吞吐量。

介绍分布式消息服务RabbitMQ变更配置内容。 当需要处理大量消息时，RabbitMQ实例的扩容是一种常见的解决方案。扩容可以增加RabbitMQ集群的吞吐量、存储能力和高可用性。分布式消息服务RabbitMQ提供三类扩容方案，分别为节点、规格和磁盘扩容，更好满足用户不同场景下的扩容需求。 节点扩容：指向RabbitMQ集群中添加更多的节点以增加系统的吞吐量和可靠性。通过扩容，可以将消息的发送和消费负载分摊到更多的节点上，从而提高系统的并发处理能力。 规格扩容：指通过增加RabbitMQ的资源配置来提升系统的处理能力和性能。 磁盘扩容：指增加磁盘的存储容量，以满足更多消息的存储需求。

本文主要介绍分布式消息服务RabbitMQ的计费概述。 通过阅读本文，您可以快速了解分布式消息服务RabbitMQ的计费模式、计费项、产品价格等信息。 计费模式 分布式消息服务RabbitMQ提供包年/包月和按需计费两种计费模式，以满足不同场景下的用户需求。包年/包月是一种预付费模式，即先付费再使用，按照订单的购买周期进行结算，因此在购买之前，您必须确保账户余额充足。按需计费是一种后付费模式，即先使用再付费，按照RabbitMQ实例实际使用时长计费。关于两种计费模式的详细介绍请参见计费模式概述。 计费项 分布式消息服务RabbitMQ的计费项由实例费用和存储空间费用组成。了解每种计费项的计费因子、计费公式等信息，请参考计费项。 如需了解实际场景下的计费样例以及各计费项在不同计费模式下的费用计算过程，请参见计费样例。 产品价格 分布式消息服务RabbitMQ的产品价格，介绍了包年/包月和按需计费的价格明细。了解更多关于产品价格的信息，请参见产品价格。

概述 本文介绍使用 RocketMQ C++（4.x） 客户端 SDK，访问分布式消息服务RocketMQ，帮助您更好地理解消息收发的完整过程。 前置条件 安装gccc++ 4.8.2 及以上版本，需支持C++11。 安装cmake 2.8.0及以上版本。 安装automake 1.11.1及以上版本。 安装autoconf 2.65及以上版本。 安装libtool 2.2.6 及以上版本。 环境准备 需要在客户端环境安装 RocketMQClientCPP 库，根据官方文档进行安装即可： 安装 CPP 动态库，推荐使用 master 分支构建。 在项目中引入 RocketMQClientCPP 相关头文件及动态库，详见实例代码头文件。 使用g++命令获得可执行文件，如： g++ o xxxProducer xxxProducer.cpp lrocketmq lpthread lz ldl lrt 使用 C++ SDK 收发普通消息 发送普通消息 include include include include include "rocketmq/DefaultMQProducer.h" using namespace std; using namespace rocketmq; int main(){ DefaultMQProducer producer("groupname"); //填写分布式消息服务RocketMQ版的接入点 producer.setNamesrvAddr("your access point"); //填写分布式消息服务RocketMQ版的ak、sk producer.setSessionCredentials("ak", "sk", "channel"); producer.start(); int count 64; for (int i 0; i < count; ++i) { //填入主题名、tag名、消息body MQMessage msg("topicname", "TAG", "msg content"); try { SendResult sendResult producer.send(msg); std::cout << "SendResult:" << sendResult.getSendStatus() << ", Message ID: " << sendResult.getMsgId() << std::endl; thisthread::sleepfor(chrono::seconds(1)); } catch (MQException e) { std::cout << "ErrorCode: " << e.GetError() << " Exception:" << e.what() << std::endl; } } std::cout << "Send " << count << " messages OK, costs" << std::endl; producer.shutdown(); return 0; }

本文介绍分布式消息服务RabbitMQ的计费项。 计费说明 分布式消息服务RabbitMQ的计费项由实例费用和存储空间费用组成。具体内容如表1所示。 表1 分布式消息服务RabbitMQ计费项 计费项 计费项说明 适用的计费模式 计费公式 实例费用 计费因子：代理规格和代理数量 包年/包月、按需计费 实例规格单价 购买时长 实例规格的单价，请参见分布式消息服务RabbitMQ产品价格中的“实例价格”。 存储空间费用 计费因子：云硬盘类型、容量 包年/包月、按需计费 存储空间单价 购买时长 存储空间的单价，请参见分布式消息服务RabbitMQ产品价格中的“存储空间价格”。 计费示例 以包年/包月计费模式为例，假设您于2023/03/08 15:50:04在“广州4”区域购买了一个包年/包月RabbitMQ实例（规格：rabbitmq.2u4g.cluster3，总存储空间：超高I/O 300GB），计费资源包括实例费用（代理规格和代理数量），以及存储空间费用（超高I/O 300GB）。购买时长为一个月，并在到期前手动续费1个月，则： 第一个计费周期为：2023/03/08 15:50:04 ~ 2023/04/08 23:59:59 第二个计费周期为：2023/04/08 23:59:59 ~ 2023/05/08 23:59:59 图1给出了上述示例配置的费用计算过程。 说明 图中价格仅供参考，实际计算请以分布式消息服务RabbitMQ 图1 包年/包月RabbitMQ实例费用计算示例 按需计费模式下，各计费项的计费示例请参见计费样例。

本章节主要介绍如何创建分布式消息服务RabbitMQ实例。 RabbitMQ是一款基于AMQP协议的开源服务，用于在分布式系统中存储转发消息，服务器端用Erlang语言（支持高并发、分布式以及健壮的容错能力等特点）编写，支持多种语言的客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP、AJAX等。 AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层的一个开放标准协议。 前提条件 在创建RabbitMQ专享版实例前，需要保证存在可使用的虚拟私有云。创建虚拟私有云的方法，请参考《虚拟私有云用户指南》。 如果您已有虚拟私有云，可重复使用，不需要多次创建。 为RabbitMQ授权 如果RabbitMQ实例需要开启IPv6功能，在创建实例前，需要授予RabbitMQ操作VPC终端节点的权限，否则会导致创建失败。 在IAM控制台创建委托，委托参数如下，其他参数保持默认，创建委托的具体步骤请参见《统一身份认证服务 用户指南》的“委托 > 委托其他云服务管理资源”章节。 委托名称：dmsadmintrust 委托类型：云服务 云服务：分布式消息服务 持续时间：永久 委托将拥有所选策略：DMS VPCEndpointAccess 操作步骤 步骤1 登录管理控制台。 步骤2 在管理控制台右上角单击，选择区域资源池。 说明 此处请选择与您的应用服务相同的区域。 步骤3 在管理控制台左上角单击，选择“企业中间件 > 分布式消息服务>RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 步骤4 单击页面右上方的“购买RabbitMQ实例”。 每个项目默认最多可以创建100个RabbitMQ专享版实例，如果您想创建更多实例，请联系客服申请增加配额。 步骤5 选择“计费模式”、“区域”、“项目”和“可用区”。 步骤6 设置“实例名称”和“企业项目”。 步骤7 设置实例信息。 1. 版本：RabbitMQ的版本号，当前仅支持3.8.35。 2. 实例类型：支持“单机”和“集群”。 单机：表示部署一个RabbitMQ代理。 集群：表示部署多个RabbitMQ代理，实现高可靠的消息存储。 3. 鲲鹏实例：“创建鲲鹏架构实例”选框，默认不勾选是X86架构的实例，勾选之后是ARM架构的鲲鹏实例。 4. 规格：根据实际情况选择规格。 说明 为了保证服务的稳定可靠，RabbitMQ采用了默认的40%高水位配置。当内存占用率到达40%高水位后，会触发流控，生产者发送消息会被阻塞。为了避免高水位的产生，请及时消费积压在队列中的消息。 5. 代理个数：根据实例情况选择代理个数。 6. 存储空间：选择RabbitMQ实例的磁盘类型和存储空间总量。 单机实例的取值范围：100GB ~ 30000GB 集群实例的取值范围为：100GB 代理数 ~ 30000GB 代理数 7. 虚拟私有云：选择已经创建好的虚拟私有云和子网。 虚拟私有云可以为您的RabbitMQ专享实例构建隔离的、能自主配置和管理的虚拟网络环境。 说明 虚拟私有云和子网在RabbitMQ实例创建完成后，不支持修改。 子网开启IPv6后，RabbitMQ实例支持IPv6功能。 实例创建成功后，不支持修改IPv6开关。 使用内网IPv6方式连接实例需要通过VPC终端节点实现，使用期间会产生VPC终端节点的费用。 8. 安全组：在“安全组”下拉列表，可以选择已经创建好的安全组。 安全组是一组对弹性云主机的访问规则的集合，为同一个VPC内具有相同安全保护需求并相互信任的弹性云主机提供访问策略。 您可以单击右侧的“管理安全组”，跳转到网络控制台的“安全组”页面，查看或创建安全组。 图1 设置实例信息 步骤8 设置连接RabbitMQ实例的用户名和密码。 步骤9 设置实例购买时长。 当选择了“包年/包月”付费模式时，页面才显示“购买时长”参数，您需要根据业务需要选择。 步骤10 单击“更多配置”，设置更多相关信息。 1. 设置“公网访问”。 您可以选择是否打开公网访问开关。 如果选择了开启，表示访问RabbitMQ实例可以通过弹性IP访问。这时页面会显示“弹性IP地址”，在“弹性IP地址”区域，您可下拉选择已有的弹性IP。另外，您可单击右侧的“创建弹性IP”，跳转到网络控制台的弹性IP页面，购买弹性IP。 图2 购买RabbitMQ实例时公网访问设置 说明 公网访问与VPC内访问相比，可能存在网络丢包和抖动等情况，且访问时延有所增加，因此建议仅在业务开发测试阶段开启公网访问进行调试。 如果用户在虚拟私有云的服务页面手动解绑定或删除EIP，相应RabbitMQ实例的公网访问功能会自动关闭。 2. 设置“SSL”。 客户端连接实例时SSL认证的开关。开启SSL，则数据加密传输，安全性更高。 SSL开关在实例创建完成后不支持修改，请明确是否需要开启 。 3. 设置“标签”。 标签用于标识云资源，当您拥有相同类型的许多云资源时，可以使用标签按各种维度（例如用途、所有者或环境）对云资源进行分类。 如果您已经预定义了标签，在“标签键”和“标签值”中选择已经定义的标签键值对。另外，您可以单击右侧的“查看预定义标签”，系统会跳转到标签管理服务页面，查看已经预定义的标签，或者创建新的标签。 您也可以直接在“标签键”和“标签值”中设置标签。 当前每个RabbitMQ实例最多支持设置20个不同标签。 4. 设置实例的描述信息。 步骤11 填写完上述信息后，单击“立即购买”，进入“规格确认”页面。 步骤12 确认实例信息无误且阅读并同意《分布式消息服务RabbitMQ服务等级协议》、《天翼云分布式消息服务协议》后，提交请求。 步骤13 在实例列表页面查看实例是否创建成功。 创建实例大约需要3到15分钟，此时实例的“状态”为“创建中”。 当实例的“状态”变为“运行中”时，说明实例创建成功。 如果创建实例失败，在信息栏的“创建失败任务”中查看创建失败的实例。请删除创建失败的RabbitMQ实例，然后重新创建。如果重新创建仍然失败，请联系客服。

本文介绍了分布式消息服务RabbitMQ提供的SDK语言版本。 SDK列表 下表提供了分布式消息服务RabbitMQ支持的SDK列表。 编程语言 参考文档 :: Java 开发指南Java Python 开发指南Python

参数 说明 示例 实例 选择分布式消息服务Kafka实例。 instancexxx Topic 选择分布式消息服务Kafka实例的Topic。 topic1 消息体 选择消息体（Body）的内容，更多内容请参考 部分事件 plaintext $.data.value 消息键值 选择消息键值（Key）的内容，更多内容请参考 部分事件 plaintext $.data.value

分布式消息服务Kafka 当事件目标选择分布式消息服务Kafka时，事件目标的type值为kafka，eventTargets中的resourceKey字段中含义如下表所示。 resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务Kafka实例Id。 说明 实例Id可在分布式消息服务Kafka管理控制台实例详情页获取。 无 topic 是 CONSTANT Kafka主题。 无 value 是 CONSTANT ORIGINAL JSONPATH TEMPLATE 消息体。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 key 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息键值。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 请求示例： plaintext { "eventBusName": "test", "eventRuleName": "test", "desc": "", "filterPattern": "{}", "eventTargets": [ { "type": "kafka", "eventTargetName": "1gQfuPljs9MJsWM1eHYch", "params": [ { "resourceKey": "instanceId", "value": "356b3496b87xxxxxxxxfe7deef7fe35", "form": "CONSTANT" }, { "resourceKey": "topic", "value": "test2", "form": "CONSTANT" }, { "resourceKey": "value", "value": "", "form": "ORIGINAL" }, { "resourceKey": "key", "value": "", "form": "EMPTY" } ] } ] } 分布式消息服务RocketMQ 当事件目标选择分布式消息服务RocketMQ时，事件目标的type值为rocketmq，eventTargets中的resourceKey字段中含义如下表所示。 resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务RocketMQ实例ID。 说明 实例Id可在分布式消息服务RocketMQ管理控制台实例详情页获取。 无 topic 是 CONSTANT RocketMQ主题。 无 body 是 CONSTANT ORIGINAL JSONPATH TEMPLATE 消息体。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 keys 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息索引。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 properties 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息自定义属性。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 tags 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息标签。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 请求示例： plaintext { "eventBusName": "test", "eventRuleName": "test", "desc": "", "filterPattern": "{}", "eventTargets": [ { "type": "rocketmq", "eventTargetName": "lMVBoqi55L3Q6QaCKFWYA", "params": [ { "resourceKey": "instanceId", "value": "89ceb110331e4c968499744c2ccbdbcc", "form": "CONSTANT" }, { "resourceKey": "topic", "value": "TopicA", "form": "CONSTANT" }, { "resourceKey": "body", "value": "", "form": "ORIGINAL" }, { "resourceKey": "properties", "value": "", "form": "EMPTY" }, { "resourceKey": "keys", "value": "", "form": "EMPTY" }, { "resourceKey": "tags", "value": "", "form": "EMPTY" } ] } ] }

本章节介绍了天翼云分布式消息服务协议。 天翼云分布式消息服务协议

本文介绍分布式消息服务RocketMQ的退订内容。 针对天翼云分布式消息服务RocketMQ退订操作，具体请参考退订流程。

本文主要介绍分布式消息服务RabbitMQ的请求条件。 您可以在创建自定义策略时，通过添加“请求条件”（Condition元素）来控制策略何时生效。请求条件包括条件键和运算符，条件键表示策略语句的 Condition元素，分为全局级条件键和服务级条件键。全局级条件键（前缀为g:）适用于所有操作，服务级条件键（前缀为服务缩写，如dms:）仅适用于对应服务的操作。运算符与条件键一起使用，构成完整的条件判断语句。 分布式消息服务RabbitMQ通过IAM预置了一组条件键，例如，您可以先使用dms:ssl条件键检查RabbitMQ实例是否开启SSL，然后再允许执行操作。下表显示了适用于RabbitMQ服务特定的条件键。 表1 分布式消息服务RabbitMQ请求条件 RabbitMQ条件键 运算符 描述 dms:publicIP Bool Null 是否开启公网 dms:ssl Bool Null 是否开启SSL

本文主要介绍如何导出分布式消息服务RabbitMQ的实例。 操作场景 本章节指导您在控制台导出实例列表。 操作步骤 1. 登录管理控制台。 2. 在管理控制台右上角单击，选择区域。 说明 请选择RabbitMQ实例所在的区域。 3. 在管理控制台左上角单击，选择“企业中间件 > 分布式消息服务 > RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 4. 单击“导出”，导出实例列表。 5. 单击“导出 > 导出全部数据到XLSX”/“导出 > 导出已选中数据到XLSX”，导出实例列表。

本文带您了解分布式消息服务RocketMQ的产品架构。 系统部署架构如下图所示： Nameserver Nameserver是一个几乎无状态节点，每个集群部署3个Nameserver节点，节点之间无同步信息。它主要提供Broker注册、Topic路由管理等功能。 Broker 分布式消息中间件核心组件，提供消息生产、消费，主从同步、数据刷盘等核心功能。可以横向扩展、在线扩容以提高集群性能。每个Broker由一个Master节点（主节点）和一个Slave节点（备节点）组成，与Nameserver集群的所有节点建立长连接，并定时注册Topic等信息。 Producer 生产者，一般为应用调用API进行消息生产。Producer与Nameserver集群中的其中一个节点（随机选择）建立长连接，定期从Nameserver取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。 Consumer 消费者，一般为应用调用API进行消息消费。Consumer与Producer一样，与一个Nameserver建立长连接并取Topic路由信息。Consumer与提供Topic服务的Master建立长连接，进行消息消费。

本章介绍如何使用主子账号体系管理子账号权限 概述 分布式消息服务MQTT已接入主子账号体系，可区分两种账号权限，实现主账号对子账号的数据权限管理与功能权限管理，支持系统策略和自定义策略的授权方式。本章介绍如何使用主子账号体系管理子账号权限。 背景 1. 主账号 ：用户在天翼云注册后自动创建，该账号对其所拥有的资源具有完全的访问权限，可以重置用户密码、分配用户权限等。如果需要多人共同使用天翼云资源，由于账号是付费主体为了确保账号安全，建议创建子用户来进行日常管理工作。 2. 子账号 ：主账号认证为企业账号后，在天翼云用户中心页面创建出来的账号。子账号的用户名、密码统一由主账号创建管理。子账号同样可以登录访问天翼云控制台，登录入口与主账号相同，受主账号赋予的权限限制。 3. 企业项目： 将云资源、企业成员按项目进行管理，通过企业项目将云资源、带有权限的用户组绑定到一起，用户使用项目内云资源的权限受用户组的授权限制。 注意 一个实例只能归属一个企业项目（可变更），一个子账号可以同时在多个企业项目中。 4. 策略： 是描述一组权限集的语言，它可以精确地描述被授权的资源集和操作集，通过策略，用户可以自由搭配需要授予的权限集。通过给用户组授予策略，用户组中的用户就能获得策略中定义的权限。 5. 系统策略： 系统预置的常用权限集，主要针对不同云服务的只读权限或管理员权限，比如对组件的只读权限、普通用户权限和管理员权限等等；系统策略只能用于授权，不能编辑和修改。 6. 数据权限： 看到的数据不一样。主账号看到所有实例，子账号只能看到所属项目中的实例。 7. 功能权限： 主账号可以进行所有控制台操作，子账号对单个组件实例拥有的操作权限由主账号授权。 8. 功能权限授权： 给子账号在企业项目A下增加一个策略，即代表该子账号对企业项目A下的实例拥有了策略中定义的权限，策略以外的操作会被禁止。

分布式消息服务RocketMQ 当事件目标选择分布式消息服务RocketMQ时，事件目标的type值为rocketmq，params中的resourceKey字段中含义如下表所示。 resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务RocketMQ实例ID。 说明 实例Id可在分布式消息服务RocketMQ管理控制台实例详情页获取。 无 topic 是 CONSTANT RocketMQ主题。 无 body 是 CONSTANT ORIGINAL JSONPATH TEMPLATE 消息体。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 keys 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息索引。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 properties 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息自定义属性。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 tags 是 CONSTANT EMPTY JSONPATH TEMPLATE 消息标签。 如果form选择TEMPLATE，则在此处配置模板。详见事件内容转换。 请求示例： plaintext { "type": "rocketmq", "params": [ { "resourceKey": "instanceId", "value": "89ceb110331e4c968499744c2ccbdbcc", "form": "CONSTANT" }, { "resourceKey": "topic", "value": "TopicA", "form": "CONSTANT" }, { "resourceKey": "body", "value": "", "form": "ORIGINAL" }, { "resourceKey": "properties", "value": "", "form": "EMPTY" }, { "resourceKey": "keys", "value": "", "form": "EMPTY" }, { "resourceKey": "tags", "value": "", "form": "EMPTY" } ] }

本文介绍分布式消息服务RocketMQ的入门指引。 本章节将为您介绍分布式消息服务RocketMQ入门的基本流程，主要包括环境准备、控制台创建RocketMQ实例、创建Topic以及生成消费验证，帮助您快速上手RocketMQ。 操作流程 步骤说明 操作流程如下： 1.环境准备 创建RocketMQ实例先要准备好虚拟私有云、子网和安全组。 2.创建RocketMQ实例 在订购分布式消息RocketMQ填写和确认实例名称、引擎类型、计费模式等信息，确认费用后点击下一步，等待开通流程结果通知成功后完成创建实例。 3.创建主题和订阅组 开通实例后，在控制台相关页面按照指引创建主题和订阅组，用于发送和接收消息。 4.创建应用用户和分配权限 创建应用用户并分配主题和订阅组权限，实现各应用共享或者独占消息队列的效果。 5.生产消费验证 以上工作完成后，通过控制台拨测功能生成生产和消费测试数据，用户应用也可按照规范接入RocketMQ，发送、消费消息。

介绍分布式消息服务RabbitMQ按需转包周期。 操作步骤 1、登录分布式消息服务RabbitMQ消息控制台，可以看到当前租户下面的实例列表。 2、点击需要变更实例栏 > 更多 > 按需转包周期。 3、进入到按需转包周期页面，在弹出来的确认窗口选择续订时长，点击确认即可。

说明 分布式消息服务RocketMQ兼容了社区版 HTTP SDK，您可以使用社区版 HTTP SDK接入分布式消息服务RocketMQ。 前提条件 1. 下载社区 C SDK到本地并解压。 2. 使用Visual Studio打开sln文件导入工程。 发送普通消息 using System; using Aliyun.MQ; using Aliyun.MQ.Model; namespace MQ.Sample { public class Producer { // 填写分布式消息服务RocketMQ控制台HTTP接入点 private const string endpoint "${HTTPENDPOINT}"; // 填写AccessKey，在管理控制台创建 private const string accessKeyId "${ACCESSKEY}"; // 填写SecretKey 在管理控制台创建 private const string secretAccessKey "${SECRETKEY}"; // 消息所属的Topic，在消息队列RocketMQ版控制台创建。 private const string topicName "${TOPIC}"; // Topic所属实例ID，默认实例为空 private const string instanceId "${INSTANCEID}"; private static MQClient client new Aliyun.MQ.MQClient(accessKeyId, secretAccessKey, endpoint); static MQProducer producer client.GetProducer(instanceId, topicName); static void Main(string[] args) { try { // 循环发送4条消息。 for (int i 0; i messages null; try { messages consumer.ConsumeMessage( 3, // 一次最多消费3条(最多可设置为16条) 3 // 长轮询时间3秒（最多可设置为30秒） ); } catch (Exception exp1) { if (exp1 is MessageNotExistException) { Console.WriteLine(Thread.CurrentThread.Name + " No new message, " + ((MessageNotExistException)exp1).RequestId); continue; } Console.WriteLine(exp1); Thread.Sleep(2000); } if (messages null) { continue; } List handlers new List<>(); Console.WriteLine(Thread.CurrentThread.Name + " Receive Messages:"); // 处理业务逻辑 foreach (Message message in messages) { Console.WriteLine(message); handlers.Add(message.ReceiptHandle); } // Message.nextConsumeTime前若不确认消息消费成功，则消息会重复消费 // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样 try { consumer.AckMessage(handlers); Console.WriteLine("Ack message success:"); foreach (string handle in handlers) { Console.Write("t" + handle); } Console.WriteLine(); } catch (Exception exp2) { // 某些消息的句柄可能超时了会导致确认不成功 if (exp2 is AckMessageException) { AckMessageException ackExp (AckMessageException)exp2; Console.WriteLine("Ack message fail, RequestId:" + ackExp.RequestId); foreach (AckMessageErrorItem errorItem in ackExp.ErrorItems) { Console.WriteLine("tErrorHandle:" + errorItem.ReceiptHandle + ",ErrorCode:" + errorItem.ErrorCode + ",ErrorMsg:" + errorItem.ErrorMessage); } } } } catch (Exception ex) { Console.WriteLine(ex); Thread.Sleep(2000); } } } }}

说明 分布式消息服务RocketMQ兼容了社区版 HTTP SDK，您可以使用社区版 HTTP SDK接入分布式消息服务RocketMQ。 前提条件： 1. 在PHP安装目录下的composer.json文件中加入社区PHP SDK 依赖。 2. 使用Composer安装依赖。 composer install 发送普通消息 client new MQClient( // 填写分布式消息服务RocketMQ控制台HTTP接入点 "${HTTPENDPOINT}", // 填写AccessKey，在管理控制台创建 "${ACCESSKEY}", // 填写SecretKey 在管理控制台创建 "${SECRETKEY}" ); // 所属的 Topic $topic "${TOPIC}"; // Topic所属实例ID，默认实例为空NULL $instanceId "${INSTANCEID}"; $this>producer $this>client>getProducer($instanceId, $topic); } public function run() { try { for ($i 1; $i putProperty("a", $i); // 设置消息KEY $publishMessage>setMessageKey("MessageKey"); $result $this>producer>publishMessage($publishMessage); print "Send mq message success. msgId is:" . $result>getMessageId() . ", bodyMD5 is:" . $result>getMessageBodyMD5() . "n"; } } catch (Exception $e) { printr($e>getMessage() . "n"); } }}$instance new NormalProducerExample();$instance>run();?> 消费普通消息 client new MQClient( // 填写分布式消息服务RocketMQ控制台HTTP接入点 "${HTTPENDPOINT}", // 填写AccessKey，在管理控制台创建 "${ACCESSKEY}", // 填写SecretKey 在管理控制台创建 "${SECRETKEY}" ); // 所属的 Topic $topic "${TOPIC}"; // 您在控制台创建的 Consumer ID(Group ID) $groupId "${GROUPID}"; // Topic所属实例ID，默认实例为空NULL $instanceId "${INSTANCEID}"; $this>consumer $this>client>getConsumer($instanceId, $topic, $groupId, "TagA"); } public function run() { // 在当前线程循环消费消息，建议是多开个几个线程并发消费消息 while (True) { try { // 长轮询消费消息 // 长轮询表示如果topic没有消息则请求会在服务端挂住3s，3s内如果有消息可以消费则立即返回 $messages $this>consumer>consumeMessage( 3, // 一次最多消费3条(最多可设置为16条) 3 // 长轮询时间3秒（最多可设置为30秒） ); } catch (MQExceptionMessageResolveException $e) { // 当出现消息Body存在不合法字符，无法解析的时候，会抛出此异常。 // 可以正常解析的消息列表。 $messages $e>getPartialResult()>getMessages(); // 无法正常解析的消息列表。 $failMessages $e>getPartialResult()>getFailResolveMessages(); $receiptHandles array(); foreach ($messages as $message) { // 处理业务逻辑。 $receiptHandles[] $message>getReceiptHandle(); printf("MsgID %sn", $message>getMessageId()); } foreach ($failMessages as $failMessage) { // 处理存在不合法字符，无法解析的消息。 $receiptHandles[] $failMessage>getReceiptHandle(); printf("Fail To Resolve Message. MsgID %sn", $failMessage>getMessageId()); } $this>ackMessages($receiptHandles); continue; } catch (Exception $e) { if ($e instanceof MQExceptionMessageNotExistException) { // 没有消息可以消费，接着轮询 printf("No message, contine long polling!RequestId:%sn", $e>getRequestId()); continue; } printr($e>getMessage() . "n"); sleep(3); continue; } print "consume finish, messages:n"; // 处理业务逻辑 $receiptHandles array(); foreach ($messages as $message) { $receiptHandles[] $message>getReceiptHandle(); printf("MessageID:%s TAG:%s BODY:%s nPublishTime:%d, FirstConsumeTime:%d, nConsumedTimes:%d, NextConsumeTime:%d,MessageKey:%sn", $message>getMessageId(), $message>getMessageTag(), $message>getMessageBody(), $message>getPublishTime(), $message>getFirstConsumeTime(), $message>getConsumedTimes(), $message>getNextConsumeTime(), $message>getMessageKey()); printr($message>getProperties()); } // $message>getNextConsumeTime()前若不确认消息消费成功，则消息会重复消费 // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样 printr($receiptHandles); try { $this>ackMessages($receiptHandles); } catch (Exception $e) { if ($e instanceof MQExceptionAckMessageException) { // 某些消息的句柄可能超时了会导致确认不成功 printf("Ack Error, RequestId:%sn", $e>getRequestId()); foreach ($e>getAckMessageErrorItems() as $errorItem) { printf("tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%sn", $errorItem>getReceiptHandle(), $errorItem>getErrorCode(), $errorItem>getErrorCode()); } } } print "ack finishn"; } } public function ackMessages($receiptHandles) { try { $this>consumer>ackMessage($receiptHandles); } catch (Exception $e) { if ($e instanceof MQExceptionAckMessageException) { // 某些消息的句柄可能超时，会导致消费确认失败。 printf("Ack Error, RequestId:%sn", $e>getRequestId()); foreach ($e>getAckMessageErrorItems() as $errorItem) { printf("tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%sn", $errorItem>getReceiptHandle(), $errorItem>getErrorCode(), $errorItem>getErrorCode()); } } } }}$instance new ConsumerExample();$instance>run();?>