本文主要介绍与RabbitMQ、RocketMQ的差异。 分布式消息服务Kafka与RabbitMQ、RocketMQ的对比。 表1 分布式消息服务Kafka与开源Kafka的差异说明 功能项 RocketMQ Kafka RabbitMQ 优先级队列 不支持 不支持 支持。建议优先级大小设置在010之间。 延迟队列 支持 不支持 不支持 死信队列 支持 不支持 支持 消息重试 支持 不支持 不支持 消费模式 支持客户端主动拉取和服务端推送两种方式 客户端主动拉取 支持客户端主动拉取以及服务端推送两种模式 广播消费 支持 支持 支持 消息回溯 支持 支持。Kafka支持按照offset和timestamp两种维度进行消息回溯。 不支持。RabbitMQ中消息一旦被确认消费就会被标记删除。 消息堆积 支持 支持。考虑吞吐因素，Kafka的堆积效率比RabbitMQ总体上要高。 支持 持久化 支持 支持 支持 消息追踪 支持 不支持 支持。RabbitMQ中可以采用Firehose或者rabbitmqtracing插件实现，但开启rabbitmqtracing插件会影响性能，建议只在定位问题过程中开启。 消息过滤 支持 支持 不支持，但可以自行封装。 多租户 支持 不支持 支持 多协议支持 兼容RocketMQ协议 只支持Kafka自定义协议。 RabbitMQ基于AMQP协议实现，同时支持MQTT、STOMP等协议。 跨语言支持 支持多语言的客户端 采用Scala和Java编写，支持多种语言的客户端。 采用Erlang编写，支持多种语言的客户端。 流量控制 待规划 支持client和user级别，通过主动设置可将流控作用于生产者或消费者。 RabbitMQ的流控基于CreditBased算法，是内部被动触发的保护机制，作用于生产者层面。 消息顺序性 单队列（queue）内有序 支持单分区（partition）级别的顺序性。 不支持。需要单线程发送、单线程消费并且不采用延迟队列、优先级队列等一些高级功能整体配合，才能实现消息有序。 安全机制 支持SSL认证 支持SSL、SASL身份认证和读写权限控制。 与Kafka相似 事务性消息 支持 支持 支持

场景说明 RocketMQ的修改实例的场景描述如下： 在使用RocketMQ时，可能会遇到需要修改实例的场景，例如： 业务变更：当业务需求发生变化，需要修改RocketMQ实例的名称以反映新的业务逻辑时，可以进行实例名称的修改。例如，当新增或调整了某个业务模块，需要将其对应的实例名称修改为更准确的名称，以便于管理和监控。 消息处理速度优化：当消息的处理速度较慢或不稳定时，可以通过调整消息保存时长参数来优化消息的处理效率。例如，可以缩短消息的保存时长，减少消息在系统中的停留时间，以提高消息的实时性和处理速度。 安全配置修改：当需要修改RocketMQ实例的安全配置时，可以进行相应的修改。 需要注意的是，在进行实例的修改时，应该谨慎操作，并根据实际需求和系统情况进行调整。同时，修改实例时应该遵循RocketMQ的最佳实践和建议，以确保系统的稳定性和性能。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、进入实例列表，在实例名称点击修改按钮，直接修改实例名称字段。 4、 点击【管理】按钮进入管理菜单。点击修改配置，在弹出窗口修改实例描述和消息保存时长等信息。 注意 实例描述长度限制为0~256个字符。 您可以调整消息保存时长参数，实现消息容量消耗的控制。安全生产和稳定性保障限制，消息需要至少保留24小时。 要完整支持延时消息的功能，消息需要至少保留48小时。 4、 点击修改安全组，选择变更用户安全组。

本章节主要介绍数据仓库服务如何订阅告警通知。 用户可通过订阅DWS 的告警通知，在特定告警级别告警时收到短信、电子邮件或应用等方式的通知消息。 创建订阅 1.登录DWS管理控制台。 2.在左侧导航树，单击“告警管理”，切换至“订阅”页签。 3.在页面左上角单击“创建订阅”按钮。 4.在“订阅设置”区域，设置订阅基本信息及告警过滤。 订阅参数 参数名 参数解释 是否开启 设置是否开启告警订阅。 表示开启告警订阅。表示关闭告警订阅。关闭后停止发送已订阅告警的通知消息，但不会删除该订阅。 订阅名称 设置订阅告警的名称。 名称只能包含大写字母、小写字母、数字、和，且必须由大写字母、小写字母或数字开头。 名称长度为1～256字符。 告警级别 选择订阅告警的级别：紧急、重要、次要和提示。 5.“订阅告警列表”区域显示系统根据订阅设置筛选出的告警。在“消息通知主题名称”下拉框中，选择一个消息通知主题。 如需创建新主题，请单击“创建新主题”按钮，系统将跳转到消息通知服务控制台页面。 说明 所选择的消息通知主题，必须已授予DWS 服务向该主题发布消息的权限。如果所选主题尚未给DWS 授权，请前往消息通知服务的主题管理页面设置主题策略授权。设置主题策略时，“可发布消息的服务”需勾选“DWS”。 6.确认无误后，单击“确定”，完成创建订阅。

