场景说明 RocketMQ提供了两种连接： 生产者连接：生产者通过调用RocketMQ提供的API，与RocketMQ代理（Broker）建立连接。生产者连接主要用于发送消息到RocketMQ。 消费者连接：消费者通过调用RocketMQ提供的API，与RocketMQ代理（Broker）建立连接。消费者连接主要用于从RocketMQ订阅消息并进行消费。 RocketMQ采用基于TCP/IP的通信协议，使用长连接方式进行连接。在建立连接之前，需要配置正确的RocketMQ服务端地址和端口，并使用相应的身份验证信息（如AccessKey和SecretKey）进行认证。 建立连接后，RocketMQ客户端可以通过连接发送消息到RocketMQ集群，并从集群中接收消息进行消费。连接的建立和维护是RocketMQ消息传递的基础，确保了消息的可靠传递和高效处理。 操作步骤 首先需要在RocketMQ控制台实例详情记录下集群的接入点信息，目前默认支持Namesrv的内网访问，将该地址作为客户端程序Namesrv地址的参数，具体生产消费消息的客户端示例代码请参考快速入门生产消费验证。 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入主题管理菜单，点击【拨测】按钮，进行生产消费的拨测验证，验证开通的消息实例和主题。 5、 用户应用按照规范接入RocketMQ，发送、消费消息。

本文介绍分布式消息服务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文件,批量创建主题。格式：点击【主题模板】按钮下载。 主题模板 批量上传主题的模板，必须使用模板，才能够上传成功，模板格式如下：

创建联系人组 进入分布式消息服务MQTT管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理 点击通知组菜单，右侧点击新建联系人组，输入联系人信息信息，点击保存 创建通知策略 进入分布式消息服务MQTT管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理 点击通知策略菜单，右侧点击创建通知策略，输入信息，点击确认 创建告警规则 进入分布式消息服务MQTT管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理 点击告警规则菜单，右侧点击创建告警规则，选择告警指标，输入告警条件、告警内容等信息，点击保存 成为第一个赞同者

创建联系人组 进入分布式消息服务RocketMQ管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理。 点击通知组菜单，右侧点击新建联系人组，输入联系人信息信息，点击保存。 创建通知策略 进入分布式消息服务RocketMQ管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理。 点击通知策略菜单，右侧点击创建通知策略，输入信息，点击确认。 创建告警规则 进入分布式消息服务RocketMQ管理控制台，在实例列表页点击管理进入实例详情页，左侧菜单点击告警管理。 点击告警规则菜单，右侧点击创建告警规则，选择告警指标，输入告警条件、告警内容等信息，点击保存。

消息堆积对业务有什么影响？ 为了保证服务的稳定可靠，分布式消息服务RabbitMQ版采用了默认的40%高水位配置。当内存占用率达到40%高水位后，会触发流控，生产者流程会被阻塞。消息堆积可能造成内存高水位，为了避免高水位的产生，请及时消费积压在队列中的消息。 消费的最长保留时间是多久？ 一般情况下消息如果未被消费会一直保留，只有被消费后，才会被删除。但是如果设置了过期时间（TTL），则以TTL时间为准。

Hue连接hiveserver，不释放session，报错over max user connections如何处理？ 适用版本：MRS 3.1.0及之前的MRS 3.x版本。 1.修改两个Hue节点的以下文件： /opt/Bigdata/FusionInsightPorter8. /install/FusionInsightHue /hue/apps/beeswax/src/beeswax/models.py 2.修改文件中的396和404行的值 q self.filter(owneruser, applicationapplication).exclude(guid'').exclude(secret'') 改为 q self.filter(owneruser, applicationapplication).exclude(guidNone).exclude(secretNone) 如何重置Kafka数据？ 删除Kafka topic信息即重置Kafka数据，具体命令请参考： 删除topic： kafkatopics.sh delete zookeeper ZooKeeper集群业务IP:2181/kafka topic topicname 查询所有topic： kafkatopics.sh zookeeper ZooKeeper集群业务IP:2181/kafka list 执行删除命令后topic数据为空则此topic会立刻被删除，如果有数据则会标记删除，后续Kafka会自行进行实际删除。 Kafka目前支持的访问协议类型有哪些？ 当前支持4种协议类型的访问：PLAINTEXT、SSL、SASLPLAINTEXT、SASLSSL。 消费Kafka Topic时报错“Not Authorized to access group XXX”如何处理？ 该问题是由于集群的Ranger鉴权和集群自带的ACL鉴权冲突导致。Kafka集群使用自带的ACL进行权限访问控制，且集群的Kafka服务也开启Ranger鉴权控制时，该组件所有鉴权将由Ranger统一管理，原鉴权插件设置的权限将会失效，导致ACL权限授权未生效。可通过关闭Kafka的Ranger鉴权并重启Kafka服务来处理该问题。操作步骤如下： 1.登录FusionInsight Manager页面，选择“集群 > Kafka”。 2.在服务“概览”页面右上角单击“更多”，选择“停用Ranger鉴权”。在弹出的对话框中输入密码，单击“确定”，操作成功后单击“完成”。 3.在服务“概览”页面右上角单击“更多”，选择“重启服务”，重启Kafka服务。

