如果消息重复消费会影响您的业务处理，要对消息做幂等处理。本文介绍消息幂等的概念、适用场景以及处理方法。 概念 在消息领域，幂等是指Consumer重复消费某条消息时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统产生任何负面影响。 例如，在支付场景下，Consumer消费扣款消息，对一笔订单执行扣款操作，扣款金额为500元。如果因网络不稳定等原因导致扣款消息重复投递，Consumer重复消费了该扣款消息，但最终的业务结果是只扣款一次，扣费500元，且用户的扣款记录中对应的订单只有一条扣款流水，不会多次扣除费用。那么这次扣款操作是符合要求的，整个消费过程实现了消息幂等。 适用场景 在互联网应用中，尤其在网络不稳定的情况下，分布式消息服务RabbitMQ的消息有可能会出现重复。如果消息重复消费会影响您的业务处理，请对消息做幂等处理。消息重复的可能原因如下： 发送时消息重复 当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时Producer意识到消息发送失败并尝试再次发送消息，Consumer后续会收到两条内容相同并且Message ID也相同的消息。 投递时消息重复 消息消费的场景下，消息已投递到Consumer并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。为了保证消息至少被消费一次，分布式消息服务RabbitMQ的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，Consumer后续会收到两条内容相同并且Message ID也相同的消息。 负载均衡时消息重复（包括但不限于网络抖动、服务端重启以及Consumer应用重启） 当分布式消息服务RabbitMQ的服务端或客户端重启、扩容或缩容时，会触发Rebalance，此时Consumer可能会收到重复消息。

本文带您了解什么是分布式消息服务RocketMQ产品。 分布式消息服务RocketMQ是一款低成本、高可靠、高性能的消息中间件产品，兼容开源RocketMQ客户端，提供高效可靠的消息传递服务，解决分布式应用系统之间的消息数据通信难题，用于系统间的解耦，用户只需专注业务，无需部署运维，适用于电商、金融、政企等多样业务场景。 产品示意图 分布式消息服务RocketMQ发布订阅基本流程如下： 1、Producer连接Nameserver，产生数据放入不同的Topic； 2、对于RocketMQ，一个Topic可以分布在各个Broker上，我们可以把一个Topic分布在一个Broker上的子集定义为一个Topic分片； 3、将Topic分片再切分为若干等分，其中的一份就是一个Queue。每个Topic分片等分的Queue的数量可以不同，由用户在创建Topic时指定； 4、Consumer连接Nameserver，根据Broker分配的Queue来消费数据。 核心组件 分布式消息服务RocketMQ主要由Producer、Broker、Consumer三部分组成，其中Producer负责生产消息，Consumer负责消费消息，Broker负责存储消息。Broker在实际部署过程中对应1台或者多台服务器，每个Broker可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的Broker。Message Queue用于存储消息的物理地址，每个Topic中的消息地址存储于多个Message Queue中。ConsumerGroup由多个Consumer实例构成。更多信息请参见产品架构。

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

可信云服务可以通过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的管理员权限。

介绍分布式消息服务RabbitMQ删除虚拟主机的操作内容。 场景描述 在RabbitMQ中，删除虚拟主机（Virtual Host）的场景描述如下： 应用迁移或清理：当需要迁移应用或进行系统清理时，可能需要删除不再使用的虚拟主机。例如，如果某个应用已经停用或迁移到其他环境，可以删除相关的虚拟主机，以释放资源和减少管理工作。 安全审计和合规性要求：根据安全审计和合规性要求，可能需要删除不再需要的虚拟主机。例如，当某个虚拟主机涉及敏感数据或权限配置存在问题时，可以选择删除该虚拟主机以避免安全风险。 故障处理和恢复：在某些情况下，当虚拟主机发生故障或出现严重问题时，可能需要删除虚拟主机并重新创建。这可以作为一种故障处理和恢复的手段，以解决虚拟主机相关的问题并恢复正常运行。 系统资源管理和优化：如果虚拟主机过多或占用了过多的系统资源，可能需要删除一些不再使用或不需要的虚拟主机。这有助于优化系统资源的利用和提高整体性能。 需要注意的是，在删除虚拟主机之前，务必确保相关的队列、交换机和绑定等对象已经被删除或迁移。否则，删除虚拟主机可能会导致数据丢失或系统异常。 操作步骤 （1）在虚拟主机管理页面，在目标虚拟主机行点击“删除”，即可删除虚拟主机。 注意事项： 删除虚拟主机会删除该虚拟主机内所有数据且不可恢复，请谨慎操作。

