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

本章节主要介绍插件类问题。 支持的RabbitMQ插件有哪些？ RabbitMQ实例购买后，支持的插件如下，其中，方括号中为空的表示还未安装，标记为[E]的插件是明确安装的，标记为[e]的插件是隐式安装的，也就是说，这些插件是作为其它的插件的依赖而进行安装的。 [ ] rabbitmqamqp10 3.8.35 [ ] rabbitmqauthbackendcache 3.8.35 [ ] rabbitmqauthbackendhttp 3.8.35 [ ] rabbitmqauthbackendldap 3.8.35 [ ] rabbitmqauthmechanismssl 3.8.35 [ ] rabbitmqconsistenthashexchange 3.8.35 [ ] rabbitmqdelayedmessageexchange 3.8.9 [ ] rabbitmqeventexchange 3.8.35 [ ] rabbitmqfederation 3.8.35 [ ] rabbitmqfederationmanagement 3.8.35 [ ] rabbitmqjmstopicexchange 3.8.35 [E] rabbitmqmanagement 3.8.35 [e] rabbitmqmanagementagent 3.8.35 [ ] rabbitmqmqtt 3.8.35 [ ] rabbitmqpeerdiscoveryaws 3.8.35 [ ] rabbitmqpeerdiscoverycommon 3.8.35 [ ] rabbitmqamqp10 3.8.35 [ ] rabbitmqauthbackendcache 3.8.35 [ ] rabbitmqauthbackendhttp 3.8.35 [ ] rabbitmqauthbackendldap 3.8.35 [ ] rabbitmqauthmechanismssl 3.8.35 [ ] rabbitmqconsistenthashexchange 3.8.35 [ ] rabbitmqdelayedmessageexchange 3.8.9 [ ] rabbitmqeventexchange 3.8.35 [ ] rabbitmqfederation 3.8.35 [ ] rabbitmqfederationmanagement 3.8.35 [ ] rabbitmqjmstopicexchange 3.8.35 [E] rabbitmqmanagement 3.8.35 [e] rabbitmqmanagementagent 3.8.35 [ ] rabbitmqmqtt 3.8.35 [ ] rabbitmqpeerdiscoveryaws 3.8.35 [ ] rabbitmqpeerdiscoverycommon 3.8.35 RabbitMQ控制台支持安装的插件有：rabbitmqamqp10、rabbitmqdelayedmessageexchange、rabbitmqfederation、rabbitmqsharding、rabbitmqshovel、rabbitmqtracing、rabbitmqmqtt、rabbitmqwebmqtt、rabbitmqstomp、rabbitmqwebstomp和rabbitmqconsistenthashexchange，实例创建后，以上插件默认关闭，如需开启，在控制台实例详情的“插件管理”页面开启，具体操作请参见开启实例插件。 RabbitMQ插件功能可用于测试和迁移业务等场景，不建议用于生产业务。详情请参考约束与限制。 说明 2023年9月30日之前购买的RabbitMQ实例，不支持在控制台开启以下插件：rabbitmqtracing、rabbitmqmqtt、rabbitmqwebmqtt、rabbitmqstomp、rabbitmqwebstomp、rabbitmqconsistenthashexchange，如果需要开启，请在控制台提工单，技术人员在后台开启插件，开启过程中，对业务没有任何影响。 如果需要激活其他未安装且控制台不支持开启的插件（如rabbitmqrandomexchange），需要在控制台提工单，技术人员在后台开启插件，开启过程中，对业务没有任何影响。 未开启SSL的RabbitMQ单机实例，支持开启rabbitmqshovel插件。RabbitMQ集群实例和开启SSL的RabbitMQ单机实例，不支持开启rabbitmqshovel插件。

本章节介绍了有关分布式消息服务RocketMQ实例消息类的常见问题及解答。 是否支持消费验证？ 分布式消息服务RocketMQ当前不支持消费验证功能。 RocketMQ的消息保留时间是多少？可以修改吗？ RocketMQ的消息保留时间是48小时，且不支持修改。 RocketMQ支持的最大消息大小是多少？可以修改吗？ RocketMQ支持的最大消息大小是4M，且不支持修改。 RocketMQ副本存储形式是怎样的？可以修改吗？ RocketMQ消息是一主两从3个副本的存储形式，且不支持修改。 消息创建时间在哪设置 ？ 消息创建时间是由生产客户端在生产消息时设置的。