APM适合多种运维监控场景,此篇例举部分应用场景如下。 场景一：定位应用性能瓶颈与故障异常 场景说明 随着业务不断发展，业务逻辑日益复杂，导致应用性能问题分析与定位日益艰难，给监控运维带来了巨大的挑战： 应用之间的依赖关系复杂，难以梳理。 调用链路长，排查和定位群体困难。 接口调用、数据库调用关系复杂，管理难度大。APM提供了大型分布式应用的异常诊断能力，当应用出现请求失败或性能下降时，通过应用拓扑、调用链、性能指标监控等能力组合，可以帮助用户快速定位问题。 业务价值 应用拓扑展示：自动梳理业务应用，以拓扑图的方式全面展示相关应用调用关系。 丰富的指标监控：提供包括JVM、资源监控、Netty内存等基础监控；Kafka、RocketMQ等消息监控；Mysql，redis，es等数据库监控；httpClient、grpc等调用监控。 慢SQL分析：通过自定义的慢查询阈值、结合SQL的调用频次，获取导致数据库性能下降的不规范的SQL语句。 告警：针对接口响应时间、异常调用、数据库、JVM等性能指标做一定阈值的告警，先于客户之前发现并解决问题。 场景二：容器运维监控 场景说明 众所周知，Prometheus 是容器场景的最佳监控工具，但自建 Prometheus 对运维人力有限的中小企业成本过高，对业务快速发展的企业又易出现性能瓶颈。因此使用云托管 Prometheus 已成为越来越多上云企业的首选。

本文介绍分布式消息服务RocketMQ的生产者实例查询操作内容。 场景描述 RocketMQ显示当前在线的生产者实例有以下作用： ● 监控生产者状态：通过显示当前在线的生产者实例，可以实时监控生产者的状态。可以了解生产者的连接情况、运行状态和发送速率等信息，帮助及时发现生产者异常或故障，以便进行及时处理和调整。 ● 故障排查与追踪：通过显示当前在线的生产者实例，可以帮助定位消息发送失败的原因。可以查看每个生产者实例的运行情况，包括发送的消息数量、发送延迟等信息，方便排查故障和进行问题定位。 综上所述，显示当前在线的生产者实例可以帮助监控生产者状态、故障排查等，为系统的稳定运行和性能调优提供重要支持。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入生产者实例查询菜单，列出了指定集群，指定Broker下的生产组和关联的Topic的情况。 列表展示了指定集群和指定broker下的生产组和关联topic的情况。 连接实例 显示该生产组，当前在线的生产者实例列表。

本文为您分布式消息服务MQTT消息收发内容。 会话机制 终端 clean sessiontrue,断线后会话信息清除，再次上线后之前所有的订阅关系以及离线消息丢失。 clean sessionfalse断开连接的情况下，MQTT Broker也会为断连客户端保存一个会话，默认2小时，超期未重连订购关系清除；对于clean sessionfalse的客户端断线重连后可接收Qos>0的离线消息。对于客户端因网络等各种原因断线，需要加上重连和订购关系重新订购机制。 离线消息 对于clean sessionfalse的客户端，在未超出会话失效期，断线重连后可接收Qos>0的离线消息。 系统主题 系统主题 说明 mq2mqtt 用于云端服务向终端发送消息。发往该主题消息会转发至MQTT Broker实现云端与移动端互通 mqttdeviceconnect 设备上线主题，内容 {"clientid":客户端ID,"ts":上线时间戳 } mqttdevicedisconnect 设备下线主题，内容 {"clientid":客户端ID,"ts":下线时间戳 } SDK支持 分布式消息服务MQTT支持标准的MQTT协议，理论上适配所有的MQTT客户端SDK。 推荐对应的第三方 SDK 如下表： 语言/平台 推荐的第三方SDK Java Eclipse Paho SDK iOS MQTTClientFramework Android Eclipse Paho SDK JavaScript Eclipse Paho JavaScript Python Eclipse Paho Python SDK C Eclipse Paho C SDK C Eclipse Paho C SDK Golang Eclipse Paho Golang SDK Node.js MQTTJS PHP MosquittoPHP

身份认证 CTIAM 统一身份认证（Identity and Access Management， 简称：CTIAM）是提供用户进行权限管理的基础服务，可以帮助您安全的控制天翼云服务和资源的访问及操作权限，包括：用户身份认证、权限分配、访问控制等功能。具体介绍请参考统一身份认证产品介绍。 您可以创建IAM用户，并为其设置关分布式消息服务MQTT实例权限，该用户就可以通过用户名和密码访问授权的实例资源。具体请参见统一身份认证快速入门创建IAM用户。 访问控制 权限控制 购买分布式消息服务MQTT实例之后，您可以使用CTIAM为企业中的员工设置不同的访问权限，以达到不同员工之间的权限隔离，通过CTIAM进行精细的权限管理。 VPC和子网 虚拟私有云（Virtual Private Cloud，VPC）为分布式消息服务MQTT构建隔离、私密的虚拟网络环境，提升消息队列的安全性，并简化用户的网络部署。您可以完全掌控自己的专有网络，VPC丰富的功能帮助您灵活管理云上网络，包括创建子网、设置安全组和网络ACL、管理路由表、申请弹性公网IP和带宽等。通过子网与其他网络隔离，独享网络资源，提高网络安全性。具体内容请参见虚拟私有云用户指南创建虚拟私有云和子网 。

