本文为您介绍分布式消息服务MQTT的API第三方认证授权更新MySQL认证授权。 POST ContentType：application/json 请求体参数说明： 参数名 参数类型 是否必填 参数描述 id Integer 是 id ip String 是 ip port Integer 是 端口 dbName String 是 数据库名 accessAccount String 是 用户名 accessKey String 是 密码 authKeyCryType String 是 认证密码类型，plain和sha256 prodInstId String 是 实例id 返回参数说明： 参数名 参数类型 参数描述 message String 描述状态 statusCode Integer 800成功 其他失败 返回示例：成功 { "message": "success","statusCode": "800"} 返回示例：失败 { "message": "success","statusCode": "900"}

本章节主要介绍如何查看分布式消息服务RabbitMQ的监控数据。 操作场景 云监控对RabbitMQ实例的运行状态进行日常监控，可以通过控制台直观的查看RabbitMQ实例各项监控指标。 前提条件 已创建RabbitMQ专享版实例，且实例中有可消费的消息。 操作步骤 1. 登录管理控制台。 2. 在管理控制台右上角单击，选择区域。 说明 此处请选择与您的应用服务相同的区域。 3. 在管理控制台左上角单击，选择“企业中间件”>“分布式消息服务”>“RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 4. 通过以下任意一种方法，查看监控数据。 在RabbitMQ实例名称后，单击“查看监控数据”。跳转到云监控页面，查看实例、节点和队列的监控数据，数据更新周期为1分钟。 单击RabbitMQ实例名称，进入实例详情页。在左侧导航栏单击“监控”，进入监控页面，查看实例、节点和队列的监控数据，数据更新周期为1分钟。 在“队列”的监控页面中，如果队列在默认Vhost下，会直接显示队列名。如果队列不在默认Vhost下，队列名称显示为：队列所在的Vhost名称队列名。例如：队列test01在Vhost13142708中，此时监控中显示的队列名为“Vhost13142708test01”。 图1 队列的监控页面

灵活及时 队列处理能力支持按需自动扩展，及时且方便地完成系统扩展，消息投递时间可至毫秒级，从而保证消息及时性。分布式消息服务RocketMQ具有很高的灵活性，可以满足各种不同的业务需求。主要体现： 支持多种消息模型：RocketMQ支持多种消息模型，包括消息队列模型和发布/订阅模型。在消息队列模型中，消息发送方将消息发送到一个队列，消息接收方从队列中读取消息。这种模型适用于顺序消息和事务性消息等场景。而在发布/订阅模型中，消息发送方将消息发布到一个主题，所有订阅该主题的消费者都会收到消息。这种模型适用于实时通知、数据分发等场景。 支持灵活的消息过滤机制：RocketMQ可以通过对消息的属性进行过滤，只有满足条件的消息才会被消费者接收。这样可以实现消息的动态路由和选择性消费，提升系统的灵活性和效率。 支持消息延迟发送和定时消费：RocketMQ可以设置消息的延迟时间，使消息在指定的时间后才能被消费者接收。这对于实现定时任务和延迟处理非常有用。 良好的可扩展性：RocketMQ采用了分布式架构，并且支持主从复制和消息分区机制。可以根据业务需求，动态扩展消息生产者、消息消费者和消息存储节点的数量，以满足大规模消息处理和高并发访问的要求。 高可靠 集群节点采用主备模式，具有主备故障自动切换功能；并且提供对消息的持久化能力，多副本冗余；提供消息数据自动删除功能。分布式消息服务RocketMQ具有高可靠性的特点，以下是RocketMQ实现高可靠性的关键特性： 主从复制：RocketMQ采用了主从复制的架构，在生产者发送消息时，消息会首先写入主节点，并异步复制到多个从节点。这样即使主节点发生故障，从节点也能够接管并继续提供服务。 可靠消息存储：RocketMQ使用Write Ahead Log (WAL)技术来保证消息的可靠存储。在消息写入之前，会先将消息写入磁盘的顺序文件中，然后再写入内存。当RocketMQ重启时，可以通过检查磁盘上的文件来恢复之前未被消费的消息。 消息可重复存储：RocketMQ使用消息的唯一ID来确保消息的幂等性。如果一条消息因为网络问题或其他原因发送失败，RocketMQ可以根据消息的ID判断是否已经成功发送过，并避免消息的重复发送。 容灾备份：RocketMQ支持Broker集群模式和多数据中心的部署方式，可以将消息数据进行容灾备份。当某个Broker节点发生故障时，其他节点可以继续提供服务，确保系统的可用性。 高可用性设计：RocketMQ采用了多个Broker节点组成的集群，并通过主从复制和故障切换来实现高可用性。当某个Broker节点发生故障时，其他节点会自动接管其工作，保证消息的正常处理。