本文介绍如何在事件总线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 消费位点 开始消费的位置。仅支持从最早位点进行消费。 最早位点：从最初位点开始消费。 最早位点

发送消息 参考如下示例代码。 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()

参数说明 参数 说明 示例 实例 选择分布式消息RocketMQ实例。 instancexxx Topic 选择RocketMQ实例的Topic。 topic1 消息体 选择消息体（Body）的内容，更多内容请参考事件内容转换。 全部事件 自定义属性 选择自定义属性（Properties）的内容，更多内容请参考事件内容转换。 空 索引 选择索引（Keys）的内容，更多内容请参考事件内容转换。 空 标签 选择标签（Tags）的内容，更多内容请参考事件内容转换。 空 结果验证 1. 使用事件总线发布消息功能，发送一个自定义事件。 2. 您可以在分布式消息服务RocketMQ控制台确认是否接收到事件。 1. 登录分布式消息服务RocketMQ控制台，然后在左侧导航栏选择实例列表。 2. 在实例列表页面单击目标实例名称。 3. 在消息查询页面，按时间查询即可查看消息内容，如图2所示。 图2 在分布式消息服务RocketMQ管理控制台中查看消息详情

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

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

本文为您介绍分布式消息服务MQTT与其他服务关系。 虚拟私有云(CTVPC ，Virtual Private Cloud) 虚拟私有云为分布式消息服务MQTT提供一个逻辑隔离的区域，构建一个安全可靠、可配置和管理的虚拟网络环境。更多信息请参见虚拟私有云。 弹性云主机（CTECS，Elastic Cloud Server） 弹性云主机由 CPU、内存、镜像、云硬盘组成，同时结合VPC、安全组、数据多副本保存等能力，打造一个既高效又可靠安全的计算环境，确保分布式消息服务MQTT持久稳定运行。更多信息请参见弹性云主机。 云硬盘（CTEVS，Elastic Volume Service） 云硬盘是一种可弹性扩展的块存储设备，可以为分布式消息服务MQTT提供高性能、高可靠的块存储服务。更多信息请参见云硬盘。 弹性IP（Elastic IP，EIP） 弹性IP是可以独立申请的公网 IP 地址，包括公网IP地址与公网出口带宽服务。可以与分布式消息服务MQTT动态绑定和解绑，实现云资源的互联网访问。针对需要公网访问分布式消息服务MQTT实例的需求，用户可开通弹性IP后，在MQTT实例页面进行绑定。更多信息请参见弹性IP。

分布式消息服务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()