使用Kafka客户端（MRS 3.x及之后版本） 1. 进入ZooKeeper实例页面： 登录FusionInsight Manager，具体请参见访问FusionInsight Manager（MRS 3.x及之后版本）。然后选择“集群 > 待操作的集群名称 > 服务 > ZooKeeper > 实例”。 2. 查看ZooKeeper角色实例的IP地址。 记录ZooKeeper角色实例其中任意一个的IP地址即可。 3. 登录安装客户端的节点。 4. 执行以下命令，切换到客户端目录，例如“/opt/hadoopclient/Kafka/kafka/bin”。 cd /opt/hadoopclient/Kafka/kafka/bin 5. 执行以下命令，配置环境变量。 source /opt/hadoopclient/bigdataenv 6. 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户。如果当前集群未启用Kerberos认证，则无需执行此命令。 kinit Kafka用户 7. 登录FusionInsight Manager，选择“集群 > 待操作的集群名称 > 服务 >ZooKeeper > 配置 > 全部配置”，搜索参数“clientPort”,记录“clientPort”的参数值。 8. 创建一个Topic： sh kafkatopics.sh create topic 主题名称 partitions主题占用的分区数 replicationfactor主题的备份个数 zookeeper ZooKeeper角色实例所在节点IP 地址 : clientPort /kafka 例如：sh kafkatopics.sh create topic TopicTestpartitions 3replicationfactor 3 zookeeper 10.10.10.100:2181/kafka 9. 执行以下命令，查询集群中的Topic信息： sh kafkatopics.sh list zookeeper ZooKeeper角色实例所在节点IP 地址:clientPort /kafka 例如：sh kafkatopics.sh list zookeeper 10.10.10.100:2181/kafka 10. 删除步骤8中创建的Topic： sh kafkatopics.sh delete topic主题名称 zookeeper ZooKeeper角色实例所在节点IP 地址 : clientPort /kafka 例如：sh kafkatopics.sh delete topic TopicTest zookeeper 10.10.10.100:2181/kafka

本章节介绍了如何删除分布式消息服务RocketMQ实例。 操作场景 分布式消息服务RocketMQ控制台支持删除RocketMQ实例，且可实现批量删除RocketMQ实例、一键式删除创建失败的RocketMQ实例。 须知： RocketMQ实例删除后，实例中原有的数据将被删除，且没有备份，请谨慎操作。 前提条件 RocketMQ实例状态为运行中、故障时，才能执行删除操作。 删除RocketMQ实例 1. 登录分布式消息服务RocketMQ控制台。 2. 选择以下任意一种方法删除实例。 在待删除的RocketMQ实例所在行，单击“更多 > 删除”。 单击待删除的RocketMQ实例名称，进入实例详情页面。在页面右上角，单击“更多 > 删除”。 如需批量删除实例，勾选待删除的实例，单击“删除”。 说明：RocketMQ实例状态为创建中、重启中时不允许执行删除操作。 3. 单击“确定”，完成删除RocketMQ实例。 删除RocketMQ实例大约需要1到60秒。 删除创建失败的RocketMQ实例 1. 登录分布式消息服务RocketMQ控制台。 如果当前存在创建失败的RocketMQ实例，信息栏会显示“创建失败任务”及失败数量。 图1 创建失败任务 2. 单击“创建失败任务”后的图标或者数量，弹出“创建失败任务”对话框。 3. 选择以下任意一种方法，删除创建失败的RocketMQ实例。 单击“清理失败任务”，一键式删除所有创建失败的RocketMQ实例。 在需要删除的创建失败的RocketMQ实例所在行，单击“删除任务”，删除创建失败的RocketMQ实例。

操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入监控菜单，下拉选择broker和brokerID，查看集群的监控信息：

说明 分布式消息服务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); } } } }}

本节介绍了如何查看分布式消息服务RabbitMQ产品实例。 实例列表 1、进入分布式消息服务RabbitMQ管理控制台查看已购买的实例列表，若列表为空，可点击右上角【创建实例】进入购买页面，创建实例详情见具体操作步骤。 2、支持按照实例ID查询，输入查询内容，点击【查询】按钮即可展示需要的实例数据。 3、查看实例基本信息，包括实例ID、规格、VPC、计费模式、创建时间、到期时间、状态。其中状态说明见下文。 运行状态 （1）登录天翼云，进入分布式消息服务RabbitMQ管理控制台。 （2）当前页面会列出所购买的RabbitMQ实例，并查看状态，状态说明如下 状态 说明 运行中 RabbitMQ实例正常运行状态。 已关闭 RabbitMQ实例处于离线的状态。 变更中 RabbitMQ实例正在进行规格变更操作。 变更失败 RabbitMQ实例处于规格变更失败状态。 暂停 RabbitMQ专享版实例处于已冻结状态，用户可以在“更多”中续费开启冻结的RabbitMQ实例。 注销 RabbitMQ实例已经过期并关闭，需要重新购买实例。