本文为您介绍分布式消息服务MQTT的API设备客户端查询设备订阅信息。 GET URI参数说明： 参数名 参数类型 是否必填 参数描述 prodInstId String 是 实例ID clientId String 是 设备ID 返回参数说明： 参数名 参数类型 参数描述 statusCode Integer 800成功 其他失败 message String 描述状态 returnObj Object 返回对象 returnObj.node String 设备节点 returnObj.clientid String 设备ID returnObj.qos Integer qos级别 returnObj.topic String 主题 返回示例：成功 {"statusCode": 800,"message": "success","returnObj": [{"node": "MQTT2@172.31.0.217","clientid": "DEFAULT@TESTSUB2","qos": 1,"topic": "test/aaa"}]} 返回示例：失败 {"returnObj": null,"message": "error","statusCode": "900"}

本文为您介绍分布式消息服务MQTT的API第三方认证授权配置MySQL认证授权。 POST ContentType：application/json 请求体json参数说明： 参数名 参数类型 是否必填 参数描述 ip String 是 mysql ip port Integer 是 端口 accessAccount String 是 mysql用户 accessKey String 是 mysql用户密码 dbName String 是 数据库名 authKeyCryType String 是 认证密码类型，plain和sha256 prodInstId String 是 实例id 返回参数说明： 参数名 参数类型 参数描述 message String 描述状态 statusCode Integer 800成功 其他失败 返回示例： { "message": "success","statusCode": "800"} 返回示例：失败 {"returnObj": null,"message": "error","statusCode": 900}

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

本文为您介绍分布式消息服务MQTT的终端节点。 终端节点（Endpoint）即调用API的 请求地址 ，不同服务不同区域的终端节点不同。 Endpoint mqttglobal OpenAPI接入，请使用以下终端节点（Endpoint）：

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

本文为您介绍分布式消息服务MQTT的API认证授权查认证用户ACL。 GET URI参数说明： 参数名 参数类型 是否必填 参数描述 prodInstId String 是 实例ID userName String 是 认证用户 返回参数说明： 参数名 参数类型 参数描述 statusCode Integer 800成功 其他失败 message String 描述状态 returnObj Object 返回对象 returnObj.allow String allow 动作权限 true false returnObj.topic String 主题 returnObj.action String action 动作 pub、sub、pubsub returnObj.login String 认证用户 返回示例：成功 {"statusCode": 200,"message": "success","returnObj": [{"allow": true,"topic": "","action": "pubsub","login": "test"}]} 返回示例：失败 {"returnObj": null,"message": "error","statusCode": "900"}

安装依赖 plaintext pip3 install pahomqtt 代码示例 python import ssl import random from paho.mqtt import client as pahoclient 填入您在mqtt控制台创建的ACL账号密码。 USERNAME "yourusername" AUTHPASSWORD "yourpassword" 是否使用tls加密传输 isTls True class MQTTClient: def init(self): 填写mqtt云消息服务的接入点，去掉:{端口号}部分 self.broker "tcp://localhost" 指定连接客户端的id self.clientid "ctgmqttclienttest" self.client None def onconnect(self, client, userdata, flags, rc): 连接建立成功 if rc 0: print("Connected to MQTT Broker!") else: print(f"Failed to connect, return code {rc}") def ondisconnect(self, client, userdata, rc): 连接丢失 print(f"Disconnected with result code {rc}") 可以在这里添加重连逻辑 def onmessage(self, client, userdata, msg): 收到消息的回调 print(f"Received {msg.payload.decode()} from {msg.topic} topic") def onpublish(self, client, userdata, mid): 成功发送消息到服务端 print(f"Message {mid} published successfully") def connectmqtt(self): 创建客户端实例 self.client pahoclient.Client(clientidself.clientid) self.client.usernamepwset(USERNAME, AUTHPASSWORD) 设置回调函数 self.client.onconnect self.onconnect self.client.ondisconnect self.ondisconnect self.client.onmessage self.onmessage self.client.onpublish self.onpublish 设置TLS if isTls: 创建不验证证书的SSL上下文 context ssl.createdefaultcontext() context.checkhostname False context.verifymode ssl.CERTNONE self.client.tlssetcontext(context) 设置其他连接选项 self.client.connect(self.broker.replace("tcp://", "").replace("ssl://", ""), port8883 if isTls else 1883) self.client.loopstart()