本章节介绍了如何讲其他厂商或自建的业务迁移到分布式消息服务RocketMQ的实践方案。 操作场景 RocketMQ业务迁移是指将其他厂商或者自建的RocketMQ迁移到天翼云分布式消息服务RocketMQ。 前提条件 1. 配置网络环境 分布式消息服务RocketMQ实例分VPC内以及公网地址两种网络连接方式。如果使用公网地址，则消息生产与消费客户端需要有公网访问权限，并配置如下安全组。 表1 安全组规则 方向 协议 端口 源地址 说明 入方向 TCP 8200 0.0.0.0/0 公网访问元数据节点的端口 入方向 TCP 1010010199 0.0.0.0/0 访问业务节点的端口 2. 购买分布式消息服务RocketMQ实例 具体请参考购买RocketMQ实例。 操作步骤 1. 迁移元数据至分布式消息服务RocketMQ实例。 1. 获取其他厂商或自建RocketMQ实例的元数据。 2. 登录主机，下载RocketMQ软件包。 wget i < 1. 解压软件包。 unzip rocketmqall4.9.4binrelease.zip 1. （可选）如果RocketMQ实例开启了ACL访问控制，执行mqadmin命令时，需要鉴权。切换到解压后的软件包目录下，在“conf/tools.yml”文件中，增加如下内容。 accessKey: secretKey: accessKey和secretKey表示在控制台“用户管理”页面，创建的用户名和密钥。 1. 进入解压后的软件包目录下，执行以下命令，查询集群名称。sh ./bin/mqadmin clusterList n {nameserver地址及端口号}例如：“nameserver地址及端口号”为“192.168.0.65:8100”。 2. sh ./bin/mqadmin clusterList n 192.168.0.65:8100执行以下命令，导出元数据。未开启SSL的实例，执行以下命令。sh ./bin/mqadmin exportMetadata n {nameserver地址及端口号} c {RocketMQ集群名称} f {导出的元数据文件的存放路径}例如：“nameserver地址及端口号”为“192.168.0.65:8100”，“RocketMQ集群名称”为“DmsCluster”，“导出的元数据文件的存放路径”为“/tmp/rocketmq/export”。 1. sh ./bin/mqadmin exportMetadata n 192.168.0.65:8100 c DmsCluster f /tmp/rocketmq/export已开启SSL的实例，执行以下命令。JAVAOPTDtls.enabletrue sh ./bin/mqadmin exportMetadata n {nameserver地址及端口号} c {RocketMQ集群名称} f {导出的元数据文件的存放路径}例如：“nameserver地址及端口号”为“192.168.0.65:8100”，“RocketMQ集群名称”为“DmsCluster”，“导出的元数据文件的存放路径”为“/tmp/rocketmq/export”。 3. JAVAOPTDtls.enabletrue sh ./bin/mqadmin exportMetadata n 192.168.0.65:8100 c DmsCluster f /tmp/rocketmq/export在控制台迁移元数据。登录控制台。 4. 单击RocketMQ实例的名称，进入实例详情页面。 5. 在左侧导航栏，选择“元数据迁移”，进入迁移任务列表页面。 6. 单击“创建迁移任务”，弹出“创建迁移任务”对话框。 参考，设置迁移任务的参数。 参数 说明 任务名称 您可以自定义迁移任务的名称，用于区分不同的迁移任务。 是否同名覆盖 如果开启同名覆盖，会对已有的同名元数据的配置进行修改。例如：原实例Topic01的读队列个数为3，云上实例Topic01的读队列个数为2，开启同名覆盖后，云上实例Topic01的读队列个数变为3。如果不开启同名覆盖，同名元数据的迁移将失败。例如：原实例的Topic包含Topic01和Topic02，云上实例的Topic包含Topic01和Topic03，不开启同名覆盖，原实例Topic01的迁移将失败。 元数据 上传。 1. 单击“确定”。迁移完成后，在迁移任务列表页面查看“任务状态”。 1. 当“任务状态”为“迁移完成”，表示所有元数据都已成功迁移。 2. 当“任务状态”为“迁移失败”，表示元数据中部分或全部元数据迁移失败。单击迁移任务名称，进入迁移任务详情页，在“迁移结果”中查看迁移失败的Topic/消费组名称，以及失败原因。 2. 迁移生产消息至分布式消息服务RocketMQ版实例。将生产客户端的元数据连接地址改为分布式消息服务RocketMQ版实例的元数据连接地址，重启生产业务，使得生产者将新的消息发送到分布式消息服务RocketMQ版实例中。 3. 迁移消费消息至分布式消息服务RocketMQ版实例。待消费组中的消息消费完之后，将消费客户端的元数据连接地址改为分布式消息服务RocketMQ版实例的元数据连接地址，重启消费业务，使得消费者从分布式消息服务RocketMQ版实例中消费消息。 4. 如果有多个RocketMQ实例需要迁移到同一个分布式消息服务RocketMQ版实例中，请依次进行迁移