检查Kafka配置的直接内存大小 在FusionInsight Manager首页，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 配置 > 全部配置 > Broker（角色） > 环境变量”。将“KAFKAHEAPOPTS”参数中配置的“Xmx”值参考如下说明调大。 说明 建议“KAFKAHEAPOPTS”参数中“Xmx”和“Xms”值保持一致。 建议根据“Kafka直接内存资源状况”调整“KAFKAHEAPOPTS”的值为“Kafka使用的直接内存大小”的两倍（可根据实际业务场景进行修改）。“Kafka直接内存资源状况”可在FusionInsight Manager首页，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 实例”，选择上报告警实例主机名对应的角色。单击图表区域右上角的下拉菜单，选择“定制 > 进程 > Kafka直接内存资源状况”进行查看。 3. 保存配置，并重启Kafka服务。 4. 观察界面告警是否清除。 是，处理完毕。 否，执行步骤7。 收集故障信息 在FusionInsight Manager界面，选择“运维 > 日志 > 下载”。 5. 在“服务”中勾选待操作集群的“Kafka”。 6. 单击右上角的设置日志收集的“开始时间”和“结束时间”分别为告警产生时间的前后10分钟，单击“下载”。 7. 请联系运维人员，并发送已收集的故障日志信息。 告警清除 此告警修复后，系统会自动清除此告警，无需手工清除。 参考信息 无。

本节介绍云等保专区产品的专用术语。 术语 说明 :: cURL cURL是一个利用URL语法在命令行下工作的文件传输工具，可用于检测系统是否可以访问目标站点。 Dig Dig是一个在类Unix命令行模式下查询DNS信息（包括NS记录、A记录、MX记录等）的工具。 基线核查 基线核查是指对主机操作系统、数据库、软件和容器的配置进行安全检测，并提供检测结果说明和加固建议。 基线核查可以帮您进行系统安全加固，降低入侵风险并满足安全合规要求。 漏洞扫描 漏洞扫描是指基于漏洞数据库，通过扫描等手段对指定的远程或者本地计算机系统的安全脆弱性进行检测，发现可利用漏洞的一种安全检测（渗透攻击）行为。 引擎 本文的“引擎”为扫描核心技术，即最终进行漏洞扫描工作的服务。 资产 即扫描器所扫描的主机、数据库、网站等。 DDoS 分布式拒绝服务攻击（Distributed Denial of Service Attack，DDoS）是指处于不同位置的多个攻击者同时向一个或数个目标发动攻击，或者一个攻击者控制了位于不同位置的多台机器并利用这些机器对受害者同时实施攻击。由于攻击的发出点是分布在不同地方的，这类攻击称为分布式拒绝服务攻击，其中的攻击者可以有多个。 EDR 端点检测与响应（Endpoint Detection & Response，EDR）是一种主动的安全方法，可以实时监控端点，并搜索渗透到防御系统中的威胁。EDR是一种新兴的技术，可以更好地了解端点上发生的事情，提供关于攻击的上下文和详细信息。 认证 是一种信用保证形式。按照国际标准化组织（ISO）和国际电工委员会（IEC）的定义，由国家认可的认证机构证明一个组织的产品、服务、管理体系符合相关标准、技术规范（TS）或其强制性要求的合格评定活动。 虚拟机 虚拟机（Virtual Machine）指通过软件模拟的具有完整硬件系统功能的、运行在一个完全隔离环境中的完整计算机系统。 在实体计算机中能够完成的工作在虚拟机中都能够实现。 在计算机中创建虚拟机时，需要将实体机的部分硬盘和内存容量作为虚拟机的硬盘和内存容量。每个虚拟机都有独立的CMOS、硬盘和操作系统，可以像使用实体机一样对虚拟机进行操作。 AES 密码学中的高级加密标准（Advanced Encryption Standard，AES），又称Rijndael加密法，是美国联邦政府采用的一种区块加密标准。这个标准用来替代原先的DES（Data Encryption Standard），已经被多方分析且广为全世界所使用。 Apache Apache是一款Web服务器软件。它可以运行在几乎所有广泛使用的计算机上，由于其跨平台和安全性被广泛使用，是最流行的Web服务器端软件之一。 CC攻击 CC攻击（Challenge Collapsar Attack，挑战黑洞攻击）是DDoS攻击的一种类型，使用代理服务器向受害服务器发送大量假冒合法的请求，造成被攻击服务器资源耗尽，一直到宕机崩溃。 DES DES（Data Encryption Standard，数据加密标准）是一种使用密钥加密的块算法，1977年被美国联邦政府的国家标准局确定为联邦资料处理标准（FIPS），并授权在非密级政府通信中使用，随后该算法在国际上广泛流传开来。 HA 高可靠性（High Availability，简称HA）能够在通信线路或设备发生故障时提供备用方案，防止由于单个产品故障或链路故障导致网络中断，保证网络服务的连续性。 LACP LACP（Link Aggregation Control Protocol，链路聚合控制协议）是一种基于IEEE802.3ad标准的协议。 LACP协议通过LACPDU（Link Aggregation Control Protocol Data Unit，链路聚合控制协议数据单元）与对端交互信息。链路聚合往往用在两个重要节点或繁忙节点之间，既能增加互联带宽，又提供了连接的可靠性。 LDAP LDAP（Lightweight Directory Access Protocol，是轻量目录访问协议）是互联网上目录服务的通用访问协议。 LDAP服务可以有效解决众多网络服务的用户账户问题，LDAP服务器是用于查询和更新LDAP目录的服务器，包括用户账号目录。 MTU 最大传输单元（Maximum Transmission Unit，MTU）用来通知对方所能接受服务单元的最大尺寸，说明发送方能够接受的有效荷载大小。 SSL SSL（Secure Sockets Layer，安全套接字协议）及TLS（Transport Layer Security，继任者传输层安全）是为网络通信提供安全及数据完整性的一种安全协议。TLS与SSL在传输层与应用层之间对网络连接进行加密。 VRRP 虚拟路由冗余协议（Virtual Router Redundancy Protocol，简称VRRP）是由IETF提出的解决局域网中配置静态网关出现单点失效现象的路由协议，它是一种路由容错协议，也可以叫做备份路由协议。 WebShell Webshell就是以asp、php、jsp或者cgi等网页文件形式存在的一种命令执行环境，也可以将其称为一种网页后门。 黑客在入侵了一个网站后，通常会将asp或php后门文件与网站服务器Web目录下正常的网页文件混在一起，然后就可以使用浏览器来访问asp或者php后门，得到一个命令执行环境，以达到控制网站服务器的目的。 Kafka Kafka是一种高吞吐量的分布式发布订阅消息系统，可以处理消费者规模的网站中所有动作流数据。这些数据通常由于吞吐量要求而通过处理日志和日志聚合来解决。 SNMP SNMP（Simple Network Management Protocol，简单网络管理协议）是标准IP网络管理协议，支持目前主流的网络管理系统。 SQL SQL（Structured Query Language，结构化查询语言）是一种数据库查询和程序设计语言，用于存取数据以及查询、更新和管理关系数据库系统；同时也是数据库脚本文件的扩展名。 Syslog Syslog是一种行业标准的协议，可用来记录设备的日志。 Syslog日志消息既可以记录在本地文件中，也可以通过网络发送到接收Syslog的服务器。服务器可以对多个设备的Syslog消息进行统一的存储，或者解析其中的内容做相应的处理。常见的应用场景是网络管理工具、安全管理系统、日志审计系统。

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