本文主要介绍如何在控制台删除消息。 操作场景 说明 消息删除后无法恢复，请谨慎操作。 前提条件 删除消息前，请先在客户端中设置“auto.offset.reset”参数。“auto.offset.reset”用来指定当Kafka中没有初始偏移量或者当前偏移量不存在（例如当前偏移量已被删除）时，消费者的消费策略。取值如下： latest：偏移量自动被重置到最晚偏移量。 earliest：偏移量自动被重置到最早偏移量。 none：向消费者抛出异常。 操作步骤 1. 登录管理控制台。 2. 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 3. 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 4. 单击Kafka实例名称，进入实例详情页面。 5. 在左侧导航栏选择“Topic管理”，进入Topic列表页面。 6. 在待删除消息的Topic所在行，单击“消息删除”，弹出“消息删除”对话框。 7. 设置消息删除参数，如下表所示。 图消息删除 表 消息删除参数说明 参数 说明 分区 选择消息所在的分区编号。 偏移量 输入偏移量，此偏移量之前的数据将被删除。 说明 如果“偏移量”设置为“1”，表示删除分区中所有的消息。 如果您输入的偏移量不在指定分区的最早偏移量和最晚偏移量之间，消息将不会被删除。 如果需要删除多个分区的消息，单击“添加分区”，设置需要删除消息的分区和偏移量。每次最多可选择10个分区。 8. 单击“确定”，弹出“清理结果”对话框，单击“确定”，完成消息的删除。 图 清理结果

分布式消息服务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" } ] }

事件目标是事件规则的一部分，负责消费经事件规则过滤与转换后的事件。 事件目标 事件总线EventBridge支持以下事件目标： 函数计算 分布式消息服务RocketMQ 分布式消息服务RabbitMQ 分布式消息服务Kafka HTTP、HTTPS地址

案例背景 某电商平台订单处理系统采用分布式微服务架构，其概要结构图如下，核心流程通过分布式消息服务Kafka集群实现异步解耦： 订单创建：用户在前端提交订单，请求经过负载均衡器转发至订单服务集群。订单服务处理业务逻辑，生成包含订单详情的消息数据，并将该消息可靠地发送至Kafka集群的特定Topic。 下游处理：库存服务（负责扣减库存）、支付服务（负责发起支付请求）和物流服务（负责生成物流配送单）是主要的下游消费者。它们各自订阅Kafka集群中的相应Topic，并行地拉取并处理消息。 Kafka集群作为异步通信的核心，其稳定性和性能对整个订单系统的可用性至关重要。在高并发的场景下，Kafka Broker可能会面临各种异常情况，其中CPU负载过高是常见的一种，可能由突发流量洪峰、JVM垃圾回收或底层基础设施瓶颈等原因引起。Kafka Broker的CPU高负载会直接影响消息的处理速度，导致生产和消费的延迟，甚至引发整个系统的不稳定。 故障演练服务提供了分布式消息服务KafkaBroker节点CPU高负载演练动作，通过模拟这种真实的故障场景，我们可以量化分析Kafka集群在Broker节点CPU受限时的表现，验证订单处理系统及其下游服务对Kafka异常的容忍度和应对能力，并提升团队在真实故障发生时的异常响应与恢复能力。

分布式消息服务RabbitMQ 当事件源的事件提供方选择分布式消息服务RabbitMQ时，请求参数中的type为rabbitmq，eventSourceParams 中的resourceKey 字段含义如下表所示： resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务RabbitMQ实例ID。 无 vhost 是 CONSTANT vhost名称。 无 queueName 是 CONSTANT 队列名。 无 请求示例如下： plaintext { "eventSourceName": "test", "eventBusName": "test", "desc": "", "type": "rabbitmq", "eventSourceParams": [ { "resourceKey": "instanceId", "value": "e8c05d2c9f44d77a703373c6737df98b", "form": "CONSTANT" }, { "resourceKey": "vhost", "value": "/", "form": "CONSTANT" }, { "resourceKey": "queueName", "value": "test1", "form": "CONSTANT" } ] } 分布式消息服务MQTT 当事件源的事件提供方选择分布式消息服务MQTT时，请求参数中的type为mqtt，eventSourceParams 中的resourceKey 字段含义如下表所示： resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务MQTT实例ID。 无 topic 是 CONSTANT 主题名称。 无 dataFormat 否 CONSTANT 数据格式,不填时默认为JSON编码格式： JSON格式编码：JSON 文本格式编码：Text 二进制格式编码：Binary 无 请求示例如下： plaintext { "eventSourceName": "test", "eventBusName": "test", "desc": "desc", "type": "mqtt", "eventSourceParams": [ { "resourceKey": "instanceId", "value": "26554575ff4dxxxxxf69ee17838d11ca", "form": "CONSTANT" }, { "resourceKey": "topic", "value": "test1", "form": "CONSTANT" }, { "resourceKey": "dataFormat", "value": "JSON", "form": "CONSTANT" } ] }