介绍分布式消息产品选型对比项。 特性 Kafka RabbitMQ RocketMQ 功能 支持功能较少，不支持延迟发送，消息重试等功能 功能丰富，支持多个队列种类（优先级队列、延迟队列、死信队列镜像队列等），提供丰富的策略分配 功能完善，支持事务消息、定时消息、事务消息等 单机吞吐量 十万级 万级 几万级 稳定性 队列/分区多时性能不稳定 消息堆积时，性能不稳定 队列较多、消息堆积时性能保持稳定 可用性 非常高（分布式）具有主备故障自动切换 较高，基于主从架构实现高可用性 非常高（分布式）具有主备故障自动切换 选型建议 性能要求高，数据量大，适合产生大量数据的互联网服务的数据收集业务，如日志采集处理、需对接大数据应用等，kafka是首选 数据量少，吞吐量需求不大；数据可靠性要求较高，对功能丰富性极高 可靠性要求很高且性能要求较高的场景以及业务削峰场景，如电商、订单处理等

开源对比 相较于开源自建Kafka，分布式消息服务Kafka在低成本运维、分区规模、消息查询、ACL访问控制、可视化配置、运维监控、集群巡检、稳定可靠、安全保证、简单易用等方面更具优势。更多信息请参见开源对比。 产品优势 分布式消息服务Kafka具备高可用性、高安全性、可靠性、全托管等优势，使其成为大规模数据处理和实时流处理的理想选择。更多信息请参见产品优势。 功能特性 分布式消息服务Kafka的功能特性主要体现在以下几个方面： 消息能力 广播消息：在同一个消费组内对所有消费者投递相同消息。 消息回溯：支持根据时间重置消费进度。 消息数据自动删除功能：在磁盘满后，在保护期外的数据，能自动删除，保证服务可用性。 自动故障切换功能：生产消费自动负载均衡，消息节点故障时自动主备切换，保证服务的连续性。 队列能力 高吞吐，消息多副本异步复制。 高可靠，消息多副本同步复制。 可视化管理 应用用户管理：多个应用可调用同一个消息服务，通过应用用户，对消息服务下的应用接入权限进行管理。 主题管理：支持对实例下的主题进行管理，执行创建删除等操作。 消费组管理：支持对实例下的消费组进行管理。 Broker监控：提供Broker详细信息以及多维度的监控指标查看。 Topic监控：提供Topic详细信息以及多维度的监控指标查看。

本节介绍分布式消息服务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）是指每秒可以生产消息和消费消息的总次数，可以理解为对应规格每秒生产消息和消费消息的总吞吐量。

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

本文介绍如何在事件总线EventBridge控制台添加分布式消息服务MQTT作为事件流中的事件提供方。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 分布式消息服务MQTT 开通分布式消息服务MQTT。 操作步骤 1. 登录事件总线控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在事件源（Source） 配置面板，事件提供方 选择分布式消息MQTT，选择或填入主题等配置然后单击确认，如图1所示。 图1 创建事件流时分布式消息服务MQTT作为流的事件提供方 参数说明 参数 说明 示例 MQTT实例 选择MQTT实例。 instancexxx MQTT主题 选择或输入MQTT的主题。 topicxxx 事件示例 plaintext { "id":"b5771f766cdf48edb1bad15418c", "source":"ctyun:mqtt", "specversion":"1.0", "type":"mqtt:Topic:SendMessage", "subject":"ctyun:mqtt:bb9fdb421610242ac110002:16789:topic/topictest", "time":"20240305T13:52:18.374Z", "datacontenttype":"application/json", "data":{ "body":{ "key": "value" }, "props": { "topic": "topictest" } }, "ctyunaccountid":"16789", "ctyunregion":"bb9fdb421610242ac110002", } data 字段包含的参数解释如下表所示。 参数 类型 示例值 描述 props Object {"topic":"topictest"} 消息属性。 body Object {"key":"value"} 消息体，默认以JSON格式编码。

本文为您介绍分布式消息服务MQTT 的入门指引。 本章节将为您介绍分布式消息服务MQTT入门的基本流程，主要包括环境准备、创建实例、连接实例及消息收发，帮助您快速上手MQTT。 操作流程 步骤说明 1.环境准备 创建MQTT实例先要准备好虚拟私有云、子网和安全组，可选弹性公网IP。 2.创建实例 在订购分布式消息MQTT填写和确认实例名称、计费模式等信息，确认费用后点击下一步，等待开通流程结果通知成功后完成创建实例。 3.连接实例 创建用户密码、父主题等管理信息，对用户进行主题授权，绑定弹性IP，终端设备和云端应用通过MQTT客户端接入。 4.消息收发 使用MQTT SDK接入终端连接地址进行消息生产消费。

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