本文为您介绍分布式消息服务MQTT的API参考调用前须知。 本说明提供了分布式消息服务MQTT产品API的描述、语法、参数说明及示例等内容。

查询消息轨迹 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“消息查询”，进入“消息查询”页面。 4. 选择以下任意一种方法，查询消息。 按Topic查询：“Topic”选择待查询消息的Topic名称，“队列”选择待查询消息的队列，“存储时间”选择待查询消息的时间段，单击“查询”。 按Message ID查询：“Topic”选择待查询消息所在的Topic名称，“消息ID”输入待查询消息的Message ID，单击“查询”。 按Message Key查询：“Topic”选择待查询消息所在的Topic名称，“消息ID”输入待查询消息的Message Key，单击“查询”。 5. 在待查询消息所在行，单击“消息轨迹”，查看消息的轨迹，确定是否生产/消费成功。 消息轨迹的参数说明如表1所示。 表1 消息轨迹的参数说明 参数 参数说明 生产者状态 生产者状态如下： 1.发送成功：消息发送成功，服务端已经成功存储消息。 2.提交成功：允许消费者消费此事务消息。 3.回滚：事务消息将被丢弃，不允许消费者消费此事务消息。 4.未知，待确认：事务消息状态暂时无法确定，等待固定时间后，服务端向生产者进行消息回查。 生产耗时 生产者发送消息的耗时。 生产地址 生产者的IP地址和端口号。 消费者状态 消费者状态如下： 1.消费成功消费超时消费异常 2.消费返回NULL消费失败 消费时间 消费消息的时间。 消费耗时 消费者消费消息的耗时。 消费地址 消费者的IP地址和端口号。

本文为您介绍分布式消息服务MQTT的API设备客户端查设备轨迹。 GET URI参数说明： 参数名 参数类型 是否必填 参数描述 prodInstId String 是 实例Id clientId String 是 设备ID startTime String 是 起始时间，时间格式"yyyyMMddHHmmss" endTime String 是 截止时间，时间格式"yyyyMMddHHmmss"时间段不支持跨天，如果 startTime小于endTime当天起始时间，则以endTime那天起始时间为准查询 返回参数说明 参数名 参数类型 参数描述 returnObj Object 返回对象 returnObj.clientId String 设备ID returnObj.action String 动作connect、disconnect returnObj.time String 时间 yyyyMMdd HH:mm:ss message String 描述状态 statusCode Integer 800成功 其他失败 返回示例：成功 {"returnObj": [{“clientId”:"aaa","action":"connect","time":"20220911 11:22:33},{“clientId”:"aaax","action":"disconnect","time":"20220911 11:22:33}]"message": "success","statusCode": "800"} 返回示例：失败 {"returnObj": null,"message": "error","statusCode": "900"}

总结 开源 RabbitMQ 不提供原生延时语义 可通过两种方式实现延时消息能力： 死信队列 + TTL：通用、稳定，但灵活性有限 延时消息插件：精度高、使用简单，但依赖部署环境 实际选型需综合考虑： 延迟精度要求 运维复杂度 消息规模与资源成本