本节主要介绍分布式消息服务Kafka的创建Topic说明 背景信息 Kafka的主题（Topic）是对消息进行分类和组织的概念。主题是Kafka中最基本的逻辑单元，用于区分不同类型的消息。 每个主题可以有多个分区（Partition），每个分区都是一个有序的消息日志。分区允许消息在集群中进行并行处理，提高了吞吐量和并发性能。 主题的特点包括： 逻辑分类：主题可以根据业务需求或数据类型进行逻辑分类，例如订单主题、日志主题、用户行为主题等。 多分区：主题可以被分成多个分区，每个分区都有自己的消息顺序。分区可以在不同的节点上进行存储和处理，实现了水平扩展和负载均衡。 副本复制：每个分区可以有多个副本（Replica），副本用于提供高可用性和容错性。副本可以分布在不同的节点上，当节点故障时可以自动进行故障转移。 持久化存储：主题中的消息被持久化存储在磁盘上，确保消息的可靠性和持久性。消费者可以随时消费主题中的历史消息，而不仅仅是最新的消息。 通过使用主题，Kafka能够有效地组织和管理大规模的消息流。主题的分区和副本机制提供了高可用性和容错性，使得Kafka成为处理大规模实时数据流的理想选择。 前提条件 已购买并部署分布式消息服务Kafka 操作步骤 创建Topic （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“Topic管理”后、点击“创建Topic”。 （5）点击“创建Topic”后，输入Topic名称、分区数等参数。 （6）创建好Topic后，可以点击“更多”按钮中的“生产消息”来测试其是否正常。

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

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

介绍分布式消息服务Kafka生产拨测功能操作内容。 场景描述 通过生产消息，确保Kafka集群能够稳定接收并传递生产者发送的消息，同时消费者能够成功订阅并处理这些消息，以此确认集群的基本功能和网络连接均处于正常状态。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“Topic管理”后进入主题管理页面。 （5）选择需要生产消息的Topic，点击其右侧“更多”，在下拉框中点击“生产消息”按钮。 （6）在弹窗中，输入消息内容、消息条数等信息，点击确认即可发送。

本文主要介绍日志处理工具Logstash接入Kafka。 最佳实践概述 场景描述 Kafka与Logstash互联，可实现将Kafka作为inputs把消息传递给Logstash进行处理、或者是作为outputs将Kafka用作消息目的地。 技术架构图 暂无。 方案优势 可以异步处理数据：防止突发流量。 解耦：当发生异常的时候不会影响上游工作。 前提条件 需已购买Kafka实例、创建Topic，并且已成功消费消息。 资源规划 本实践方案内容仅涉及Kafka专享版实例和Logstash。 分布式消息服务 Figure 分布式消息服务 资源类型 配置项 配置明细 说明 :::: 企业中间件 DMS Kafka专享实例 需已购买kafka专享实例，创建好Topic，并成功消费消息。 Logstash下载与安装 下载并安装Logstash，验证运行成功。 方案正文 配置Kafka 1、登录分布式消息服务Kafka版控制台，按需要单击“实例名称”，进入实例基本信息页面。 图 Kafka控制台实例列表 2、在“连接地址”模块，获取Kafka连接IP地址。 图 连接IP地址与端口 3、创建Logstashtest的Topic。 图 创建Topic

本文主要介绍分布式消息服务RabbitMQ的消息持久化最佳实践。 使用场景 默认情况下，RabbitMQ生产者生产的消息存储在内存中，当节点宕机或重启时，如何确保消息不丢失呢？RabbitMQ通过持久化机制实现，持久化包括Exchange持久化、Queue持久化和Message持久化。持久化是将内存中的消息写入到磁盘中，以防异常情况导致内存中的消息丢失。但是磁盘的读写速度远不如内存，开启消息持久化后，RabbitMQ的性能会下降。与惰性队列不同，持久化消息会在磁盘和内存中各存储一份，只有在内存空间不够时，才会将内存中的消息删除，存储到磁盘中。 说明 非持久化队列、Exchange在重启之后会丢失。 非持久化消息在重启之后会丢失（经过持久化队列/Exchange的消息不会自动变为持久化消息）。 持久化消息在尚未完成持久化时，如果服务器重启，消息会丢失。 设置Exchange持久化 在RabbitMQ的Web界面创建Exchange时，设置“durable”为“true”，如图1所示，设置成功后如图2所示。 图1 设置Exchange持久化 图2 持久化的Exchange