本文为您介绍分布式消息服务MQTT续订和退订内容。 场景描述 分布式消息服务MQTT为用户提供全面周到的服务，支持用户续订和退订的需求。 续订：针对包周期消息实例服务，用户可在到期前进行服务周期延长操作，即续订操作。 退订：用户如不需要继续使用该分布式消息服务MQTT实例，可进行删除实例操作，即退订操作。 续订 1、登录MQTT消息控制台，可以看到当前租户下面的实例列表。 2、点击需要变更实例栏 > 更多 > 续订。 3、 进入到续订页面，在弹出来的确认窗口选择续订时长，点击确认即可。 退订 1、登录MQTT消息控制台，可以看到当前租户下面的实例列表。 2、点击需要变更实例栏 > 更多 > 退订。 3、 进入到退订页面，在弹出来的确认窗口点击确认即可。 注意 退订的实例处于冻结状态，请务必在实例退订前停止全部的应用。 在申请退订前，请做好数据备份工作，退订后数据将保留15个自然日，15天后相关数据将不予保留，且不会进行备份，务必谨慎操作。

Kafka是一个拥有高吞吐、可持久化、可水平扩展，支持流式数据处理等多种特性的分布式消息流处理中间件，采用分布式消息发布与订阅机制，在日志收集、流式数据传输、在线/离线系统分析、实时监控等领域有广泛的应用。 分布式消息服务Kafka是一款基于开源社区版Kafka提供的消息队列服务，向用户提供计算、存储和带宽资源独占式的Kafka专享实例。使用分布式消息服务Kafka，资源按需申请，按需配置Topic的分区与副本数量，即买即用，您将有更多精力专注于业务快速开发，不用考虑部署和运维。

本文带您了解分布式消息服务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建立长连接，进行消息消费。

概述 本文介绍使用 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; }

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

客户端是否可以连接同个RabbitMQ下多个Vhost？ 客户端可以连接同个RabbitMQ下多个Vhost。 Vhost（Virtual Hosts）是RabbitMQ的基本特性，每个Vhost相当于一个相对独立的RabbitMQ服务器，每个Vhost数据目录不同，共用一个进程。性能上，连接多个Vhost和单独使用一个Vhost差别不大，只是RabbitMQ进程多一些对象，建议使用业务模型实测。 Vhost的相关介绍，请参考官网文档Virtual Hosts。 为什么R abbitMQ 集群只有一个连接地址 ？ RabbitMQ集群实例的连接地址，实际上是实例的LVS节点地址（负载均衡地址），客户端连接实例时，通过负载均衡器将客户端请求分发到集群实例的各个节点。

本节介绍了RabbitMQ 接入方式。 安全接入点 RabbitMQ 安全接入点支持 "PLAIN"、"AMQPLAIN" 授权机制。 1、访问控制 RabbitMQ "PLAIN"、"AMQPLAIN"授权机制需要创建用户，从而获得对应虚拟主机的访问权限。 2、接入步骤 （1）新建用户（集群管理>用户>新建用户） （2）运行demo 客户端关键参数设置 "PLAIN"、"AMQPLAIN" 授权机制的客户端关键参数配置 String host "192.168.0.0"; //安全接入点ip Integer port 5672; //安全接入点port String username "xxx"; //集群管理用户列表的用户名 String password "xxx"; String vhost "/"; ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost(vhost); SSL接入点 RabbitMQ 安全接入点支持 "EXTERNAL" 授权机制 1、访问控制 无 2、接入步骤 （1）下载SSL证书（实例概览>导出服务>下载SSL文件） （2）运行demo 客户端关键参数设置 "EXTERNAL" 授权机制的客户端关键参数配置 java String host "192.168.0.0"; //SSL接入点ip int port 5671; //SSL接入点port //以下2个ssl文件可通过控制台获取安装包, 具体的获取方式可以查看2.2.1接入步骤的第二小节 String ksFile "D:tmpsslclientrabbitmqkey.p12"; String tksFile "D:tmpssltruststore"; String vhost "/"; char[] keyPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore ks KeyStore.getInstance("PKCS12"); ks.load(new FileInputStream(ksFile), keyPassphrase); KeyManagerFactory kmf KeyManagerFactory.getInstance("SunX509"); kmf.init(ks, keyPassphrase); char[] trustPassphrase null; trustPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore tks KeyStore.getInstance("JKS"); tks.load(new FileInputStream(tksFile), trustPassphrase); TrustManagerFactory tmf TrustManagerFactory.getInstance("SunX509"); tmf.init(tks); SSLContext c SSLContext.getInstance("tlsv1.2"); c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setVirtualHost(vhost); connectionFactory.setSaslConfig(DefaultSaslConfig.EXTERNAL); connectionFactory.useSslProtocol(c);

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