本章节主要介绍如何快速创建Kafka流式集群。 本章节为您介绍如何快速创建一个Kafka流式集群，Kafka集群使用Kafka和Storm组件提供一个开源高吞吐量，可扩展性的消息系统。广泛用于日志收集、监控数据聚合等场景，实现高效的流式数据采集，实时数据处理存储等。 快速创建Kafka流式集群 1.登录MRS管理控制台。 2.单击“创建集群”，进入“创建集群”页面。 3.在创建集群页面，选择“快速创建”页签。 4.参考下列参数说明配置集群基本信息，参数详细信息请参考创建自定义集群。 区域：默认即可。 集群名称：可以设置为系统默认名称，但为了区分和记忆，建议带上项目拼音缩写或者日期等。例如：“mrs20200321”。 版本类型：默认选择普通版（不同版本提供的组件有所不同，请根据需要选择版本类型）。 集群版本：不同版本集群提供的组件有所不同，请根据需要选择集群版本。 组件选择：选择“Kafka流式集群”。 可用区：默认即可。 虚拟私有云：默认即可。如果没有虚拟私有云，请单击“查看虚拟私有云”进入虚拟私有云，创建一个新的虚拟私有云。 子网：默认即可。 CPU架构：默认即可。 集群节点：请根据自身需要选择集群节点规格数量等。MRS 3.x及之后版本集群Master节点规格不能小于64GB。 集群高可用：默认即可。MRS 3.x版本暂时不支持该参数。 LVM：默认即可。MRS 3.x版本暂时不支持该参数。 用户名：默认为“root/admin”，root用于远程登录ECS机器，admin用于登录集群管理页面。 密码：设置root用户和admin用户密码。 确认密码：再次输入设置的root用户和admin用户密码。 5.勾选“确认授权”开通通信安全授权，通信安全授权详情请参考授权安全通信。 6.单击“立即申请”。 当集群开启Kerberos认证时，需要确认是否需要开启Kerberos认证，若确认开启请单击“继续”，若无需开启Kerberos认证请单击“返回”关闭Kerberos认证后再创建集群。 7.单击“返回集群列表”，可以查看到集群创建的状态。单击“访问集群”，可以查看集群详情。 集群创建的状态过程请参见 集群概览章节查看集群状态部分 集群列表参数中的“状态”参数说明。 集群创建需要时间，所创集群的初始状态为“启动中”，创建成功后状态更新为“运行中”，请您耐心等待。 MRS系统界面支持同一时间并发创建10个集群，且最多支持管理100个集群。

有序消息 消费消息的顺序要同发送消息的顺序一致，在RocketMQ中，主要有两种有序消息。 普通有序消息 有序消息的一种，在正常情况下可以保证完全的顺序消息，但是一旦发生通信异常，Broker重启，由于队列总数发生变化，哈希取模后定位的队列会变化，产生短暂的消息顺序不一致。如果业务能容忍在集群异常情况（如某个Broker宕机或者重启）下，消息短暂的乱序，使用普通顺序方式比较合适。 严格有序消息 有序消息的一种。无论正常异常情况都能保证顺序，但是牺牲了分布式Failover特性，即Broker集群中只要有一台机器不可用，则整个集群都不可用（或者影响hash值对应队列的使用），服务可用性大大降低。如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。若业务能容忍短暂乱序，推荐普通有序消费。