本文介绍分布式消息服务RocketMQ入门指引的创建主题和订阅组内容。 背景信息 在实例创建完成后，需要创建主题和订阅组来进行消息实例的日常功能运转。 主题：在RocketMQ中，主题（Topic）是消息发布的逻辑分组。它类似于一个消息的分类或者标签，帮助用户将不同类型的消息进行归类和管理。通常情况下，一个主题可以包含多个消息生产者和多个消息消费者。通过使用主题，RocketMQ能够实现高效的消息发布和订阅机制，帮助用户更好地管理和组织消息。 订阅组：订阅组是 RocketMQ 中的一个重要概念，用于实现消息的发布与订阅模式。一个订阅组可以包含多个消费者实例，这些实例共同消费同一个主题下的消息。当消息被发送到主题时，订阅组中的每个消费者实例会按照一定规则来均衡地接收消息，并进行相应的处理。订阅组是 RocketMQ 提供的一种灵活且可扩展的方式，用于实现消息的发布与订阅模式，并保证消息在消费者之间的均衡分配和可靠处理。 集群：RocketMQ集群是一种由多个节点（或者称为Broker）组成的分布式消息中间件系统。每个节点都具有相同的功能并且可以处理和存储消息。通过将消息分发到不同的节点，RocketMQ集群能够实现高可用性和可伸缩性。通过使用RocketMQ集群，可以实现消息传输的并行处理、容错性和高可用性，满足高并发场景下的消息传递需求。 创建主题 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入主题管理菜单，点击【新建主题】按钮 5、 在弹出的新建主题页面，填写如下字段信息 1）默认展示当前集群名称，不可修改。 2）选择主题所在的Broker，按照实例创建时候选择的主备节点对数列出每个broker，可复选。 3）填写主题名称，名字限制2到64个字符，超过限制会导致创建主题失败，用户创建主题只能包含大小写字母数字以及和符号。 4）按照实际需求填写主题备注。 5）填写每个Broker分区数，分区数必须大于0，小于等于8，创建严格顺序队列时，设置分区数为1，且只能选择一个Broker。 6）选择生产模式，RocketMQ是一个开源的分布式消息中间件，它支持两种消息生产模式：有序和无序。 有序消息生产模式（Ordered Message）是指按照特定规则将消息发送到相同的Message Queue中，并且确保消息在消费者端按照相同的顺序进行消费。这种模式适用于那些需要严格按照消息顺序进行处理的场景，比如订单处理、流程审批等。 无序消息生产模式（Unordered Message）是指消息发送到不同的Message Queue中，每个Queue都是独立的。消费者可以并行地从多个Queue中消费消息，而无需关心消息的顺序。这种模式适用于那些不需要严格按照消息顺序处理的场景，比如日志收集、异步通知等。 需要注意的是，无论是有序还是无序消息生产模式，RocketMQ都提供了高可靠性的消息传输和存储，并支持水平扩展和高吞吐量的特性。根据具体的业务需求，选择适合的消息生产模式能够更好地满足应用的要求。 7）选择主题的读写权限，支持读写、只读、只写3类权限。 8） 完成主题信息填写后，保存确认即可新增主题。 9） 若希望批量创建主题，可点击【批量创建】按钮 批量创建 注意：输入的主题名不要带空格等特殊字符。 通过上传csv文件,批量创建主题。格式：点击【主题模板】按钮下载。 主题模板 批量上传主题的模板，必须使用模板，才能够上传成功，模板格式如下：

本文介绍分布式消息服务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实例费用计算示例 按需计费模式下，各计费项的计费示例请参见计费样例。

应用架构 支持的功能模块 应用双活 管控通道 注册配置中心RCC 应用双活 接入层 云原生网关CGW 应用双活 服务层 SpringCloud 应用双活 服务层 Dubbo 应用双活 数据层 关系型数据库MySQL版 应用双活 数据层 关系型数据库PostgreSQL版 应用双活 消息层 分布式消息服务RocketMQ 数据双活 管控通道 注册配置中心RCC 数据双活 接入层 云原生网关CGW 数据双活 服务层 SpringCloud 数据双活 服务层 Dubbo 数据双活 数据层 关系型数据库MySQL版 数据双活 数据层 关系型数据库PostgreSQL版 数据双活 消息层 分布式消息服务RocketMQ

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

在业务场景允许的情况下，优先选择无序消息，或者在业务能变通的情况下，将有序消息转化为无序消息。 无序消息的优点： 生产者可以使用多进程、多线程往同一个Topic发送消息，性能更好。 消费者可以使用多进程、多线程同时消费，性能较好。 可以充分使用集群的Failover特点，无须依赖自动主备切换（切换过程服务会中断），包括： 当集群中某一Broker节点故障时，不影响业务消息生产，消息将failover发送到其它节点； 当集群中某一Broker节点故障时，不影响其它节点数据消费，故障恢复后即可消费。 能动态地扩容。 有序消息的缺点： 对于有序消息，当节点故障时，Queue数不会变化，生产与消费都会出现异常，直到故障节点恢复。 对于有序消息，需要将所有消息消费完，并且停止客户端，才能扩容。