RabbitMQ实例是否支持持久化，如何定时备份数据？ RabbitMQ支持消息数据持久化，可从客户端连接RabbitMQ并设置消息持久化，也可在RabbitMQ集群管理工具界面创建队列时设置消息持久化。 不支持客户自定义定时备份数据，或从界面触发备份数据。 RabbitMQ实例开启SSL后，证书怎么获取？ RabbitMQ实例开启SSL后只做单向认证，不需要证书。 RabbitMQ实例的SSL开关是否支持修改？ 不支持动态修改，即如果实例创建时没有选择开启，创建完成之后，不支持修改，建议在实例创建时将开关打开。 RabbitMQ实例是否支持扩容？ 单机版本的RabbitMQ实例支持扩大存储空间，以及扩容代理规格，暂不支持缩容。 集群版本的RabbitMQ实例支持扩大存储空间和代理个数，以及扩容代理规格，暂不支持缩容。 如何清空队列数据？ 1. 登录Web UI。 2. 在“Queues”页签，单击需要清空数据的队列名称，进入队列详情页面。 3. 单击“Purge Messages”，清空队列数据。 RabbitMQ支持双向认证吗？ 不支持。 RabbitMQ支持升级CPU和内存吗？ RabbitMQ支持扩容/缩容代理规格，具体请参见变更实例规格。 如何关闭RabbitMQ的WebUI？ 创建RabbitMQ实例后，如果想要关闭RabbitMQ的WebUI，只要您在安全组入方向中不开放15672端口（实例未开启SSL时的端口）或者15671（实例开启SSL时的端口），此时就无法登录WebUI界面。

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

本文为您介绍分布式消息服务MQTT的API第三方认证授权查询MySQL认证授权配置。 GET URI参数说明： 参数名 参数类型 是否必填 参数描述 prodInstId String 是 实例ID 返回参数说明： 参数名 参数类型 参数描述 returnObj Object 返回对象 returnObj.id Integer id returnObj.ip String ip returnObj.port Integer 端口 returnObj.dbName String 数据库名 returnObj.accessAccount String mysql用户名 returnObj.authKeyCryType String 认证密码类型，plain和sha256 returnObj.prodInstId String 实例ID returnObj.enableStatus Integer enableStatus 0停用 1启用 returnObj.updateTime String 更新时间 返回示例：成功 {"returnObj": [{"id": 2,"ip": "127.0.0.1","port": 3323,"dbName": "test","accessAccount": "test","authKeyCryType": "plain","salt": null,"tenantId": 3127,"prodInstId": "7661f2c0afa849258b4e9f4038142f9d","enableStatus": 0,"updateTime": "20220823 10:41:02"}]"message": "success","statusCode": "800"} 返回示例：失败 {"returnObj": null,"message": "error","statusCode": 900}

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