分布式消息服务RocketMQ实例在被删除后，会被临时存入回收站中，此时实例中的数据尚未被彻底删除，在保留天数内支持从回收站中恢复此实例。超过保留天数的实例会被彻底删除，无法恢复。 约束与限制 回收站中的实例不会收取实例的费用，但是会收取存储空间的费用。 包年/包月的实例退订后会存入回收站中，此时不会收取实例的费用，但是收取存储空间的费用。 回收站中的按需实例，不支持实例恢复。 回收站管理 1. 登录分布式消息服务RocketMQ管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 恢复RocketMQ实例 1. 登录分布式消息服务RocketMQ管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 3. 在待恢复RocketMQ实例所在行，单击“恢复”。 删除回收站中的实例 1. 登录分布式消息服务RocketMQ管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 3. 在待删除RocketMQ实例所在行，单击“销毁”。 注意 回收站中的RocketMQ实例删除后，实例中原有的数据将被删除，且没有备份，请谨慎操作。

本文对分布式消息服务RocketMQ的计费项进行说明。 4.0版本 以下适用于华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 节点。 分布式消息服务RocketMQ费用计算分为两部分，一部分为实例费用，一部分为存储空间费用。 计费项 含义 适用的计费模式 实例费用 用户选择的选择的实例规格节点数计费，实例规格单价请参考产品资费，节点数为Name Server集群固定3节点+Broker（代理） 2节点。 按需/包周期 存储空间费用 用户选择的实例存储空间计费（每个实例规格您都可以选择高IO、超高IO等多种不同的云硬盘类型以满足您的业务需求）。 按需/包周期 3.0版本 以下适用于南京3、上海7、重庆2、乌鲁木齐27、石家庄20、内蒙6、晋中、北京5 节点。 分布式消息服务RocketMQ按照消息队列的规格版本可分为基础版、中级版和高级版。 计费项：消息队列的规格版本。

本文主要介绍了自动续订规则及操作流程。 为避免由于未及时对资源采取续订操作，资源被到期冻结或超期释放，客户购买包月包年产品后，可设置开通自动续订。开通自动续订后，系统将在资源到期前自动续订，无需客户再手动操作。 适用范围 自动续订仅针对采用包月、包年计费模式的资源。 已到期资源不支持设置/修改自动续订。 自动续订仅适于 付费用户 ，不适用于非付费用户。 目前支持设置自动续订的产品有：弹性云主机、GPU云主机、物理机、云桌面、云硬盘、对象存储经典版、云主机备份、专属云计算独享型、专属云存储独享型、弹性IP、共享带宽、天翼云SDWAN、云间高速、关系数据库MySQL版、关系数据库Postgre SQL版、分布式关系数据库、分布式缓存服务Redis版、分布式缓存服务Memcache、文档数据库服务、云HBASE数据库、分布式消息服务RocketMQ、分布式消息服务RabbitMQ、分布式消息服务Kafka、Web应用防火墙、服务器安全卫士、域名无忧、DDos高防IP、云解析、登录保护、网站安全监测、内容安全。 单次最多支持20个资源实例批量续订。 开通、变更、关闭自动续订 用户在续订管理页可开通自动续订功能，变更自动续约周期，或关闭自动续订。 不关闭自动续订的情况下，只要预付费账户余额充足，或为后付费客户，系统将持续按设定的周期自动续订下去。 预付费用户可在官网自主控制自动续订功能的开通、变更、关闭。后付费用户需要客户经理协助开启自动续订权限后才可以自主管理。