本章节主要介绍分布式消息服务RabbitMQ的消息持久化。 使用场景 默认情况下，RabbitMQ生产者生产的消息存储在内存中，当节点宕机或重启时，如何确保消息不丢失呢？RabbitMQ通过持久化机制实现，持久化包括Exchange持久化、Queue持久化和Message持久化。 持久化是将内存中的消息写入到磁盘中，以防异常情况导致内存中的消息丢失。但是磁盘的读写速度远不如内存，开启消息持久化后，RabbitMQ的性能会下降。 与惰性队列不同，持久化消息会在磁盘和内存中各存储一份，只有在内存空间不够时，才会将内存中的消息删除，存储到磁盘中。 说明 非持久化Queue、Exchange在重启之后会丢失。 非持久化Message在重启之后会丢失（经过持久化Queue/Exchange的消息不会自动变为持久化消息）。 持久化消息在尚未完成持久化时，如果服务器重启，消息会丢失。 设置Exchange持久化 在RabbitMQ WebUI页面创建Exchange时，设置“durable”为“true”，如图1所示，设置成功后如图2所示。 图1 设置Exchange持久化 图2 持久化的Exchange

如何获取RocketMQ实例连接信息请参阅收集连接信息。 开发指南详细介绍Java、Go和Python客户端访问分布式消息服务RocketMQ的示例代码，具体如表1所示。 表1 示例代码 客户端语言 示例代码 :: Java 收发普通消息 收发顺序消息 收发事务消息 发送定时/延时消息 Go（TCP协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时/延时消息 Python（TCP协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时/延时消息

云服务名称 区域 支持IAM 支持企业项目 支持IAM 支持企业项目 系统策略 分布式消息服务Kafka 全局 是 是 是 是 有 分布式消息服务RabbitMQ 全局 是 是 是 是 有 分布式消息服务RocketMQ 全局 是 是 是 是 有 微服务云应用平台 全局 是 是 是 是 有 云容器引擎 全局 是 是 是 是 有 弹性容器实例 全局 是 是 是 是 有 函数计算 全局 是 是 是 是 有 云日志服务LTS 全局 是 是 是 是 有 容器镜像服务 全局 是 是 是 是 有 分布式消息服务MQTT 全局 是 是 是 是 有 应用服务网格 全局 是 是 是 是 有 微服务引擎 全局 是 是 是 是 有

分布式消息服务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按照消息队列的规格版本可分为基础版、中级版和高级版。 计费项：消息队列的规格版本。

背景信息 分布式消息服务RabbitMQ版的消息都会被投入到一个或多个队列中。Consumer Tag是消费者客户端的标识符。您可以在分布式消息服务 RabbitMQ 版的消费者客户端设置Consumer Tag。 操作步骤 创建队列 1.登录管理控制台。 2.进入RabbitMQ管理控制台。 3.在实例列表页在操作列，目标实例行点击“管理”。 4.点击“队列管理”后，点击“新建”按钮。 5.点击“新建”后出现以下窗口，选择虚拟主机，输入队列名字，选择存储节点，然后点击确定即可创建队列。 参数说明如下表所示： 参数 描述 虚拟主机 选择创建队列所属的虚拟主机 名称 队列的名称。以amq.开头的为保留字段，因此不能使用。例如：amq.test。 存储节点 队列数据存储节点 是否持久化 队列元数据是否持久化到磁盘 是否自动删除 最后一个Consumer取消订阅后，Queue是否自动删除。 其他参数 Message TTL消息过期时间：number型(单位:ms) Auto expire队列过期时间，过期后队列自动删除：number型(单位:ms) Max length队列能保存的最大消息数：number型(单位:个) Max length bytes队列能保存的最大消息量：number型(单位:字节) Overflow behaviour 超过队列的最大设定值后消息接收策略：drophead,rejectpublish drophead：删除头部消息,一般就是最早发送的消息，保证队列可用 rejectpublish：拒绝接受新的消息，保证消息不丢失 Dead letter exchange死信交换器名称 Dead letter routing key死信路由键 Maximum priority队列最大优先级：要开启消息的优先级，必须设置消息所在队列的优先级 Lazy mode队列惰性模式：default、lazy default：默认值，普通队列 lazy：惰性队列，尽可能将消息存到磁盘中，会引起I/O操作比较多，内存消耗极少（有大量堆积的持久化消息建议使用） Master Locator 队列保存位置：clientlocal、minmasters、random clientlocal：队列创建时所用连接的节点 minmasters： 集群中节点主数量最少的节点 random：由rabbitmq服务器随机指定一个节点