消费消息 import com.rabbitmq.client.; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; public class RabbitmqConsumer { //队列名称 private final static String QUEUENAME "helloMQ"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接工厂 ConnectionFactory factory new ConnectionFactory(); //设置主机ip factory.setHost("127.0.0.1"); //设置amqp的tcp端口号 factory.setPort(5672); //设置用户名密码 factory.setUsername("YOUR USERNAME"); factory.setPassword("YOUR PASSWORD"); //设置Vhost factory.setVirtualHost("/"); //基于网络环境合理设置超时时间 factory.setConnectionTimeout(30 1000); factory.setHandshakeTimeout(30 1000); factory.setShutdownTimeout(0); Connection connection factory.newConnection(); Channel channel connection.createChannel(); //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。 channel.queueDeclare(QUEUENAME, false, false, false, null); System.out.println(" [] Waiting for messages. To exit press CTRL+C"); Consumer consumer new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message new String(body, StandardCharsets.UTF8); System.out.println("Received message: '" + message + "'"); } }; channel.basicConsume(QUEUENAME, true, consumer); } } SSL生产消息 import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DefaultSaslConfig; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import java.io.FileInputStream; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.security.; import java.security.cert.CertificateException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class RabbitmqProducerSsl { // private final static String EXCHANGENAME "exchangeTest"; private final static String QUEUENAME "helloMQ"; // private final static String ROUTINGKEY "test"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { // 创建连接工厂 ConnectionFactory factory new ConnectionFactory(); // 设置主机ip factory.setHost("127.0.0.1"); // 设置amqp的ssl端口号 factory.setPort(5671); String ksFile "/sslpath/ssl/clientrabbitmqkey.p12"; String tksFile "/sslpath/ssl/truststore"; SSLContext c null; try { char[] keyPassphrase "YOUR PASSPHRASE".toCharArray(); KeyStore ks KeyStore.getInstance("PKCS12"); ks.load(new FileInputStream(ksFile), keyPassphrase); KeyManagerFactory kmf KeyManagerFactory.getInstance("SunX509"); kmf.init(ks, keyPassphrase); char[] trustPassphrase "YOUR PASSPHRASE".toCharArray(); KeyStore tks KeyStore.getInstance("JKS"); tks.load(new FileInputStream(tksFile), trustPassphrase); TrustManagerFactory tmf TrustManagerFactory.getInstance("SunX509"); tmf.init(tks); c SSLContext.getInstance("tlsv1.2"); c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); } catch (KeyStoreException e) { throw new RuntimeException(e); } catch (NoSuchAlgorithmException e) { throw new RuntimeException(e); } catch (CertificateException e) { throw new RuntimeException(e); } catch (UnrecoverableKeyException e) { throw new RuntimeException(e); } catch (KeyManagementException e) { throw new RuntimeException(e); } factory.setSaslConfig(DefaultSaslConfig.EXTERNAL); factory.useSslProtocol(c); // 设置Vhost，需要在控制台先创建 factory.setVirtualHost("/"); // 基于网络环境合理设置超时时间 factory.setConnectionTimeout(30 1000); factory.setHandshakeTimeout(30 1000); factory.setShutdownTimeout(0); // 创建一个连接 Connection connection factory.newConnection(); // 创建一个频道 Channel channel connection.createChannel(); // 指定一个队列 channel.queueDeclare(QUEUENAME, false, false, false, null); for (int i 0; i < 100; i++) { // 发送的消息 String message "Hello rabbitMQ!" + i; // 往队列中发送一条消息，使用默认的交换器 channel.basicPublish("", QUEUENAME, null, message.getBytes(StandardCharsets.UTF8)); System.out.println(" Sent message: '" + message + "'"); TimeUnit.MILLISECONDS.sleep(100); } //关闭频道和连接 channel.close(); connection.close(); } }

可信云服务可以通过IAM委托的方式访问其他云服务的资源。可信实体为天翼云服务的IAM委托，包括普通云服务委托和云服务关联委托。本文介绍事件总线EventBridge的服务内联委托。 什么是服务内联委托 在某些场景下，事件总线EventBridge为了完成自身的某个功能，需要获取其他云服务的访问权限，因此，事件总线EventBridge创建了与云服务内联委托，即服务内联委托CtyunAssumeRoleForEventBridge。 使用事件总线EventBridge，系统提供的服务内联委托及其包含的系统权限策略如下： 服务内联委托：CtyunAssumeRoleForEventBridge 系统权限策略：CtyunAssumePolicyForEventBridge CtyunAssumeRoleForEventBridge 服务内联委托CtyunAssumeRoleForEventBridge具有获取访函数列表、函数详情以及调用函数的权限；具有对分布式消息服务Kafka、分布式消息服务RocketMQ、分布式消息服务MQTT与分布式消息服务RabbitMQ的管理员权限；具有专有网络VPC、VPCE的管理员权限。 服务内联委托CtyunAssumeRoleForEventBridge被授予权限策略CtyunAssumePolicyForEventBridge，该权限策略的内容如下： plaintext { "Version": "1.1", "Statement": [ { "Action": [ "cf:inst:InvokeFunction", "cf:inst:GetFunction", "cf:inst:ListFunctions", "KAFKA::", "MQ2::", "mqtt::", "AMQP::", "vpce::", "vpc::" ], "Resource": [ "" ], "Effect": "Allow" } ] } 以下是使用事件总线EventBridge时，需要使用服务内联委托的场景： 建立函数计算规则时，需要委托事件总线EventBridge具有获取访函数列表、函数详情以及调用函数的权限。 建立消息中间件事件源与事件目标时，需要委托事件总线EventBridge具有对分布式消息服务Kafka、分布式消息服务RocketMQ、分布式消息服务MQTT与分布式消息服务RabbitMQ的管理员权限。 建立网络端点时，需要委托事件总线EventBridge具有专有网络VPC、VPCE的管理员权限。

本文为您介绍分布式消息服务MQTT的计费模式。 目前天翼云分布式消息服务MQTT提供包周期（包年/包月）、按需2种计费模式供您灵活选择，使用越久越便宜。 下表列出两种模式的区别： 计费模式 包年/包月 按需计费 付费方式 预付费按照订单的购买周期结算。 后付费按照云服务器实际使用时长计费。 计费周期 按订单的购买周期计费。 按小时结算。 实例升级 暂不支持实例升级。 暂不支持实例升级。 更改计费模式 不支持变更为按需资源。 不支持变更为包周期资源。 变更规格 暂不支持变更实例规格。 暂不支持变更实例规格。 适用场景 适用于可预估资源使用周期的场景，价格比按需计费模式更优惠。对于长期使用者，推荐该方式。 适用于消息资源需求波动的场景，可以随时开通，随时删除。 包周期（包年/包月）：天翼云提供包月和包年的购买模式。这种购买方式相对于按需付费则能够提供更大的折扣，对于长期使用者，推荐该方式。包周期计费按照订单的购买周期来进行结算。 按需计费：这种购买方式比较灵活，可以即开即停，支持秒级计费。实例从“开通”开启计费到“删除”结束计费，按实际购买时长（精确到秒）计费。 计费模式变更 ：天翼云分布式消息服务MQTT目前不支持计费模式变更。

本节介绍了分布式消息服务RabbitMQ产品的实例规格。 （一）以下规格选型适用于14个节点 华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 注意 通用型规格已调整为白名单特性，如需了解该规格参数请联系技术支持。 云原生引擎已调整为白名单特性，如需了解该引擎请联系技术支持。 单机版实例面向用户体验和业务测试场景，无法保证性能和高可用。如果需要在生产环境使用RabbitMQ实例，建议购买集群版实例。 TPS参考值，是指以2K大小的消息为例的每秒处理消息条数，测试场景为不开启持久化的非镜像队列，实时生产实时消费，队列无积压。此数据仅供参考，在实际业务中，队列数、消息堆积、连接数、channel、消费者数、镜像队列、优先级队列、消息持久化和exchange类型等因素会对TPS造成较大影响，不一定能够达到上述性能。 1、RabbitMQ实例规格 RabbitMQ兼容开源RabbitMQ 3.8版本，提供集群和单机两种部署架构实例，支持X86和ARM计算，主机类型为Intel、海光和鲲鹏计算增强型，节点可选择3节点、5节点、7节点、9节点。 存储类型可选择高IO（SAS）、超高IO（SSD），默认选择超高IO。 2、目前所有的可选规格如下：

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

说明 分布式消息RabbitMQ旧资费产品根据实例规格分为基础版和高级版，存储空间按照不同类型收费。 旧资费产品目前在 芜湖2、南京3、上海7、重庆2、北京5 资源池开放订购。 包周期 RabbitMQ当前支持包年/包月的计费方式。且按天翼云的包年订购优惠策略享受相关的一次性包年订购优惠。 RabbitMQ队列实例价格如下： 实例类型 实例规格 节点数量 标准资费(元/月) 基础版（300G） 4C8G 3 3248 高级版（300G） 8C16G 3 6321 其中，RabbitMQ队列额外存储空间价格如下： 存储类型 标准资费（元/100GB/月） 普通IO 30 按需计费 RabbitMQ当前支持按需的计费方式，计费颗粒度为按小时。 RabbitMQ队列实例价格如下： 实例类型 实例规格 节点数量 标准资费(元/小时) 基础版（300G） 4C8G 3 6.73 高级版（300G） 8C16G 3 13.13 其中，RabbitMQ队列额外存储空间价格如下： 存储类型 标准资费（元/100GB/小时） 普通IO 0.05

本章节主要介绍设置分布式消息服务RabbitMQ实例镜像队列。 镜像队列，允许集群将队列镜像到其他节点上，当集群某一节点宕机后，队列能自动切换到镜像中的其他节点，保证服务的可用性。 如果您需要了解RabbitMQ Web UI相关功能和概念，请自行查阅RabbitMQ官网。本章节仅介绍登录RabbitMQ实例的Web页面设置镜像队列的操作步骤。 操作步骤 步骤 1 登录RabbitMQ实例的Web UI。 步骤 2 在菜单栏，选择“Admin”。 图1 选择Admin菜单 步骤3 （可选）选择右侧导航栏“Virtual Hosts”，然后输入“Name”，单击“Add virtual host”，创建Vhost。 如果您需要设置指定Vhost，请执行本步骤；如果不需要，请直接执行步骤4。 图2 创建Vhost 步骤 4 选择右侧导航栏“Policies”，为Vhost设置规则。 如果为指定的Vhost设置，请在“Virtual Host”选择步骤3创建的Vhost；如果没有，则默认为“/”。 图3 设置Vhost规则 参数解释如下： Name: policy的名称，用户自定义。 Pattern: queue的匹配模式（正则表达式）。 Definition: 镜像定义，包括三个部分hasyncmode、hamode、haparams。 hasyncmode: 表示镜像队列中消息的同步方式，有效取值范围为：automatic和manually。 automatic：表示自动向master同步数据。 manually：表示手动向master同步数据。 hamode: 指明镜像队列的模式，有效取值范围为：all、exactly和nodes。 all：表示在集群所有的节点上进行镜像。 exactly：表示在指定个数的节点上进行镜像，节点的个数由haparams指定。 nodes：表示在指定的节点上进行镜像，节点名称通过haparams指定。 haparams: hamode模式需要用到的参数。 Priority: 可选参数，policy的优先级。 步骤 5 单击“Add policy”。 规则添加成功后如图4所示。 图4 Vhost规则

本文主要介绍分布式消息服务RabbitMQ的产品优势。 天翼云分布式消息服务RabbitMQ完全兼容开源社区版本，旨在为用户提供便捷高效的消息队列。业务无需改动即可快速迁移上云，为您节省维护和使用成本。 一键式部署，免去集群搭建烦恼 专享实例只需要在实例管理界面选好规格配置，提交订单，后台将自动创建部署完成一整套RabbitMQ实例。 兼容开源，业务零改动迁移上云 兼容社区版RabbitMQ的API，具备原生RabbitMQ的所有消息处理特性。 业务系统基于开源的RabbitMQ进行开发，只需加入少量认证安全配置，即可使用天翼云分布式消息服务RabbitMQ，做到无缝迁移。 说明 RabbitMQ实例兼容开源社区RabbitMQ 3.8.35版本。 独占式体验 RabbitMQ实例采用物理隔离的方式部署，租户独占RabbitMQ实例，每个RabbitMQ之间互不影响。 高性能 单队列性能最高可达10万TPS（默认配置），增加队列可获得更高性能。 数据安全 独有的安全加固体系，提供业务操作云端审计，消息存储加密等有效安全措施。 在网络通信方面，除了提供SASL认证，还借助虚拟私有云（VPC）和安全组等加强网络访问控制。 无忧运维 天翼云提供一整套完整的监控告警等运维服务，故障自动发现和告警，避免724小时人工值守。RabbitMQ专享实例自动上报相关监控指标，如分区数、主题数、堆积消息数等，并支持配置监控数据发送规则，您可以在第一时间通过短信、邮件等获得业务消息队列的运行使用和负载状态。

RocketMQ触发器 RocketMQ触发器可以订阅分布式消息服务RocketMQ并根据消息触发关联的工作流，借此能力，使得工作流可以消费指定topic的消息，执行自定义处理逻辑。 注意事项 RocketMQ触发器订阅的RocketMQ实例必须和工作流在相同地域。 前提条件 已创建工作流。 开通分布式消息服务RocketMQ实例（RocketMQ引擎类型），详情请参考开通RocketMQ实例。 创建Topic和GroupID。 创建用户，且默认Topic权限设置为：PUBSUB，默认消费组权限为SUB。详情请参考创建用户。 触发消息格式 有两种消息格式：RawData和CloudEvent格式，可在触发器配置里选择。 CloudEvent格式： plaintext [ { "id": "21000777109E05EF04B574B8A1DF0001", "source": "ctyun.faas.trigger.rocketmq", "specversion": "1.0", "type": "rocketmq:topic:sendmessage", "datacontenttype": "application/json", "subject": ":mqfunchckzeddbxjrockettest:testforfaas", "time": "573610703T16:18:39Z", "data": { "topic": "testforfaas", "properties": { "CLUSTER": "1dafcb4049ba42df96d80b7dd2f99c5e", "CONSUMESTARTTIME": "1747987057130", "KEYS": "webtest", "MAXOFFSET": "2", "MINOFFSET": "0", "TAGS": "17479870570970", "UNIQKEY": "21000777109E05EF04B574B8A1DF0001" }, "data": "WebTestTools174798gjkS" } } ] 参数 类型 示例值 描述 id string 21000777109E05EF04B574B8A1DF0001 事件ID。标识事件的唯一值。提取自RocketMQ消息。 source string ctyun.faas.trigger.rocketmq 事件源。RocketMQ触发器固定为ctyun.faas.trigger.rocketmq。 specversion string 1.0 CloudEvents协议版本。 type string rocketmq:topic:sendmessage 事件类型。 datacontenttype string application/json 参数data的内容形式。 subject string mqfunchckzeddbxjrockettest:testforfaas 事件主体。 time string 20250522T02:04:16Z 消息被触发的时间。 data object RocketMQ触发器独有消息格式，详细参见下文RawData描述。 RawData格式 是CloudEvent格式的子集，只包含原始rabbitmq消息的信息，消息结构相当于CloudEvent的data字段，具体如下： plaintext [ { "topic": "testforfaas", "properties": { "CLUSTER": "1dafcb4049ba42df96d80b7dd2f99c5e", "CONSUMESTARTTIME": "1747987204637", "KEYS": "webtest", "MAXOFFSET": "3", "MINOFFSET": "0", "TAGS": "17479872046050", "UNIQKEY": "2100077510A605EF04B574BAE2080001" }, "data": "WebTestTools17v6fg0J" } ] 参数 类型 示例 描述 topic string testforfaas Topic名称。 properties map 消息自定义属性。 properties.CLUSTER string 1dafcb4049ba42df96d80b7dd2f99c5e RocketMQ实例ID。 properties.CONSUMESTARTTIME string 1747987204637 Unix时间戳，毫秒。 properties.KEYS string webtest 消息的key。 properties.MAXOFFSET string 3 消息队列中的最大偏移量。 properties.MINOFFSET string 0 消息队列中的最小偏移量。 properties.TAGS string 17479872046050 消息标签。 properties.UNIQKEY string 2100077510A605EF04B574BAE2080001 消息唯一键。 data string WebTestTools17v6fg0J 消息体内容。

本文主要介绍查看监控数据。 操作场景 云监控对Kafka实例的运行状态进行日常监控，可以通过控制台直观的查看Kafka实例各项监控指标。 前提条件 已创建Kafka实例，且实例中有可消费的消息。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 通过以下任意一种方法，查看监控数据。 在Kafka实例名称后，单击。跳转到云监控页面，查看实例、节点、队列和消费组的监控数据，数据更新周期为1分钟。 单击Kafka实例名称，进入实例详情页。在左侧导航栏单击“监控”，进入监控页面，查看实例、节点、队列和消费组的监控数据，数据更新周期为1分钟。

全托管 业务系统基于现有的开源 Apache Kafka 生态的代码，兼容社区版Kafka的API，具备原生Kafka的所有消息处理特性。无需任何改造，即可迁移上云，不再需要专门部署、运维，只需专注业务本身。 一键式部署 只需要在实例管理界面选好规格配置，提交订单。后台将自动创建部署完成一整套Kafka实例。 运维高效 提供多维度指标监控（队列级别）；支持消息查询、消息回溯以及消息数据过期自动删除。

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

本文介绍分布式消息服务MQTT的API请求状态码。 正常状态码 描述 800 成功 900 失败

本文主要介绍 产品定义。 Kafka是一个拥有高吞吐、可持久化、可水平扩展，支持流式数据处理等多种特性的分布式消息流处理中间件，采用分布式消息发布与订阅机制，在日志收集、流式数据传输、在线/离线系统分析、实时监控等领域有广泛的应用。 分布式消息服务Kafka是一款基于开源社区版Kafka提供的消息队列服务，向用户提供计算、存储和带宽资源独占式的Kafka专享实例。使用分布式消息服务Kafka，资源按需申请，即买即用，您将有更多精力专注于业务快速开发，不用考虑部署和运维。 关于Kafka的帮助手册阅读指引 受限于篇幅，我们提供的Kafka帮助手册重点描述产品相关的内容，以及与开源社区版Kafka的差异部分，例如Kafka的产品规格、控制台操作、客户端对接等。 如果您需要了解Kafka入门知识或消息生产、消费等方面的技术细节，请查阅Kafka官网资料。