介绍分布式消息服务MQTT购买类问题。 MQTT包周期实例不支持删除吗？ 可以删除。登录MQTT控制台，在包周期实例所在行，单击“更多 > 退订”，即可完成实例的删除。 分布式消息服务MQTT资源包可以叠加购买吗？ 天翼云分布式消息服务RocketMQ支持叠加购买。 产品订购时可选资源池节点不一致？ 已上线资源池节点的剩余容量达到一定比例后，为确保老客户权益，将不再面向新客户开放，产品订购时的可选资源节点范围以实际为准。

场景描述 分布式消息服务RocketMQ为用户提供全面周到的服务，支持用户续订和退订的需求。 ● 续订：针对包周期消息实例服务，用户可在到期前进行服务周期延长操作，即续订操作。 ● 退订：用户如不需要继续使用该分布式消息服务RocketMQ实例，可进行删除实例操作，即退订操作。 续订 1、登录RocketMQ消息控制台，可以看到当前租户下面的实例列表。 2、点击需要变更实例栏 > 更多 > 续订。 3、 进入到续订页面，在弹出来的确认窗口选择续订时长，点击确认即可。 退订 1、登录RocketMQ消息控制台，可以看到当前租户下面的实例列表。 2、点击需要变更实例栏 > 更多 > 退订。 3、进入到退订页面，在弹出来的确认窗口点击确认即可。 注意： 退订的实例处于冻结状态，请务必在实例退订前停止全部的应用。 在申请退订前，请做好数据备份工作，退订后数据将保留15个自然日，15天后相关数据将不予保留，且不会进行备份，务必谨慎操作。

顺序消息是分布式消息服务RocketMQ版提供的一种严格按照顺序来发布和消费的消息类型。 顺序消息分为全局顺序消息和分区顺序消息： 全局顺序消息：对于指定的一个Topic，将队列数量设置为1，这个队列内所有消息按照严格的先入先出FIFO（First In First Out）的顺序进行发布和订阅。 分区顺序消息：对于指定的一个Topic，同一个队列内的消息按照严格的FIFO顺序进行发布和订阅。生产者指定分区选择算法，保证需要按顺序消费的消息被分配到同一个队列。 全局顺序消息和分区顺序消息的区别仅为队列数量不同，代码没有区别。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.7。 通过以下任意一种方式引入依赖： 1. 使用Maven方式引入依赖。 org.apache.rocketmq rocketmqclient 4.9.7 org.apache.rocketmq rocketmqacl 4.9.7 2. 点击下载依赖JAR包：rocketmqall4.9.7binrelease.zip 发送顺序消息 参考如下示例代码 import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.RPCHook; import org.apache.rocketmq.remoting.common.RemotingHelper; public class ProducerFifoExample { private static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials( "accessKey", // 分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID "accessSecret" // 分布式消息服务RocketMQ控制台用户管理菜单中创建的密钥 )); } public static void main(String[] args) throws Exception { / 创建Producer，如果想开启消息轨迹，可以按照如下方式创建： DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null); / DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); // 填入控制台NAMESRV接入点地址 producer.setNamesrvAddr("XXX:xxx"); ; // 如果需要开启SSL，请增加此行代码 producer.start(); for (int i 0; i { // 选择适合自己的分区选择算法，保证同一个参数得到的结果相同。 Integer id (Integer) arg; int index id % mqs.size(); return mqs.get(index); }, orderId); System.out.printf("%s%n", sendResult); } catch (Exception e) { e.printStackTrace(); } } producer.shutdown(); } } 注意 上述代码中，相同id的消息需要保证顺序，不同id的消息不需要保证顺序，所以在分区选择算法中以“id/队列个数的余数”作为消息发送的队列。