本文介绍分布式消息服务RocketMQ入门指引的生产消费验证内容。 背景信息 RocketMQ的生产消费验证是指在使用RocketMQ进行消息生产和消费时的验证过程。具体而言，验证包括以下几个方面： 生产者验证：RocketMQ提供了丰富的生产者API，开发人员可以使用这些API将消息发送到RocketMQ的消息队列中。在验证阶段，可以通过发送消息并检查返回结果来确保消息成功发送到Broker节点。此外，生产者还应该验证消息的顺序性、事务性以及可靠性等方面。 消费者验证：RocketMQ的消费者可以订阅特定的消息主题，从而消费这些主题下的消息。在验证阶段，消费者应该能够正确地从Broker节点拉取消息并进行消费处理。消费者还可以验证消息的顺序性、重试机制以及消息过滤等功能。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入Topic管理菜单，点击【拨测】按钮，进行生产消费的拨测验证，验证开通的消息实例和主题。 1）生产测试拨测： 选择消息类型，默认普通消息。 填写需要产生的测试消息数量，以及每条消息的大小，默认每条消息1KB，建议不超过4MB(4096KB)。 选择已建的消息主题，若无选项，请新增主题，详见上文创建主题和订阅组。 点击【测试】按钮，按照已填写规格及数量产生测试消息数据，展示消息数据的信息，包括消息ID(messageID)、发送状态、主题名（topic名）、Broker名、队列ID。 拨测功能涉及消息发送状态码，以下是RocketMQ消息发送状态码及其说明： ✧ SENDOK（发送成功）：表示消息成功发送到了消息服务器。 ✧ FLUSHDISKTIMEOUT（刷新磁盘超时）：表示消息已经成功发送到消息服务器，但是刷新到磁盘上超时。这可能会导致消息服务器在宕机后，尚未持久化到磁盘上的数据丢失。 ✧ FLUSHSLAVETIMEOUT（刷新从服务器超时）：表示消息已经成功发送到消息服务器，但是刷新到从服务器上超时。这可能会导致主从同步不一致。 ✧ SLAVENOTAVAILABLE（从服务器不可用）：表示消息已经成功发送到消息服务器，但是从服务器不可用。这可能是由于网络故障或从服务器宕机引起的。 ✧ UNKNOWNERROR（未知错误）：表示发送消息时遇到了未知的错误。一般情况下建议重试发送消息。 ✧ MESSAGESIZEEXCEEDED（消息大小超过限制）：表示消息的大小超过了消息服务器的限制。需要检查消息的大小是否合适。 ✧ PRODUCETHROTTLE（消息生产被限流）：表示消息生产者的频率超出了消息服务器的限制。这可能是由于消息发送频率过高引起的。 ✧ SERVICENOTAVAILABLE（服务不可用）：表示消息服务器不可用。这可能是由于网络故障或者消息服务器宕机引起的。 请注意，以上状态码仅适用于RocketMQ消息发送阶段，并且不代表消息是否成功被消费者接收。同时，这些状态码也可能因版本变化而有所不同，建议查阅官方文档获取最新信息。 2）消费测试拨测： 选择消息顺序，下拉选择无序/有序，默认选项为无序。 RocketMQ是一种开源的分布式消息中间件，它支持有序消息和无序消息。 ✧ 有序消息是指消息的消费顺序与发送顺序完全一致。在某些业务场景下，消息的处理需要保证顺序性，例如订单的处理或者任务的执行。RocketMQ提供了有序消息的支持，通过指定消息的顺序属性或使用消息队列的分区机制，可以确保消息按照指定的顺序进行消费。 ✧ 无序消息则是指消息的消费顺序与发送顺序无关。无序消息的特点是高吞吐量和低延迟，适用于一些不要求严格顺序的业务场景，如日志收集等。 在RocketMQ中，有序消息和无序消息的实现方式略有不同。有序消息需要借助MessageQueue的分区机制和消费者端的顺序消息消费来实现。而无序消息则是通过消息的发送和接收的并发处理来实现的。 总的来说，RocketMQ既支持有序消息也支持无序消息，根据业务需求选择合适的消息类型来满足业务的要求。 选择消费方式，目前仅提供pull方式。值得注意的是，RocketMQ还提供了推送（push）方式的消费模式，其中消息队列服务器会主动将消息推送给消费者。但在当前仅限于pull方式的消费模式。 填写消费数量。 下拉选择选择已建的消息主题和订阅组，若无选项，请新增主题和订阅组，详见上文创建主题和订阅组。 点击【测试】按钮，按照已填写规格及数量产生消费数据，展示消息数据的信息，包括消息ID(messageID)、主题名称（topicName）、生成时间、存储时间、队列ID、消费状态。 拨测功能涉及消息消费状态码，RocketMQ消费状态码是指在消息消费过程中，对消费结果进行标识的状态码。以下是常见的RocketMQ消费状态码： ✧ CONSUMESUCCESS（消费成功）：表示消息成功被消费。 ✧ RECONSUMELATER（稍后重试）：表示消费失败，需要稍后再次进行消费。 ✧ CONSUMEFAILURE（消费失败）：表示消息消费出现异常或失败。 ✧ SLAVENOTAVAILABLE（从节点不可用）：表示消费者无法访问从节点来消费消息。 ✧ NOMATCHEDMESSAGE（无匹配的消息）：表示当前没有匹配的消息需要消费。 ✧ OFFSETILLEGAL（偏移量非法）：表示消费的偏移量参数不合法。 ✧ BROKERTIMEOUT（Broker超时）：表示由于Broker超时导致消费失败。 5、 用户应用按照规范接入RocketMQ，发送、消费消息。 1）生产者示例API 以下适用于南京3、上海7、重庆2、乌鲁木齐27、保定、石家庄20、内蒙6、晋中、北京5节点。 ctgmq引擎版本，SDK下载方式详见环境准备其他工具章节。 package com.ctg.guide; import com.ctg.mq.api.CTGMQFactory; import com.ctg.mq.api.IMQProducer; import com.ctg.mq.api.PropertyKeyConst; import com.ctg.mq.api.bean.MQMessage; import com.ctg.mq.api.bean.MQSendResult; import com.ctg.mq.api.exception.MQException; import com.ctg.mq.api.exception.MQProducerException; import java.util.Properties; / Producer，发送消息 / public class Producer { public static void main(String[] args) throws InterruptedException, MQException { Properties properties new Properties(); properties.setProperty(PropertyKeyConst.ProducerGroupName, "producergroup"); properties.setProperty(PropertyKeyConst.NamesrvAddr, "10.50.208.1:9876;10.50.208.2:9876;10.50.208.3:9876"); properties.setProperty(PropertyKeyConst.NamesrvAuthID, "app4test"); properties.setProperty(PropertyKeyConst.NamesrvAuthPwd, ""); properties.setProperty(PropertyKeyConst.ClusterName, "defaultMQBrokerCluster"); properties.setProperty(PropertyKeyConst.TenantID, "defaultMQTenantID"); IMQProducer producer CTGMQFactory.createProducer(properties);//建议应用启动时创建 int connectResult producer.connect(); if(connectResult ! 0){ return; } for (int i 0; i mqResultList) { //mqResultList 默认为1，可通过批量消费数量设置 for(MQResult result : mqResultList) { //TODO System.out.println(result); } return ConsumerTopicStatus.CONSUMESUCCESS;//对消息批量确认(成功) //return ConsumerTopicStatus.RECONSUMELATER;//对消息批量确认(失败) } }); } } 以下适用于华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3节点。 rocketmq引擎版本，SDK下载方式详见环境准备其他工具章节。 importorg.apache.rocketmq.client.consumer.DefaultMQPushConsumer; importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; importorg.apache.rocketmq.client.exception.MQClientException; publicclassPushConsumer{ public static void main(String[] args) throws Exception{ AclClientRPCHook rpcHook newAclClientRPCHook(newSessionCredentials(ACCESSKEY, SECRETKEY)); DefaultMQPushConsumer consumer new DefaultMQPushConsumer(rpcHook); consumer.setConsumerGroup("ConsumerGroupName"); // 填入元数据地址 consumer.setNamesrvAddr("XXX:xxx"); ; //如果需要开启SSL，请增加此行代码 consumer.subscribe("YOUR TOPIC",""); consumer.registerMessageListener((MessageListenerConcurrently)(msgs, context)>{ System.out.printf("%s Receive New Messages: %s %n",Thread.currentThread().getName(), msgs); returnConsumeConcurrentlyStatus.CONSUMESUCCESS; }); consumer.start(); System.out.printf("Consumer Started.%n"); } } 示例参数说明： Namesrv地址 Namesrv地址可从控制台查看，多个地址按分号分隔。 应用用户和密码 这个应用用户和密码就是控制台创建的应用用户和密码。 租户id和集群名 集群名和租户id可以从应用用户管理查询。 订阅组 消费组名需要在控制台提前创建好。

此小节介绍企业主机安全与其他云服务的关系。 使用企业主机安全服务，您将可以同时使用消息通知服务接收告警通知信息，使用统一身份认证服务管理用户权限，利用云审计服务审计用户行为。 弹性云主机 企业主机安全服务的Agent软件可安装在ECS服务器上。 关于弹性云主机的详细内容，请参见弹性云主机用户指南。 消息通知服务 消息通知服务（Simple Message Notification，简称SMN），是一个可拓展的高性能消息处理服务。 开启告警通知前，您需先配置“消息通知服务”。 开启消息通知服务后，当您的主机遭受攻击或被检测出有高危风险时，您将接收到企业主机安全服务发送的各项风险告警通知。 在“告警通知”界面，您可以根据运维计划选择“每日告警通知”和“实时告警通知”。 统一身份认证服务 统一身份认证服务（Identity and Access Management，简称IAM），是一个免费的身份管理服务。通过IAM服务，您可以根据用户的身份，对用户的权限进行精细化隔离和控制。IAM是权限管理的基础服务，无需付费即可使用。 关于IAM的详细内容，请参统一身份认证服务用户指南。 企业管理 企业中有多个项目，多个项目的资源需要分开结算，且分属不同人员进行管理。同时项目可以单独启动或停止，对其他项目没有影响。那么可以针对企业中的每个项目，分别建立企业项目，管理各自的资源，并且针对不同的企业项目，设置不同的人员进行管理。

section3aef9b71341e43c4)。 非简单请求的跨域访问 注意 非简单请求的跨域访问需要在API的分组中创建一个“请求方法”为“OPTIONS”的API，作为预检请求。 预检请求API的参数设置，请参考以下说明填写。详细的使用指导可参考非简单请求。 a. 在“前端定义”中，参数填写说明如下： 请求方法：选择“OPTIONS” 请求协议：选择与已开启CORS的API相同的请求协议 请求Path：填斜杠/ b. 在“安全配置”中，安全认证选“无认证”，勾选“开启支持跨域CORS”。 c. 后端服务选择Mock 简单请求 对于简单请求，您需要开启简单跨域访问。 场景一：已开启CORS，且后端服务响应消息中未指定跨域头时，API网关接受任意域的请求，并返回“AccessControlAllowOrigin”跨域头，示例如下： 浏览器发送一个带Origin字段的请求消息： GET /simple HTTP/1.1 Host: www.test.com Orgin: ContentType: application/xwwwformurlencoded; charsetutf8 Accept: application/json Date: Tue, 15 Jan 2019 01:25:52 GMT Origin：此字段必选，表示请求消息所属源，上例中请求来源于“ 后端服务返回响应消息： HTTP/1.1 200 OK Date: Tue, 15 Jan 2019 01:25:52 GMT ContentType: application/json ContentLength: 16 Server: apigateway {"status":"200"} API网关响应消息： HTTP/1.1 200 OK Date: Tue, 15 Jan 2019 01:25:52 GMT ContentType: application/json ContentLength: 16 Server: apigateway XRequestId: 454d689fa69847610b3ca486458fb08b AccessControlAllowOrigin: {"status":"200"} AccessControlAllowOrigin：此字段必选，“”表示API网关接受任意域的请求。 场景二：已开启CORS，且后端服务响应消息中指定跨域头时，后端服务响应的跨域头将覆盖API网关增加的跨域头，示例如下： 浏览器发送一个带Origin字段的请求消息： GET /simple HTTP/1.1 Host: www.test.com Orgin: ContentType: application/xwwwformurlencoded; charsetutf8 Accept: application/json Date: Tue, 15 Jan 2019 01:25:52 GMT Origin：此字段必选，表示请求消息所属源，上例中请求来源于“ 后端服务返回响应消息： HTTP/1.1 200 OK Date: Tue, 15 Jan 2019 01:25:52 GMT ContentType: application/json ContentLength: 16 Server: apigateway AccessControlAllowOrigin: {"status":"200"} AccessControlAllowOrigin：表示后端服务接受“ API网关响应消息： HTTP/1.1 200 OK Date: Tue, 15 Jan 2019 01:25:52 GMT ContentType: application/json ContentLength: 16 Server: apigateway XRequestId: 454d689fa69847610b3ca486458fb08b AccessControlAllowOrigin: {"status":"200"} 后端服务响应消息中的跨域头覆盖API网关响应消息中的跨域头。

云服务名称 是否支持条件键 云通信短信 否 弹性文件服务 否 分布式消息服务Kafka 否 账务 是 分布式消息服务RabbitMQ 否 分布式消息服务RocketMQ 否 云硬盘 是 天翼云电脑（政企版） 是 内网DNS 否 CRM业务中台 是 分布式缓存服务Redis版 否 弹性云主机 是 弹性负载均衡 否 镜像服务 是 云监控 否 弹性伸缩服务 否 虚拟私有云 是 物理机 否 云硬盘备份 否 云主机备份 否 服务器安全卫士（原生版） 否 密钥管理 否 云间高速 否 统一身份认证 是 客服工单 是 容器云服务引擎CCSE 是 微服务应用平台MSAP 否 活动与券 是 消息管理 是 Web应用防火墙（原生版） 否 云审计 否 企业组织 是 VPC终端节点 否 NAT网关 是

本章节介绍Kafka Broker分区Leader不可用故障演练。 背景介绍 分布式消息服务 Kafka 集群中，分区 Leader 节点宕机、网络分区或资源过载等因素均可能造成分区 Leader 不可用，进而引发对应分区服务中断、客户端请求失败、Leader 重选举导致延迟升高及副本同步紊乱等问题，本演练可帮助验证集群高可用选举机制的有效性、监控告警的及时性，以及业务系统应对此类故障的容错能力与恢复效率。 基本原理 指定一个或多个分区Leader，通过调用Kafka模拟Leader故障OpenAPI，触发Leader重新选举。 故障注入 1、纳管实例资源 1. 导航至 故障演练 > 目标应用 > 应用资源页面。 2. 在资源类型页签中选择分布式消息服务Kafka，然后单击添加资源。 3. 在弹出的对话框中，勾选目标分布式消息服务Kafka实例，单击确定。 2、编排演练任务 1. 导航至 故障演练 > 目标应用 > 演练管理 页面，单击新建演练。 2. 在基本信息 页面，按提示填写演练名称和描述，然后单击下一步。 3. 在演练对象配置页面： 配置动作组 ：为动作组 命名，资源类型选择分布式消息服务Kafka。 添加实例 ：单击添加实例 ，勾选上一步中添加的分布式消息服务Kafka实例。 添加故障动作 ：单击立即添加 ，在列表中选择分区Leader不可用动作。 4. 在弹出的参数配置框中，配置所需参数，然后单击确定。 持续时间：故障动作持续时间。 主题分区：指定需要模拟分区Leader不可用的目标对象。