本节介绍了RabbitMQ接入的代码示例。 安全接入点(PLAIN、AMQPLAIN授权机制) java import com.rabbitmq.client.; import java.io.IOException; public class RabbitmqAmqpDemo { public static void main(String[] args) throws Exception { String host "192.168.0.0"; //安全接入点ip Integer port 5672; //安全接入点port String username "xxx"; //集群管理用户列表的用户名 String password "xxx"; //集群管理用户列表的密码 String vhost "/"; String exchangeName "extest"; String queueName "qutest"; ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost(vhost); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, "test"); String message "Hello Aop"; for (int i 0; i < 10; i++) { channel.basicPublish(exchangeName, "test", null, message.getBytes()); System.out.println("消息发送成功"); } Channel consumeChannel connection.createChannel(); Consumer consumer new DefaultConsumer(consumeChannel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String messageGet new String(body, "UTF8"); if (messageGet.equals(message)) { System.out.println("消息消费成功"); } } }; consumeChannel.setDefaultConsumer(consumer); consumeChannel.basicConsume(queueName, false, consumer); Thread.sleep(10000); } } SSL接入点(EXTERNAL授权机制) java import com.rabbitmq.client.; 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.security.KeyStore; public class RabbitmqExternalDemo { public static void main(String[] args) throws Exception { 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 "/"; String exchangeName "extest"; String queueName "qutest"; 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); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, "test"); String message "Hello Aop"; for (int i 0; i < 10; i++) { channel.basicPublish(exchangeName, "test", null, message.getBytes()); System.out.println("消息发送成功"); } Channel consumeChannel connection.createChannel(); Consumer consumer new DefaultConsumer(consumeChannel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String messageGet new String(body, "UTF8"); if (messageGet.equals(message)) { System.out.println("消息消费成功"); } } }; consumeChannel.setDefaultConsumer(consumer); consumeChannel.basicConsume(queueName, false, consumer); Thread.sleep(10000); } }

本章介绍函数工作流如何使用RabbitMQ触发器。 前提条件 已经创建函数。 创建RabbitMQ触发器，必须开启函数工作流VPC访问。 已经创建RabbitMQ实例。 确认实例安全组规则是否配置正确。 在RabbitMQ实例详情页面的“基本信息 > 网络”，单击安全组名称，跳转到安全组页面。 选择“入方向规则”，查看安全组入方向规则。 实例未开启SSL开关 如果是VPC内访问，实例安全组入方向规则，需要允许端口5672的访问。 如果是公网访问，需要允许端口15672的访问。 实例已开启SSL开关 如果是VPC内访问，实例安全组入方向规则，需要允许端口5671的访问。 如果是公网访问，需要运行端口15671的访问。 创建RabbitMQ触发器 1. 登录函数工作流控制台，在左侧的导航栏选择“函数 > 函数列表”。 2. 选择待配置的函数，单击进入函数详情页。 3. 选择“设置 > 触发器”，单击“创建触发器”，弹出“创建触发器”对话框。 4. 设置以下信息。 触发器类型：选择“分布式消息服务RabbitMQ版 (RABBITMQ)”。 实例：选择已创建RabbitMQ实例。 交换机名称：填写用户需要使用的交换机名称。 密码：填写创建RabbitMQ实例的密码。 虚拟机名称：填写用户自定义的vhost。 批处理大小：每次从Topic消费的消息数量。 5. 单击“确定”，完成RabbitMQ触发器的创建。 说明 开启函数流VPC访问后，需要在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 消息体内容。

本文主要介绍节点重启后消费者如何重连分布式消息服务RabbitMQ。 本章节以Java中使用的RabbitMQ客户端amqpclient为例介绍节点重启后消费者如何重连。 amqpclient自带重连机制，但是自带的重连机制只会重试一次，一次连不上后就不会再执行了，这时如果消费者没有做额外的重试机制，那么这个消费者就彻底丧失的消费能力。 amqpclient在节点断连后，根据与通道建立的节点不同，产生不同的错误。 如果通道连接的是队列所在的节点，消费者就会收到一个shutdown信号，这时amqpclient的重连机制就会生效，尝试重新连接服务端。如果连上了，这个通道就会继续连接消费。如果连不上，就会执行channel.close方法，关闭这个通道。 如果通道连接的不是队列所在的节点，消费者不会触发关闭动作，而是由服务端发送的一个取消动作，这个动作对amqpclient来说并不是异常行为，所以日志上不会有明显的报错，但是连接最终还是会关闭。 amqpclient出现上面两种错误时，会分别回调handleShutdownSignal以及handleCancel方法，您可以通过重写这两种方法，在回调时执行重写的重连逻辑，就能在通道关闭后重新创建消费者的新通道继续消费。 以下提供一个简单的代码示例，此示例能够解决上面的两种错误，实现消费者的持续消费。 package rabbitmq; import com.rabbitmq.client.; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; public class RabbitConsumer { public static void main(String... args) throws IOException, TimeoutException { ConnectionFactory factory new ConnectionFactory(); factory.setHost("100.00.000.000"); factory.setPort(5672); factory.setUsername("name"); factory.setPassword("password"); Connection connection factory.newConnection(); createNewConnection(connection); } public static void createNewConnection(Connection connection) { try { Thread.sleep(1000); Channel channel connection.createChannel(); channel.basicQos(64); channel.basicConsume("queue01", false, new CustomConsumer(channel, connection)); } catch (Exception e) { // e.printStackTrace(); createNewConnection(connection); } } static class CustomConsumer implements Consumer { private final Channel channel; private final Connection connection; public CustomConsumer(Channel channel, Connection connection) { channel channel; connection connection; } @Override public void handleConsumeOk(String consumerTag) { } @Override public void handleCancelOk(String consumerTag) { } @Override public void handleCancel(String consumerTag) throws IOException { System.out.println("handleCancel"); System.out.println(consumerTag); createNewConnection(connection); } @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { System.out.println("handleShutdownSignal"); System.out.println(consumerTag); System.out.println(sig.getReason()); createNewConnection(connection); } @Override public void handleRecoverOk(String consumerTag) { } @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(" [x] Received '" + message + "'"); channel.basicAck(envelope.getDeliveryTag(), false); } } }

消费消息 plaintext using System; using System.Collections.Immutable; using System.Drawing; using Rabbit.Common.Data.Trades; using Rabbit.Common.Display; using RabbitMQ.Client; using RabbitMQ.Client.Events; namespace Rabbit.Example4.Consumer { internal sealed class Program { private static void Main() { var connectionFactory new ConnectionFactory { HostName "YOUR HOST IP", UserName "YOUR USER", Password "YOUR PASSWORD", Port 5672 }; using var connection connectionFactory.CreateConnection(); using var channel connection.CreateModel(); const string ExchangeName "dotnetexchange"; const string QueueName "dotnetqueue"; channel.ExchangeDeclare( exchange: ExchangeName, type: ExchangeType.Direct, durable: false, autoDelete: false, arguments: ImmutableDictionary .Empty); var queue channel.QueueDeclare( queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: ImmutableDictionary .Empty); channel.QueueBind( queue: queue.QueueName, exchange: ExchangeName, routingKey: QueueName); var consumer new EventingBasicConsumer(channel); consumer.Received + (sender, eventArgs) > { var messageBody eventArgs.Body.ToArray(); var trade Trade.FromBytes(messageBody); DisplayInfo .For(trade) .SetExchange(eventArgs.Exchange) .SetQueue(queue.QueueName) .SetRoutingKey(eventArgs.RoutingKey) .SetVirtualHost(connectionFactory.VirtualHost) .Display(Color.Yellow); channel.BasicAck(eventArgs.DeliveryTag, multiple: false); }; channel.BasicConsume( queue: queue.QueueName, autoAck: false, consumer: consumer); Console.ReadLine(); } } } ssl生产消息 plaintext using System.Collections.Generic; using System.Collections.Immutable; using System.Drawing; using System.Threading.Tasks; using Rabbit.Common.Data.Trades; using Rabbit.Common.Display; using RabbitMQ.Client; namespace Rabbit.dotnet.Producer { internal sealed class Program { private static async Task Main() { var connectionFactory new ConnectionFactory { HostName "YOUR HOST IP", UserName "YOUR USERNAME", Password "YOUR PASSWORD", Port 5671 }; connectionFactory.AuthMechanisms new List () { new ExternalMechanismFactory() }; connectionFactory.Ssl.CertPath @"D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqkey.p12"; // Ssl.CertPassphrase为固定参数，无需修改 connectionFactory.Ssl.CertPassphrase "YOUR PASSPHRASE"; connectionFactory.Ssl.Enabled true; connectionFactory.Ssl.CertificateValidationCallback (, , , ) > { return true; }; using var connection connectionFactory.CreateConnection(); using var channel connection.CreateModel(); const string ExchangeName "dotnetexchange"; const string QueueName "dotnetqueue"; channel.ExchangeDeclare( exchange: ExchangeName, type: ExchangeType.Direct, durable: false, autoDelete: false, arguments: ImmutableDictionary .Empty); var queue channel.QueueDeclare( queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: ImmutableDictionary .Empty); channel.QueueBind( queue: queue.QueueName, exchange: ExchangeName, routingKey: QueueName, arguments: ImmutableDictionary .Empty); while (true) { var trade TradeData.GetFakeTrade(); string routingKey QueueName; channel.BasicPublish( exchange: ExchangeName, routingKey: routingKey, body: trade.ToBytes() ); DisplayInfo .For(trade) .SetExchange(ExchangeName) .SetRoutingKey(routingKey) .SetVirtualHost(connectionFactory.VirtualHost) .Display(Color.Cyan); await Task.Delay(millisecondsDelay: 1000); } } } }

本章节主要介绍如何删除Vhost。 操作场景 本章节主要介绍删除Vhost的操作，有以下几种方式，您可以根据实际情况选择任意一种方式： 方式一：在控制台删除 方式二：使用RabbitMQ WebUI删除 方式三：调用API删除 方式一：在控制台删除 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 此处请选择RabbitMQ实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”>“分布式消息服务”>“RabbitMQ专享版”，进入分布式消息服务RabbitMQ专享版页面。 步骤 4 单击实例名称，进入实例详情页面。 步骤 5 在左侧导航栏选择“Vhost列表”，进入Vhost列表页面。 步骤 6 通过以下任意一种方法，删除Vhost。 勾选Vhost名称左侧的方框，可选一个或多个，单击信息栏左上侧的“删除Vhost”。 在待删除的Vhost所在行，单击“删除”。 步骤 7 在弹出的确认删除对话框中，单击“是”。 方式二：使用RabbitMQ WebUI删除 步骤 1 登录RabbitMQ WebUI。 步骤 2 在顶部导航栏选择“Admin”，进入Admin页面。 步骤 3 在右侧导航栏选择“Virtual Hosts”，进入Virtual Hosts页面。 图1 Virtual Hosts页面 步骤 4 单击待删除的Vhost名称，进入Vhost详情页。 图2 待删除的Vhost 步骤 5 在“Delete this vhost”区域，单击“Delete this virtual host”，弹出确认删除对话框。 图3 删除Vhost 步骤 6 单击“确定”，完成Vhost的删除。

本章节主要介绍如何开启心跳。 客户端连接RabbitMQ集群实例时，如果存在消息收发时间间隔大于90秒的场景，请在客户端开启心跳并设置小于90秒的心跳超时时间，防止断连。 什么是心跳 RabbitMQ实例提供了心跳功能，以确保应用程序层及时发现中断的连接和完全无响应的对端。心跳还可以防止某些网络设备在一段时间内由于没有活动而中断TCP连接。开启心跳的方法为在连接上指定心跳超时时间。 心跳超时时间定义了对等TCP连接在多长时间后被服务端和客户端视为关闭。服务端和客户端会对配置的心跳超时时间进行协商，客户端必须配置该值来发送心跳。RabbitMQ官方团队维护的3个客户端（Java、.NET、Erlang语言）的心跳超时时间协商逻辑如下： 服务端和客户端设置的心跳超时时间都不为0时，两者间较小的值生效。 服务端和客户端任意一端设置的心跳超时时间为0，另一端不为0时，非0的值生效。 服务端和客户端的心跳超时时间都设置为0时，表示禁用心跳。 配置心跳超时时间后，RabbitMQ服务端和客户端都会向对方发送AMQP心跳帧作为心跳，发送的时间间隔为心跳超时时间的一半。客户端在两次错过心跳后，会被认为是不可达的，TCP连接将被关闭。当客户端检测到服务端由于心跳而无法访问时，需要重新连接。 说明 一些客户端（如C语言客户端）没有发送心跳的逻辑，即使配置了心跳超时时间，开启了心跳，仍然无法发送心跳。此时需要额外启动一个线程，编写发送心跳的逻辑。

本文介绍如何在事件总线EventBridge管理控制台添加分布式消息服务MQTT类型的自定义事件源。 前提条件 事件总线EventBridge 开通事件总线EventBridge并授权。 创建自定义总线。 分布式消息服务MQTT 开通分布式消息服务MQTT并创建实例。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件总线。 3. 在事件总线页面，单击目标总线名称。 4. 在左侧导航栏，单击事件源。 5. 在事件源 页面，单击添加事件源。 6. 在添加自定义事件源 面板，输入名称 和描述 ，事件提供方 选择分布式消息服务MQTT，选择或填入主题等配置然后单击确认，如图1所示。 图1 创建事件源时事件提供方选择分布式服务MQTT 参数说明 参数 说明 示例 名称 事件源名。 test MQTT实例 选择MQTT实例。 instancexxx MQTT主题 选择或输入MQTT的主题。 topicxxx 事件示例 plaintext { "id": "b5771f766cdf48edb1bad15418c", "source": "sourcetest", "specversion": "1.0", "subject": "ctyun:mqtt:bb9fdb42056f1xxxxxx2ac110002:dab4124xxxxxx7eb2774f45c6a6db69:topic/

本文介绍分布式消息服务RocketMQ可广泛应用于分布式系统异步通信、数据同步和交换、削峰填谷等场景。 行业应用 RocketMQ在多个行业中都有广泛的应用。以下是一些典型的行业应用场景： 1. 电子商务：在电子商务行业中，RocketMQ可以用于订单处理、库存管理、支付通知等异步通信和事件驱动的场景。它可以实现订单的可靠传递和处理，同时支持高并发和高可扩展性的需求。 2. 金融服务：在金融服务行业中，RocketMQ可以用于实时交易通知、资金结算、风险控制等关键业务场景。它可以确保交易消息的可靠传递和顺序处理，同时支持高吞吐量和低延迟的要求。 3. 物流与供应链：在物流与供应链行业中，RocketMQ可以用于实时物流跟踪、订单状态更新、库存管理等消息通知和事件驱动的场景。它可以确保供应链各个环节的信息同步和协调，提高物流效率和准确性。 4. 社交媒体：在社交媒体行业中，RocketMQ可以用于实时消息推送、用户关系管理、活动通知等场景。它可以支持大规模用户同时在线的需求，确保消息的低延迟和高可靠性。 5. 物联网：在物联网行业中，RocketMQ可以用于设备状态监控、告警通知、数据采集等场景。它可以处理大量的设备消息，并支持设备之间的实时通信和协作。 这些是一些典型的RocketMQ行业应用场景，由于RocketMQ的高性能和可靠性，它在更多行业中也有广泛的应用。 分布式消息服务RocketMQ主要适用于以下几种业务场景：

本文为您介绍分布式消息服务MQTT的产品优势。 分布式消息服务MQTT（Message Queuing Telemetry Transport）是一种轻量级的、发布/订阅模式的消息传递协议，通常用于分布式消息服务和物联网（IoT）应用。以下是MQTT的一些优势： 轻量级协议： MQTT是一种非常轻量级的协议，适用于各种网络环境，包括低带宽、高延迟或不稳定的网络。这使得它非常适合在IoT设备之间传递消息，因为这些设备通常有限的资源和能源。 发布/订阅模式： MQTT采用发布/订阅模式，允许客户端订阅感兴趣的主题，以接收与这些主题相关的消息。这种模式具有灵活性，能够支持多对多的通信，而不需要直接点对点的连接。 异步通信： MQTT允许异步通信，客户端可以发布消息而不必等待接收方的响应。这有助于提高系统的响应速度和吞吐量。 保留消息： MQTT支持保留消息，这意味着最新的消息可以保留在主题中，以便新订阅者可以立即获取到最新数据，而不必等待下一个消息发布。 服务质量（QoS）： MQTT允许设置不同级别的服务质量，从不保证消息送达（QoS 0）到确保消息送达且不重复传递（QoS 2）。这使得可以根据应用的需求选择适当的传递质量。 可伸缩性： MQTT协议可以在大规模分布式系统中轻松扩展，支持成千上万的客户端同时连接到Broker，使其适用于大型IoT解决方案。 消息过滤： MQTT支持使用通配符来订阅主题，这允许客户端根据特定模式匹配多个主题，以接收相关消息。 支持遗愿消息： MQTT支持遗愿消息（Will Messages），允许客户端指定在异常断开连接时发送一条预定义消息。这对于检测设备在线状态非常有用。 跨平台和多语言支持： MQTT具有广泛的跨平台和多语言支持，因此可以在各种设备和编程语言上使用，使得它非常灵活。 安全性： MQTT可以与安全机制（如TLS/SSL）结合使用，确保消息在传输过程中的安全性和隐私。 总的来说，MQTT是一种高效、灵活和可靠的协议，特别适用于分布式消息传递和IoT应用，因此在物联网、远程监控、实时通信等领域得到广泛应用。它的轻量级特性和异步通信使得它成为连接数众多的设备的理想选择。

本文带您了解分布式消息服务RocketMQ的功能特性。 消息生产 消息压缩：将较大的消息进行压缩后发送到服务端，有效利用带宽。 延迟消息：设计消费时延，消息发送到服务端后，过了预设时间才可以被消费。 事务消息：根据预设的事务，事务消息可保证分布式系统之间的数据最终一致。 消息消费 有序消费：支持普通有序消息和严格有序消息两种方式。 集群消费：一个主题可被一个或多个消费者组消费，消费者组中消费者实例可平均分摊消费信息。 消费位置设置：支持设置消费组首次启动消费的位置，包括队列头、队列尾及由客户端指定。 消息回溯：支持按时间回溯消费进度，将订阅组在某主题上的消费进度重置到过去或者未来。 完善的运维能力 应用用户管理：集群租户隔离，应用接入集群权限管理。 主题管理：支持对实例下的主题进行管理，执行创建删除等操作。 订阅组管理：支持对实例下的订阅组进行管理。 生产者和消费者管理：用户可查看当前实例下的生产者和消费者信息，并实时更新。 消息查询：按消息ID、消息逻辑偏移量、消息key。 完善的运维功能，节点状态检测、启停；实例状态检测、启停；SLA监控等。 顺序消息 顺序消息是指消费消息的顺序要同发送消息的顺序一致，在RocketMQ中，主要有两种有序消息：全局有序消息和局部有序消息（又称普通有序消息、分区有序消息）。 普通有序消息：在正常情况下可以保证完全的顺序消息，但是一旦发生通信异常造成Broker重启，队列总数发生变化，哈希取模后定位的队列会变化，因此会产生短暂的消息顺序不一致。如果业务能容忍在集群异常情况（如某个Broker宕机或者重启）下消息短暂的乱序，使用普通顺序方式比较合适。 严格有序消息：无论正常异常情况都能保证顺序，但是牺牲了分布式Failover特性，即Broker集群中只要有一台机器不可用，则整个集群都不可用（或者影响hash 值对应队列的使用），服务可用性大大降低。如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。

Messaging（消息传递）是指在分布式系统中，通过消息传递实现不同组件之间的通信。 System （消息系统）：是指消息传递所使用的消息系统，如Apache Kafka、RabbitMQ、ActiveMQ等。它是一个软件系统，用于实现异步消息传递。 Operation （操作）：是指对消息执行的操作，包括发送、接收、确认等。它是一个字符串，用于表示当前执行的操作。 DestinationKind （目标类型）：是指消息的目标类型，包括队列（Queue）和主题（Topic）两种。队列用于点对点的消息传递，主题用于发布订阅模式的消息传递。它是一个字符串，用于表示当前消息的目的地类型。 目标服务（Target Service）是指客户端需要访问的远程服务。 名称 （Name）：是指目标服务的名称，用于唯一标识服务。它是一个字符串，通常由服务提供方指定，并在服务注册中心中注册。 类型 （Type）：是指目标服务的类型，用于表示服务的功能类别。例如，Web服务、消息队列服务、数据库服务等。它是一个字符串，通常由服务提供方指定。 实例 （Instance）：是指目标服务的实例，用于标识不同的服务实例。在分布式系统中，通常会有多个服务实例提供相同的服务。它是一个字符串，通常由服务注册中心分配。

RabbitMQProducer 和KafkaProducer类似，RabbitMQProducer主要负责发送消息到RabbitMQ，监控在这一过程中的各项指标，包括： 总发送次数趋势图、总发送字节数趋势图。 不同exchange的调用次数、平均响应时间、错误数、最慢调用信息和总发送字节数。 这些指标可以帮助识别生产者在消息发送过程中可能遇到的问题，如网络延迟、服务器负载过高或配置错误等。 RabbitMQConsumer 和KafkaConsumer类似，RabbitMQConsumer主要负责处理接收到的消息，监控在这一过程中的各项指标，包括： 总消费次数趋势图、总消费字节数趋势图。 不同consumerTag/consumerClass的调用次数、平均响应时间、错误数、最慢调用信息和总消费字节数。 这些指标有助于了解消费者的健康状况，例如是否存在未确认的消息积压，消费者是否能够及时处理消息，以及队列是否因为消费者处理不及时而积压。 统一交互操作说明： 将光标移到统计图上，可以查看光标所至时间点的数据详情。 单击图标，可以将当前图表放大显示。

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

介绍分布式消息服务RabbitMQ单一活跃消费者功能。 使用场景 在默认direct模式下, 多个消费者订阅一个queue,消息会轮流发送至每个消费者。单一消费者模式下,消息只由一个消费者消费，除非该改消费者出现异常。由此可以保证消息消费的有序性。 使用示例 以下为java客户端示例，通过在声明队列时配置xsingleactiveconsumer参数实现单一活跃消费者。 Map arguments newHashMap(); arguments.put("xsingleactiveconsumer", true); channel.queueDeclare("testqueue", false, false, false, arguments);

产品优势 分布式消息服务MQTT具有协议丰富、安全可靠、低成本高性能、消息互通等优势，适用于大规模分布式系统中的消息传递和处理场景。 协议丰富： 兼容支持原生多种标准协议，如MQTT、MQTTSN、CoAP、LwM2M；兼容任何支持MQTT协议的 SDK覆盖绝大多数移动端开发平台及开发语言。 安全可靠： 数据安全，支持SSL加密通信，提供身份认证，数据传输更安全可靠。 低成本高性能： 稳定承载大规模终端设备连接，资源占用少；消息路由快速低延时，单集群支持百万规模的路由。 更多信息请参见产品优势。 应用场景 分布式消息服务MQTT可以用于许多应用场景，其灵活性和轻量级特性使其适用于各种需要实时消息传输和发布/订阅模型的应用，如物联网通信、远程监控及控制、传感器数据采集等。 物联网（IoT）通信 MQTT广泛用于连接和管理大规模的物联网设备。它允许传感器、嵌入式设备和其他物联网终端之间进行实时数据通信，以监测环境、收集数据和实现智能控制。 远程监控和控制 MQTT可用于远程监控和控制系统，如监控能源设备、工业自动化、智能家居系统等。它使操作员能够实时监测设备状态，同时可以通过发布命令来实现设备控制。 传感器数据采集 MQTT用于将传感器数据从分布式传感器网络传送到数据中心或云平台，以进行分析和处理。这对于监测环境、天气预报、农业数据收集等非常有用。 更多信息请参见应用场景。

本章节介绍了什么是分布式消息服务RocketMQ。 分布式消息服务RocketMQ是一个低延迟、弹性高可靠、高吞吐、动态扩展、便捷多样的消息中间件服务。 分布式消息服务RocketMQ具有如下特点： 兼容开源RocketMQ客户端。 提供顺序、延迟、定时、重投、死信、事务消息等功能，更好的适配电商、金融等多样的业务场景。 提供消息追踪、消息溯源、链路诊断、死信导出、监控告警等能力，帮助您全方面的了解服务状况，保证业务正常运行。 产品架构 图1 产品架构示意图 示意图说明： Broker：负责接收和存储Producer发送的消息，或者转发消息到Consumer。一个Broker由一个主节点和两个备节点组成。 NameServer：负责接收和存储Broker中的元数据。 Producer：从NameServer获取元数据，然后将消息发送到Broker。 Consumer：从NameServer获取元数据，然后从Broker拉取消息。 支持的消息类型 分布式消息服务RocketMQ支持4种消息类型。 普通消息：没有特殊功能的消息，区别于延迟消息、顺序消息和事务消息。 延迟消息/定时消息：生产者生产消息到分布式消息服务RocketMQ后，消息不会立即被消费，而是延迟到特定时间后才会发送给消费者进行消费。 顺序消息：消费者按照消息发送的顺序来消费消息。 事务消息：提供类似X/Open XA的分布事务功能，通过事务消息能达到分布式事务的最终一致。

本文为您介绍分布式消息服务MQTT的产品功能。 消息模型 支持发布/订阅消息模型，提供一对多的消息分发方式。这意味着一个发布者（发布消息的客户端）可以将消息发送到一个或多个订阅者（订阅相同主题的客户端），而不需要建立直接的点对点连接。 以下是MQTT的发布/订阅消息模型的关键特点： 主题（Topic）： 发布者在发送消息时会将消息与一个主题相关联。主题是消息的分类或标识，它用于告诉Broker消息应该发送到哪些订阅者。订阅者可以选择订阅感兴趣的主题，以接收与这些主题相关的消息。 订阅（Subscribe）： 订阅者告诉MQTT Broker它对哪些主题感兴趣。一旦订阅者订阅了特定主题，Broker会将匹配的消息发送给订阅者。订阅者可以同时订阅多个主题。 发布（Publish）： 发布者将消息发送到特定主题，然后Broker负责将消息传递给所有订阅了该主题的订阅者。这样，消息可以广播给所有订阅者，实现了一对多的消息分发。 多对多通信： MQTT支持多对多通信，这意味着多个发布者可以同时发布消息到相同或不同的主题，多个订阅者可以同时订阅一个或多个主题。这种灵活性使MQTT适用于各种应用场景，包括物联网、实时通信和分布式系统。 通配符支持： MQTT还支持通配符订阅，订阅者可以使用通配符来匹配多个主题，从而接收与通配符模式匹配的消息。这提供了更高级别的消息过滤和选择性订阅。 总之，MQTT的发布/订阅消息模型使其成为一种非常灵活的消息传递协议，特别适用于需要一对多消息分发的应用，如物联网、实时监控、传感器数据采集等。它的轻量级特性和广泛的跨平台支持也使其成为了许多分布式系统和应用程序的首选通信协议之一。

本节介绍了在RabbitMQ实例中如何创建和删除交换器。 背景信息 生产者将消息发送到交换器，由交换器将消息路由到一个或多个队列中（或者丢弃）。交换器根据Routing Key和Binding Key将消息路由到Queue。不同类型的交换器的路由规则不同。 操作步骤 新建交换器 1.登录管理控制台。 2.进入RabbitMQ管理控制台。 3.在实例列表页在操作列，目标实例行点击“管理”。 4.点击“交换器管理”后，点击“新建”按钮。 5.点击“新建”后出现以下窗口，选择虚拟主机，添加交换器名字，选择交换器类型和其他参数，然后点击“确定”即可新建交换器。 各参数说明如下 参数 描述 虚拟主机 选择创建交换器所属的虚拟主机 名称 交换器名称。以amq.开头的为保留字段，因此不能使用。例如：。 类型 Exchange类型。 是否持久化 交换器是否持久化到磁盘 是否自动删除 如果是，交换器将在至少一个队列或交换器绑定到该交换器，然后所有队列或交换器都已解除绑定时删除。 是否内置 如果是，客户端不能直接发布到这个交换器。它只能与其他交换器绑定使用。 其他参数 Alternate exchange：备份交换器是为了实现没有路由到队列的消息，声明交换机的时候添加属性alternateexchange，声明一个备用交换机，一般声明为fanout类型，这样交换机收到路由不到队列的消息就会发送到备用交换机绑定的队列中。 其中交换机类型如下表所示 交换机类型 说明 Direct 完全根据key进行投递的叫做Direct交换机。如果Routing key匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时，它会自动的以queue的名字作为routing key来绑定那个exchange。例如，绑定时设置了Routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。 Fanout 不需要key的叫做Fanout交换机。它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。 Topic 对key进行模式匹配后进行投递的叫做Topic交换机。比如符号”

ssl生产消息 plaintext package main import ( "crypto/tls" "crypto/x509" "flag" "fmt" amqp "github.com/rabbitmq/amqp091go" "io/ioutil" "log" ) var ( uri flag.String("uri", "amqps://USERNAME:PASSWORD@10.10.33.196:5671", "AMQP URI") exchangeName flag.String("exchange", "goexchange", "Durable AMQP exchange name") exchangeType flag.String("exchangetype", "direct", "Exchange type directfanouttopicxcustom") routingKey flag.String("key", "testkey", "AMQP routing key") body flag.String("body", "foobar", "Body of message") reliable flag.Bool("reliable", true, "Wait for the publisher confirmation before exiting") ) func init() { flag.Parse() } func main() { if err : publish(uri, exchangeName, exchangeType, routingKey, body, reliable); err ! nil { log.Fatalf("%s", err) } log.Printf("published %dB OK", len(body)) } func publish(amqpsURI, exchange, exchangeType, routingKey, body string, reliable bool) error { caCert, err : ioutil.ReadFile("D:tmphzmqtest0520rabbitmqsslclientcacertificate.pem") if err ! nil { return err } cert, err : tls.LoadX509KeyPair("D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqcertificate.pem", "D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqkey.pem") if err ! nil { return err } rootCAs : x509.NewCertPool() rootCAs.AppendCertsFromPEM(caCert) tlsConf : &tls.Config{ RootCAs: rootCAs, Certificates: []tls.Certificate{cert}, //ServerName: "localhost", // Optional InsecureSkipVerify: true, } connection, err : amqp.DialTLS(amqpsURI, tlsConf) if err ! nil { return fmt.Errorf("Dial: %s", err) } defer connection.Close() log.Printf("got Connection, getting Channel") channel, err : connection.Channel() if err ! nil { return fmt.Errorf("Channel: %s", err) } log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange) if err : channel.ExchangeDeclare( exchange, // name exchangeType, // type true, // durable false, // autodeleted false, // internal false, // noWait nil, // arguments ); err ! nil { return fmt.Errorf("Exchange Declare: %s", err) } // Reliable publisher confirms require confirm.select support from the // connection. if reliable { log.Printf("enabling publishing confirms.") if err : channel.Confirm(false); err ! nil { return fmt.Errorf("Channel could not be put into confirm mode: %s", err) } confirms : channel.NotifyPublish(make(chan amqp.Confirmation, 1)) defer confirmOne(confirms) } log.Printf("declared Exchange, publishing %dB body (%q)", len(body), body) if err channel.Publish( exchange, // publish to an exchange routingKey, // routing to 0 or more queues false, // mandatory false, // immediate amqp.Publishing{ Headers: amqp.Table{}, ContentType: "text/plain", ContentEncoding: "", Body: []byte(body), DeliveryMode: amqp.Transient, // 1nonpersistent, 2persistent Priority: 0, // 09 }, ); err ! nil { return fmt.Errorf("Exchange Publish: %s", err) } return nil } func confirmOne(confirms

来自：

帮助文档

分布式消息服务RabbitMQ

开发指南

Go

本章节主要介绍插件类问题。 支持的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具有较强的消息堆积能力，主要以下几个方面实现和保障。 消息存储机制：RocketMQ使用基于磁盘的消息存储机制，可以持久化大量的消息到磁盘中，避免消息丢失。 消息发送速度：RocketMQ具有高吞吐量和低延迟的特点，可以快速处理大量的消息发送请求。 消息消费速度：RocketMQ支持水平扩展和负载均衡机制，可以实现高并发的消息消费能力。 存储容量：由于RocketMQ使用磁盘存储消息，因此它的存储容量可以非常大。可以通过添加更多的磁盘来扩展存储容量。 分布式消息服务RocketMQ具有很高的并发能力，RocketMQ的并发能力主要体现于以下几个方面： 集群模式：RocketMQ支持将消息队列集群化部署，通过增加Broker节点和消费者实例，可以提高整体系统的并发处理能力。每个Broker节点和消费者实例都可以处理独立的消息流，从而实现并发处理。 分区模式：RocketMQ支持将消息划分为多个分区，每个分区可以由不同的Broker节点处理。通过使用分区模式，可以将消息负载均衡到不同的节点上，从而提高并发处理能力。 异步发送模式：RocketMQ提供了异步发送消息的方式，发送者将消息发送给Broker后立即返回，无需等待确认。这种方式可以提高发送消息的并发能力，同时也会带来一定的消息丢失的风险。 消费者线程数：RocketMQ的消费者可以通过增加线程数来提高并发处理能力。每个线程可以独立地从Broker节点拉取消息并进行处理。需要注意的是，过多的线程数可能会对系统的性能产生负面影响，需要根据实际情况进行调优。

本章节介绍了分布式消息服务RocketMQ与开源RocketMQ的差异。 分布式消息服务RocketMQ在兼容开源RocketMQ基础上，对版本特性做了一定程度的定制和增强。 表1 分布式消息服务RocketMQ与开源RocketMQ的差异说明 功能项 分布式消息服务RocketMQ 开源RocketMQ 延迟消息/定时消息 延迟消息：支持18个固定延迟时长，最长延迟2小时。 定时消息：支持任意延迟时长，最长延迟1年。 仅支持18个固定延迟时长，最长延迟2小时。 顺序消息 支持 支持 消息重试 支持 支持 死信消息 支持 支持 集群消费 支持 支持 广播消费 支持 支持 死信队列 支持 支持 消费重置 支持 支持 消息查询 支持 支持 加密传输 支持 支持 消息轨迹 支持 支持 事务消息 支持，事务消息大量堆积时，性能提升10倍 支持，事务消息大量堆积时，性能较差 死信导出 支持 不支持 数据转储 待规划 不支持 实例诊断 一键诊断消费问题 不支持 实例监控 支持以图表形式查看历史值，18+监控项 仅支持查看当前监控值，不支持CPU、内存等监控指标 ACL访问控制 灵活配置，一键生效 配置复杂 运维扩容 极速扩容，一键生效 手动扩容，操作复杂

介绍分布式消息服务RocketMQ与开源自建的对比情况。 分布式消息服务RocketMQ基于开源产品RocketMQ进行问题修复与优化，并自主研发，实现低成本、高可靠、高性能和具备监控运维能力的中间件产品。 使用开源RocketMQ遇到的问题 功能不完善：功能比较单一，针对不同应用场景无法有效支持，如消息查询，数据自动删除策略等。 可维护性差：缺乏配套监控运维能力，难以迅速发现解决如消息堆积、队列堵塞等问题。 可靠性较低：消息服务不提供主备切换能力，存在单点故障，无法保证服务高可用。 分布式消息服务RocketMQ改进点 高可用、高可靠改进： 实现自动主备切换、自动拉起功能，保证服务高可用。 实现消息删除策略，按不同的场景优先保证服务可用性或者数据安全性。 可维护性改进： 实现按生产者、消费者、数据节点、队列4种维度的运行状态监控，方便快速发现问题。 实现可视化的监控、配置、管理界面实现自动化测试，以快速迭代。 新增功能： 新增消息查询，做到可查可追踪。 重新封装SDK，简化应用使用，并提供按hash算法实现消息局部有序生产消费。 开源自建对比项 对比项 开源自建 分布式消息服务RocketMQ 自动化部署 不支持自动化部署，需要专业人员自行部署、运维 全托管PaaS，免机器资源一键自动化部署。开箱即用，按需使用，支持弹性扩容 运维监控 缺乏配套的监控运维能力 提供多维度的数据可视化监控，快速定位、处理问题 增强能力 功能单一，无法支持多种场景需求 功能丰富，支持消息重试、消息查询、消息轨迹查询、数据自动清理等定制化功能 延迟消息/定时消息 仅支持18个固定延迟时长，最长延迟2小时 支持秒级的任意延迟时长，最长延迟40天 ACL访问控制 访问控制配置复杂 灵活配置，一键生效

发送消息 参考如下示例代码。 from rocketmq.client import Producer, Message endpoint "${ENDPOINT}" 填写分布式消息服务RocketMQ控制台Namesrv接入点 accesskey "${ACCESSKEY}" 填写AccessKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID accesssecret "${SECRETKEY}" 填写SecretKey 在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户密钥 topic "${TOPIC}" 填写Topic，在管理控制台创建 producergroup "${GROUP}" 生产者组group 创建并启动生产者实例 producer Producer(producergroup) producer.setnameserveraddress(endpoint) producer.setsessioncredentials(accesskey, accesssecret, "") producer.start() msg Message(topic) msg.setbody("Hello RocketMQ") msg.setkeys("") 消息key msg.settags("") 消息tag shardingkey "key" 指定消息投递的sharding key 根据Sharding Key，发送顺序消息， ret producer.sendorderlywithshardingkey(msg, shardingkey) print(ret.status, ret.msgid, ret.offset) 关闭生产者实例，释放资源 producer.shutdown() 订阅消息 参考如下示例代码。 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, orderlyTrue)

本章节介绍了分布式消息服务RocketMQ的产品优势、以及上云的好处。 分布式消息服务RocketMQ具有如下产品优势，旨在打造一个即开即用、全托管、低延迟、弹性高可靠、动态扩展、便捷管理和多样功能的消息队列。 即开即用 简单几步即可在云上构建自己专属的消息服务，RocketMQ实例创建完成后，使用实例提供的访问地址即可快速接入。兼容开源RocketMQ，业务代码无需改造，即可上云。 全托管服务 分布式消息服务RocketMQ提供自动部署与完备的运维系统和售后服务，提供包括监控告警在内的多种运维手段，业务无需过多关注分布式消息服务RocketMQ的部署与运维工作，可以专注于自身业务的开发。 低延迟 基于天翼云网络部署，在内网访问可达微秒级时延。 弹性高可靠 基于Raft协议实现集群内部节点的管理，及时发现故障节点并进行流量迁移，保证业务的连续性可靠。 动态扩展 提供业务集群动态扩容的能力，根据业务需要动态扩容集群规模。 便捷管理 提供监控告警、消息追踪和链路诊断等多样的监控定位手段，方便问题定位和日常维护。 多样功能 提供顺序延迟、定时、重投、死信、过滤和事务消息等多样的业务功能，适配多样化的业务场景。

结果验证 1. 使用事件源触发一个事件事件。 2. 您可以在分布式消息服务RocketMQ管理控制台确认是否接收到事件。 1. 登录分布式消息服务RocketMQ管理控制台，然后在左侧导航栏选择实例列表。 2. 在实例列表页面单击目标实例名称。 3. 在消息查询页面，按时间查询即可查看消息内容，如图2所示。 图2 在分布式消息服务RocketMQ管理控制台中查看消息详情