本文介绍分布式消息服务RocketMQ入门指引的创建应用用户和密码内容。 背景信息 新建消息实例后，必须新建应用用户，然后应用才能在此消息实例上发送、消费消息。 应用用户：指MQ客户端，连接服务器生产消费时，需要进行权限校验，所以MQ客户端的用户，称为应用用户；除了用户密码的校验，还可以为用户指定Topic，代表该用户只能生产消费指定的Topic，其他Topic不能生产消费。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、进入应用用户管理（旧模式）或者角色控制（新模式）菜单。 （1）旧模式维护用户及权限 以下适用于上海6、北京4、内蒙5、西安3、重庆2、拉萨3、南京3、雄安2、晋中、郴州2、成都4、杭州2、上海7、西安4、福州3、泉州1、芜湖2、北京5、中卫2、贵州3、九江、内蒙6、武汉4、佛山3、福州4、昆明2、保定、乌鲁木齐27、辽阳1节点。 1）点击【新建用户】按钮 2）进入用户列表界面，新增用户 3）弹出框填写用户字段 默认展示租户名，不可修改。 选择集群名称，填写应用用户名，请输入大于6位字符，只能输入大小写字母，下划线，数字。 填写用户密码，请输入大于8位字符，需要包含数字大小写字母以及特殊符号(!@

本文主要介绍分布式消息服务RabbitMQ的约束与限制。 分布式消息服务RabbitMQ在某些功能存在一些约束和限制，详细如表1所示。 表1 RabbitMQ使用约束和限制 限制项 约束和限制 描述 版本 当前服务端版本为3.8.35。 兼容AMQP 091协议的客户端版本。 连接数 RabbitMQ单机和集群实例，不同实例规格的连接数上限不一致，具体限制，请参考产品规格。 通道数 5GB 磁盘剩余空间低于5GB会触发磁盘高水位，生产者流程被阻塞。 clusterpartitionhandling pauseminority 当集群发生网络分区时，代理会检查自己是否处于“少数派”（存储分区的代理数小于等于总代理数的一半称为少数派）。少数派中的代理将会自动关闭服务并定期检测网络状态，待分区恢复之后重新启动服务。如果未开启镜像队列，发生分区时少数派上的队列将无法生产消费。此策略相当于放弃了可用性而选择了数据一致性。 rabbitmqdelayedmessageexchange 插件延迟时间存在1%左右的误差，可能提前或者推迟发送消息给消费者。 实例是否开启消息延迟功能。 RabbitMQ插件 RabbitMQ插件功能可用于测试和迁移业务等场景，不建议用于生产业务。 RabbitMQ实例主要提供AMQP 091业务消息的功能，兼容相关协议的Vhost、Exchange等组件。插件相关内容为非核心功能，不建议用于生产业务。

设备偶尔出现“connection reset by peer”错误。 当两个客户端具有相同的客户端ID（Client ID）时，MQTT Broker会根据规范的要求进行互踢（Kickoff），即断开其中一个客户端的连接。这是因为MQTT协议要求每个客户端ID必须是唯一的，用于标识不同的客户端。 当发生客户端ID重复的情况时，MQTT Broker会选择保持最新连接的客户端，而断开旧连接的客户端。这是为了确保每个客户端ID都对应唯一的客户端连接，并避免潜在的冲突和数据混乱。 当旧连接的客户端被断开后，它可能仍然尝试发送数据到已经断开的TCP连接。这会导致对端Socket发送RST（Reset）报文，通知客户端连接已经被重置或关闭。这是正常的网络行为，用于清除已经断开的连接的残留数据。 资源报表的最大查询范围是多少？ 分布式消息服务MQTT支持查询7天以内的指标数据。 为什么设备都断电了，控制台还显示设备在线？ 分布式消息服务MQTT会定时检查客户端心跳，当系统检测到客户端超过指定时长（1.5keepalive时长）没有心跳，则会断开连接。 如何在控制台修改订阅关系？ 订阅关系在订阅消息的SDK代码中设置，控制台只能查询。 本地测试先发送消息，再订阅接收不到消息，设置离线消息后还是接收不到。 若先启动生产者发送消息，如果Topic从未有过订阅，则即使设置了离线消息（QoS1且cleanSessionfalse），消费者启动后也不会收到离线消息；如果Topic被消费者订阅过，消费者离线后，若设置了离线消息，这些离线消息会在消费者再次上线后被消费到。

本章节主要介绍ALM38007 Kafka默认用户状态异常的告警。 告警解释 系统每60秒周期性检测Kafka服务默认用户，当检测到该用户异常时发送此告警。 平滑次数为1，当用户状态恢复后，告警恢复。 告警属性 告警ID 告警级别 是否自动清除 38007 紧急 是 告警参数 参数名称 参数含义 来源 产生告警的集群名称。 服务名 产生告警的服务名称。 角色名 产生告警的角色名称。 主机名 产生告警的主机名称。 Trigger Condition Kafka默认用户状态异常。 对系统的影响 Kafka默认用户状态异常，会影响Broker之间的元数据同步，以及Kafka与ZooKeeper之间的交互，进而影响业务生产、消费和Topic的创建、删除等操作。 可能原因 Sssd服务异常导致。 部分Broker实例停止运行。 处理步骤 检查是否有"Sssd 服务异常" 告警 在FusionInsight Manager首页，选择“运维 > 告警 > 告警 > Kafka默认用户状态异常 > 定位信息”。查看告警上报的实例的主机名。 1. 根据告警提示的主机信息，登录到该节点上。 2. 执行 id Gn kafka ，查看返回结果是否报"No such user"。 是，记录当前节点主机名，并执行步骤4。 否，执行步骤6。 3. 在FusionInsight Manager首页，选择“运维 > 告警 > 告警”。查看所有告警信息中是否有"Sssd服务异常"告警，根据对应的告警指导进行处理。

本文介绍分布式消息服务RocketMQ的消费者实例查询操作内容。 场景描述 RocketMQ显示当前在线的消费者实例有以下作用： 监控消费者状态：通过显示当前在线的消费者实例，可以实时监控消费者的状态。可以了解消费者的连接情况、消费进度和消费速率等信息，帮助及时发现消费者异常或故障，以便进行及时处理和调整。 故障排查与追踪：通过显示当前在线的消费者实例，可以帮助定位消息消费失败的原因。可以查看每个消费者实例的消费情况，包括消费的消息数量、消费延迟等信息，方便排查故障和进行问题定位。 综上所述，显示当前在线的消费者实例可以帮助监控消费者状态、故障排查等，为系统的稳定运行提供重要支持。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入消费者实例查询菜单，列出了指定集群和Broker下的消费组消费的情况。 列表展示了指定集群和指定Broker下的消费组消费的情况。 连接实例 显示该消费组，当前在线的消费者实例列表。 堆积量 显示该消费组消费指定Topic时，还有多少未消费。

本文为您分布式消息服务MQTT的环境准备内容。 在创建天翼云分布式消息服务MQTT实例之前，您需要做一些准备工作。 首先，您需要设置一个虚拟私有云（VPC），这是一个隔离的网络环境，用于托管MQTT实例。 接下来，您需要创建一个子网，它是VPC内部的一个子网络，用于划分不同的部分和区域。 最后，您需要配置一个安全组，用于控制入站和出站的流量规则，以保证MQTT实例的安全性。 每个分布式消息服务MQTT实例都会被部署在特定的VPC中，并与特定的子网和安全组相关联。这种方式可以让您自主配置和管理MQTT实例的网络环境，并提供安全保护策略。如果您已经有了现成的VPC、子网和安全组，可以重复使用它们，无需重复创建。这样可以节省时间和资源，并确保一致性和可靠性。 VPC和子网 VPC和子网可重复使用，您也可以使用不同的VPC和子网来配置MQTT实例，您可根据实际需要进行配置。在创建VPC和子网时应注意如下要求： VPC与使用的天翼云分布式消息服务MQTT服务应在相同的区域。 如无特殊需求，创建VPC和子网的配置参数使用默认配置即可。 创建VPC和子网的操作请参考虚拟私有云创建VPC、子网搭建私有网络。 若需要在已有VPC上创建和使用新的子网，请参考虚拟私有云子网管理创建子网。

本章节介绍了如何创建分布式消息服务RocketMQ实例的Topic。 操作场景 Topic，即消息主题，消息发送与接收的基本单元。购买RocketMQ实例成功后，需要手动创建Topic，然后才能进行生产消息和消费消息。 前提条件 已购买RocketMQ实例。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 单击“创建Topic”，弹出“创建Topic”页面。 5. 参考表1，填写Topic名称和配置信息。 表1 Topic参数说明 参数 说明 Topic名称 您可以自定义Topic名称，用于区分不同的Topic。 Topic名称不能设置为相同，否则会创建失败。创建Topic后不能修改名称。 说明：如果Topic名称中包含“%”或“ 权限 Topic的权限，包括发布+订阅、发布和订阅。 关联代理 选择在某个代理上创建Topic，并设置Topic的队列个数。如果实例部署在多个代理上，单击“添加关联代理”，可以在其他代理上创建Topic，并设置Topic的队列个数。 6. 参考图1 创建Topic 7. 配置完成后，单击“确定”，完成Topic的创建。

Kafka高危操作 操作名称 操作风险 风险等级 规避措施 重大操作观察项目 删除Topic 该操作将会删除已有的主题和数据 ★★★ 采用Kerberos认证，保证合法用户具有操作权限，并确保主题名称正确 观察主题是否正常处理 删除数据目录 该操作将会导致业务信息丢失 ★★★ 请勿手动删除数据目录 观察数据目录是否正常 修改数据目录下内容（创建文件、文件夹） 该操作将会导致该节点上的Broker实例故障 ★★★ 请勿手动在数据目录下创建或修改文件及文件夹 观察数据目录是否正常 修改磁盘自适应功能“disk.adapter.enable”参数 该操作会在磁盘使用空间达到阈值时调整Topic数据保存周期，超出保存周期的历史数据可能被清除 ★★★ 若个别Topic不能做保存周期调整，将该Topic配置在“disk.adapter.topic.blacklist”参数中 在KafkaTopic监控页面观察数据的存储周期 修改数据目录“log.dirs”配置 该配置不正确将会导致进程故障 ★★★ 确保所修改或者添加的数据目录为空目录，且权限正确 观察数据目录是否正常 减容Kafka集群 该操作将会导致部分Topic数据副本数量减少，可能会导致Topic无法访问 ★★ 请先做好数据副本转移工作，然后再进行减容操作 观察分区所在备份节点是否都存活，确保数据安全 单独启停基础组件 该操作将会影响服务的一些基础功能导致业务失败 ★★★ 请勿单独启停ZooKeeper/Kerberos/LDAP等基础组件，启停基础组件请勾选关联服务 观察服务状态是否正常 重启/停止服务 该操作将会导致业务中断 ★★ 确保在必要时重启/停止服务 观察服务是否运行正常 修改配置参数 该操作将需要重启服务使得配置生效 ★★ 确保在必要时修改配置 观察服务是否运行正常 删除/修改元数据 修改或者删除ZooKeeper上Kafka的元数据可能导致Topic或者Kafka服务不可用 ★★★ 请勿删除或者修改Kafka在ZooKeeper上保存的元数据信息 观察Topic或者Kafka服务是否可用 修改元数据备份文件 修改Kafka元数据备份文件，并被使用进行Kafka元数据恢复成功后，可能导致Topic或者Kafka服务不可用 ★★★ 请勿修改Kafka元数据备份文件 观察Topic或者Kafka服务是否可用

Kafka高危操作 操作名称 操作风险 风险等级 规避措施 重大操作观察项目 删除Topic 该操作将会删除已有的主题和数据。 ★★★ 采用Kerberos认证，保证合法用户具有操作权限，并确保主题名称正确。 观察主题是否正常处理。 删除数据目录 该操作将会导致业务信息丢失。 ★★★ 请勿手动删除数据目录。 观察数据目录是否正常。 修改数据目录下内容（创建文件、文件夹） 该操作将会导致该节点上的Broker实例故障。 ★★★ 请勿手动在数据目录下创建或修改文件及文件夹。 观察数据目录是否正常。 修改磁盘自适应功能“disk.adapter.enable”参数 该操作会在磁盘使用空间达到阈值时调整Topic数据保存周期，超出保存周期的历史数据可能被清除。 ★★★ 若个别Topic不能做保存周期调整，将该Topic配置在“disk.adapter.topic.blacklist”参数中。 在KafkaTopic监控页面观察数据的存储周期。 修改数据目录“log.dirs”配置 该配置不正确将会导致进程故障。 ★★★ 确保所修改或者添加的数据目录为空目录，且权限正确。 观察数据目录是否正常。 减容Kafka集群 该操作将会导致部分Topic数据副本数量减少，可能会导致Topic无法访问。 ★★ 请先做好数据副本转移工作，然后再进行减容操作。 观察分区所在备份节点是否都存活，确保数据安全。 单独启停基础组件 该操作将会影响服务的一些基础功能导致业务失败。 ★★★ 请勿单独启停ZooKeeper/Kerberos/LDAP等基础组件启停基础组件请勾选关联服务。 观察服务状态是否正常。 重启/停止服务 该操作将会导致业务中断。 ★★ 确保在必要时重启/停止服务。 观察服务是否运行正常。 修改配置参数 该操作将需要重启服务使得配置生效。 ★★ 确保在必要时修改配置。 观察服务是否运行正常。 删除/修改元数据 修改或者删除ZooKeeper上Kafka的元数据可能导致Topic或者Kafka服务不可用。 ★★★ 请勿删除或者修改Kafka在ZooKeeper上保存的元数据信息。 观察Topic或者Kafka服务是否可用。 修改元数据备份文件 修改Kafka元数据备份文件，并被使用进行Kafka元数据恢复成功后，可能导致Topic或者Kafka服务不可用。 ★★★ 请勿修改Kafka元数据备份文件。 观察Topic或者Kafka服务是否可用。

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

本章节主要介绍分布式消息服务RabbitMQ按需转包周期。 按需付费的用户可以选择“转包周期”，变更实例计费模式为包年/包月。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 此处请选择RabbitMQ实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”>“分布式消息服务”>“RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 步骤 4 通过以下任意一种方法，实现按需实例转包周期。 勾选RabbitMQ实例名称左侧的方框，可选一个或多个，单击信息栏左上侧的“转包周期”。 在待转包周期的RabbitMQ实例所在行，单击“更多 > 转包周期”。 单击RabbitMQ实例名称，进入实例详情页面。单击右上角的“更多 > 转包周期”。 步骤 5 单击“确定”，页面跳转到“按需转包周期”页面。 步骤 6 选择续费时长，然后单击“去支付”，根据界面提示信息，支付费用，完成实例转包周期操作。

本章节主要如何介绍修改分布式消息服务RabbitMQ的实例信息。 创建RabbitMQ实例成功后，您可以根据自己的业务情况对RabbitMQ实例的部分参数进行调整。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 此处请选择RabbitMQ实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”>“分布式消息服务”>“RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 步骤 4 单击RabbitMQ实例的名称，进入实例详情页面。 步骤 5 以下参数支持修改。 实例名称 描述 企业项目 安全组 公网访问（公网访问的修改方法，请参考设置实例的公网访问。） 参数修改完成后，通过以下方式查看修改结果。 修改“公网访问”，系统跳转到“后台任务管理”页签，并显示当前任务的操作进度和结果。 修改“实例名称”、“描述”、“企业项目”和“安全组”后，右上角直接提示修改结果。