操作说明 本方案基于 天翼云 ROS（资源编排服务）控制台，实现 Kafka 消息队列实例、Topic、消费组及关联网络资源的快速搭建，有效提升消息中间件资源交付效率，降低人工配置与运维成本。 方案内置标准化全链路编排模板，一站式完成 VPC 网络、子网、安全组、Kafka 集群实例、Topic 分区、消费组的联动部署，支持自定义规格配置、存储类型选择、节点数调整等灵活参数化设置，用户可根据消息队列业务规模、吞吐量与存储需求快速适配，高效完成生产级 Kafka 消息服务资源部署。 适用场景 Kafka 消息队列集群快速搭建 一键完成生产级 Kafka 集群 + Topic + 消费组全链路部署，无需分步手动配置，大幅缩短资源交付周期。 消息中间件多环境标准化部署 统一配置规范，支持测试、预发、生产等多环境 Kafka 资源标准化批量编排，避免配置不一致问题。 临时消息业务资源快速创建与释放 适配临时数据采集、压测、活动类消息业务场景，可快速创建 / 销毁 Kafka 资源，按需使用、节约成本 操作步骤 1. 登录控制中心。 2. 在控制台首页搜索“资源编排ROS”，或在左侧产品导航栏选择“管理工具 > 资源编排ROS”，进入[资源编排控制台](

操作说明 本方案基于 天翼云 ROS（资源编排服务）控制台，实现 Kafka 消息队列实例、Topic、消费组及关联网络资源的快速搭建，有效提升消息中间件资源交付效率，降低人工配置与运维成本。 方案内置标准化全链路编排模板，一站式完成 VPC 网络、子网、安全组、Kafka 集群实例、Topic 分区、消费组的联动部署，支持自定义规格配置、存储类型选择、节点数调整等灵活参数化设置，用户可根据消息队列业务规模、吞吐量与存储需求快速适配，高效完成生产级 Kafka 消息服务资源部署。 适用场景 Kafka 消息队列集群快速搭建 一键完成生产级 Kafka 集群 + Topic + 消费组全链路部署，无需分步手动配置，大幅缩短资源交付周期。 消息中间件多环境标准化部署 统一配置规范，支持测试、预发、生产等多环境 Kafka 资源标准化批量编排，避免配置不一致问题。 临时消息业务资源快速创建与释放 适配临时数据采集、压测、活动类消息业务场景，可快速创建 / 销毁 Kafka 资源，按需使用、节约成本 操作步骤 1. 登录控制中心。 2. 在控制台首页搜索“资源编排ROS”，或在左侧产品导航栏选择“管理工具 > 资源编排ROS”，进入[资源编排控制台](

topicconfigs)。 − 图中⑤区域表示队列监控数据统计。 − 图中⑥区域表示队列分区信息，包括分区消息数(Latest Offset)，分区leader(Leader)，副本列表(Replicas)，同步副本列表(In Sync Replicas)。 队列详情页 查看消费组列表 导航栏中单击Consumers页签，即可查看当前集群中的消费组列表。 集群的消费组列表 查看消费组详情页 单击消费组名称可进入消费组详情页面，展示消费组消费的所有队列列表以及每个队列的可消费数(Total Lag)。 消费组详情页面 查看消费组队列详情页 单击队列名称，即可进入详情页面，查看消费组消费在队列中每个分区的消费状态。包括分区编号(Partition)，分区消息数(LogSize)，分区消费进度(Consumer Offset)，分区剩余可消费数(Lag)，最近消费该分区的消费者(Consumer Instance Owner)。 消费组队列详情页面

Java客户端必须使用Push Consumer 使用Pull可以实现的所有场景，均可使用Push实现，并且更简单。 Push其实是长轮询的Pull（依然是由客户端发起），在客户端通过配置参数是可以实现流控的，并不会出现服务端的流量压垮客户端的情况。 Push封装了拉取消息，分发给消费线程的线程模型，非流控的情况下，由后台线程主动拉取消息，并缓存在本地，消费线程池有空闲线程时，分发给消费线程，在有堆积量的情况下，可以保证消费线程一直工作，性能更高（备注：Pull只提供了拉取消息的功能，并且何时去拉取，拉取时机，这些都需要应用去控制；分发给消费线程的逻辑需要应用封装，除了增加应用工作量外，还可能有不稳定、性能问题等）。 Push经过多个大型项目的长时间的使用，更成熟稳定。 Push会自动订阅重试队列，不需要再次调用拉取重试队列的API来取得重试队列的消息（备注：Pull需要另外调用API拉取重试队列的消息）。 Pull是一种遗留的消费模式（兼容早期的API），新开发的应用，或者未上线的应用，都要求使用Push消费模式。

介绍分布式消息服务RabbitMQ查看虚拟主机操作内容。 场景描述 在RabbitMQ中，虚拟主机（Virtual Host）是一个逻辑隔离的消息代理环境，用于分隔不同应用或不同团队之间的消息流。当需要查看虚拟主机的信息时，可能有以下场景描述： 管理员查看虚拟主机配置：管理员需要查看虚拟主机的配置信息，包括虚拟主机的名称、权限设置、连接数限制等。这可以帮助管理员了解当前系统的虚拟主机情况，进行配置管理和优化。 开发人员查看虚拟主机队列信息：开发人员需要查看虚拟主机中队列的信息，包括队列的名称、消息数量、消费者数量等。这可以帮助开发人员了解当前队列的状态，监控消息的流动情况，进行故障排查和性能调优。 运维人员查看虚拟主机连接信息：运维人员需要查看虚拟主机的连接信息，包括连接的客户端IP、连接数、协议等。这可以帮助运维人员监控连接的使用情况，检测异常连接，进行资源管理和安全审计。 操作步骤 （1）在虚拟主机管理页面，点击目标虚拟主机名称，即可参看虚拟主机详情。 （2）查看虚拟主机概览，主要展示最近时间段各类型消息数量和消息tps统计。 （3）参看用户配置权限及读写权限信息。 （4）查看用户主题权限信息。

数据保护技术 RabbitMQ提供了多种数据保护技术，以确保数据在传输和存储过程中的机密性和完整性。 1. 跨AZ容灾：在不同的可用区部署多个RabbitMQ节点，确保节点之间的数据复制和同步。这样，当一个可用区发生故障时，其他可用区上的节点可以继续提供服务。 2. 副本冗余：RabbitMQ副本冗余是一种保证消息队列的高可用性和数据冗余的策略。通过在多个节点上创建副本，可以确保即使一个节点发生故障，其他节点上的副本仍然可以提供服务。 3. 数据持久化：RabbitMQ数据持久化是通过将队列、消息和交换器进行持久化，确保消息队列中的数据在节点重启或故障时不会丢失。 服务韧性 RabbitMQ服务的韧性是指其在面对各种故障和异常情况时能够保持可用性和可靠性的能力。以下是保障RabbitMQ服务韧性的关键方面： 1. AZ内实例容灾：在不同的可用区内部署多个RabbitMQ节点，使它们能够相互复制和同步消息。这样即使一个可用区发生故障，其他可用区的节点仍然可以提供服务。 2. 数据容灾：RabbitMQ数据容灾是通过持久化消息、队列和交换器、备份和复制以及高可用性集群等策略，保护数据免受损失和故障影响的措施。

本文主要介绍分布式消息服务RabbitMQ的入门实践。 当您购买了RabbitMQ实例后，可以根据自身的业务需求使用分布式消息服务RabbitMQ提供的一系列常用实践。 表1 常用最佳实践 实践 描述 RabbitMQ业务迁移 介绍将线下单机或集群实例迁移到天翼云RabbitMQ实例的方案。 RabbitMQ队列迁移 扩容节点、删除队列可能会导致队列在各个节点分布不均衡，本文介绍如何设置队列负载均衡。

本文主要介绍消息队列 Kafka 通过SSL接入的最佳实践，从而帮助您更好的使用该产品。 文中的最佳实践基于消息队列 Kafka 的 Java 客户端；对于其它语言的客户端，其基本概念与思想是通用的，但实现细节可能有差异，仅供参考。 背景 云消息队列 Kafka 版实例如果选择SASLSSL接入时，可能会出现性能较差的情况。可以通过在客户端指定密码套件的方式，手动选择性能和安全性都较高的套件进行TLS通讯。 SSL握手时客户端发送Hello Client 并带上客户端支持的密码套件，服务端收到握手请求后，获取客户端带来的密码套件和服务端支持的密码套件取交集。密码套件加载顺序受客户端jdk版本影响，不同jdk版本，密码套件顺序不一样，可能会导致性能和安全性等无法保证。因此为了保证性能，SSL连接时客户端可以指定密码套件。 推荐的性能和安全性较高的密码套件 TLSECDHERSAWITHAES256GCMSHA384 TLSECDHERSAWITHAES128GCMSHA256 步骤 1.先按照SASLSSL协议接入Kafka，SASLSSL接入可参考文档 SASLSSL接入点接入 2.在此基础上增加 ssl.cipher.suites 配置 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLSSL"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"f6eca4dbc78df78d63fba980e448185f";");//注：上面的密码f6eca4dbc78df78d63fba980e448185f，为用户管理里面创建用户时填入的密码进行md5的结果，md5取32位小写 props.put("ssl.truststore.location","/kafka/client.truststore.jks"); props.put("ssl.truststore.password","sJses2tin1@23"); props.put("ssl.endpoint.identification.algorithm",""); props.put("ssl.cipher.suites","TLSECDHERSAWITHAES256GCMSHA384,TLSECDHERSAWITHAES128GCMSHA256");//密码套件支持多个，用半角逗号分开

本章节介绍了如何查询分布式消息服务RocketMQ实例的Topic。 操作场景 Topic创建成功后，查询Topic相关的配置和状态信息。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 单击需要查询的Topic名称，进入Topic详情页面。 在详情页上方可以查看Topic名称、关联代理、读队列个数、写队列个数和权限。 在详情页下方可以查看Topic在每个代理上的队列状态，包括队列ID、最小偏移量、最大偏移量和消息更新时间。还可以查看消费组消费此Topic的情况，包括消费组名称、最大重试次数和广播消费。 图 Topic详情

配置项 操作说明 示例 触发器类型 选择Kafka触发器。 Kafka触发器 名称 填写自定义的触发器名称。 kafkatrigger Kafka实例 选择已创建的Kafka实例。 Topic 选择已创建的Kafka实例的Topic。 Group ID • 快速创建 ：推荐方案。自动创建以GROUPFCTrigger{triggername}{uuid}命名的Group ID。 • 使用已有 ：选择Kafka实例已有的GroupID，请您注意不要与已有的业务混用GroupID，否则会影响已有的消息收发。 消费任务并发数 消费者的并发数量，有效取值范围为[1,20]，建议不超过Topic的分区数。该值同时影响投递到函数的并发数。 消费位点 选择消息的消费位点，即触发器从kafka消息队列开始拉取消息的位置。 • 最早位点 ：从最早位点开始消费。 • 最新位点 ：从最新位点开始消费。 最新位点 调用方式 选择函数调用方式。 • 同步调用 ：指触发器消费topic消息后投递到函数是同步调用，会等待函数响应后继续下一个消息投递。但消费任务并发数大于1时，多个消费者有可能会并发消费消息并投递，并发的情况视topic队列本身积存的消息而定。 • 异步调用 ：指触发器消费topic消息后投递到函数是异步调用，不会等待函数响应，可以快速消费事件。 同步调用 触发器启用状态 创建触发器后是否立即启用。默认选择开启，即创建触发器后立即启用触发器。 启用 推送配置 • 批量推送条数 ：批量推送的最大值，积压值达到后立刻推送，取值范围为[1,10000]。 • 批量推送间隔 ：批量推送的最大时间间隔，达到后立刻推送，单位秒，取值[0,15]。默认0无需等待，数据直接推送。 • 推送格式 ：函数收到的事件格式，详情请查阅触发器事件消息格式。 重试策略 消息推送函数失败后重试的策略，共两种： • 指数退避 ：指数退避重试，重试5次，重试周期为2,4,8,16,32（秒）。 • 线性退避 ：线性退避重试，重试5次，重试周期为1,2,3,4,5（秒）。 容错策略 当重试次数耗尽后仍然失败时的处理方式： • 允许容错 ：当异常发生并超过重试策略配置时直接丢弃。 • 禁止容错 ：当异常发生并超过重试策略配置时继续阻塞执行。 死信队列 当容错策略为：允许容错时，可以额外开启死信队列。当开启死信队列时且异常发生并超过重试策略配置时，消息会被投递到指定的消息队列里，当前只支持投递到kafka和rocketmq

指标类别 指标 指标名称 指标说明 单位 数据类型 默认聚合方式 PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） uri serviceUri PahoSubscriber连接的MQTT服务端uri ENUM LAST PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） msgType 报文类型 发送的报文类型 ENUM LAST PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） concurrentMax 最大并发 最大并发 INT MAX PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） errorCount 错误次数 错误次数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） errorTraceId 错误traceId 采集周期内发生错误的调用链对应的traceid STRING LAST PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） slowTraceId 慢traceId 采集周期内最慢的调用链对应的traceid STRING LAST PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） invokeCount 调用次数 调用次数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） lastError 错误信息 错误信息 STRING LAST PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） maxTime 最大响应时间 最大响应时间 INT MAX PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） totalTime 总响应时间 总响应时间 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range1 010ms 响应时间在010ms范围请求数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range2 10100ms 响应时间在10100ms范围请求数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range3 100500ms 响应时间在100500ms范围请求数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range4 5001000ms 响应时间在5001000ms范围请求数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range5 110s 响应时间在110s范围请求数 INT SUM PahoSubscriber接收Publish报文监控（message，PahoSubscriber接收Publish报文监控。） range6 10s以上 响应时间在10s以上请求数 INT SUM PahoSubscriber接收Publish报文节点维度监控（uriMessage，PahoSubscriber接收Publish报文监控。） uri serviceUri PahoSubscriber连接的MQTT服务端uri ENUM LAST PahoSubscriber接收Publish报文节点维度监控（uriMessage，PahoSubscriber接收Publish报文监控。） errorCount 错误次数 错误次数 INT SUM PahoSubscriber接收Publish报文节点维度监控（uriMessage，PahoSubscriber接收Publish报文监控。） invokeCount 调用次数 调用次数 INT SUM PahoSubscriber接收Publish报文节点维度监控（uriMessage，PahoSubscriber接收Publish报文监控。） totalTime 总响应时间 总响应时间 INT SUM 异常（exception，PahoSubscriber调用的异常信息统计。） exceptionType 异常类型 异常类型 ENUM LAST 异常（exception，PahoSubscriber调用的异常信息统计。） causeType 异常类 发生异常的类 ENUM LAST 异常（exception，PahoSubscriber调用的异常信息统计。） count 次数 该异常的发生次数 INT SUM 异常（exception，PahoSubscriber调用的异常信息统计。） message 异常消息 该异常产生的异常消息 STRING LAST 异常（exception，PahoSubscriber调用的异常信息统计。） stackTrace 异常堆栈 该异常产生的堆栈信息 CLOB LAST PahoSubscriber主题维度监控（clientReceive，PahoSubscriber主题维度监控。） clientId clientId clientId ENUM LAST PahoSubscriber主题维度监控（clientReceive，PahoSubscriber主题维度监控。） topic 主题 主题 ENUM LAST PahoSubscriber主题维度监控（clientReceive，PahoSubscriber主题维度监控。） msgReceivedCount 消息接收次数 消息接收次数 INT SUM PahoSubscriber主题维度监控（clientReceive，PahoSubscriber主题维度监控。） bytesReceived 消息接收字节数 消息接收字节数 INT SUM PahoSubscriber版本（version，PahoSubscriber版本。） version 版本 版本 STRING LAST PahoSubscriber主题汇总（total，PahoSubscriber主题汇总信息统计。） msgReceivedCount 消息接收次数 总消息接收次数 INT SUM PahoSubscriber主题汇总（total，PahoSubscriber主题汇总信息统计。） bytesReceived 消息接收字节数 总消息接收字节数 INT SUM

专属队列 专属队列是指队列对应的资源为专属资源，空闲时不释放，即无论是否使用均保留资源的队列类型。专属队列可以保证提交作业时资源一定存在。 队列弹性扩缩容 DLI提供了队列弹性扩缩容的功能。用户在创建指定规格队列后，可根据需要进行弹性扩缩容。 根据业务情况，手动更改队列规格。具体操作请参考队列弹性扩缩容。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列弹性扩缩容定时任务 DLI提供了队列弹性扩缩容定时任务的功能。用户在创建队列后，可根据需要进行弹性扩缩容定时任务。 根据业务情况，设置队列自动扩缩容的时间，由系统定时触发队列扩缩容。具体操作请参考弹性扩缩容定时任务。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列自动扩缩容 Flink作业使用队列，DLI可根据作业大小自动触发扩缩容，用户无需进行操作。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列管理页面 队列管理主要包括如下功能： 队列权限管理 创建队列 删除队列 修改队列网段 队列弹性扩缩容 弹性扩缩容定时任务 测试地址连通性 创建消息通知主题 说明 DLI作业执行失败需要通过SMN发送通知消息，因此需要获得访问和使用SMN（消息通知服务）的SMN Administrator权限。 队列管理页面显示用户创建所有的队列和服务预置的default队列。队列列表默认按创建时间排列，创建时间最近的队列显示在最前端。 队列管理参数 参数 参数说明 名称 队列的名称。 类型 队列的类型。 SQL队列 通用队列 Spark队列（兼容老版本） 规格 队列大小，单位：CUs。 CUs是队列的计价单位。1CUs1Core 4GMem。不同规格的队列对应的计算能力不一样，规格越高计算能力越好。 实际CUs 当前队列实际大小值。 弹性扩缩容 定时扩缩容的目标CU值，或当前规格CU值的最大值和最小值。 用户名 队列所有者。 描述 创建队列时，对队列的描述。如果无描述，则显示“”。 操作 删除：删除所选队列。如果队列中有正在提交或者正在运行的作业，将不支持删除操作。 权限管理：查看队列对应的用户权限信息以及对其他用户授权。 更多 − 重启：强制重启队列。 说明 只有SQL队列有“重启”操作。 − 弹性扩缩容：可以根据需要选择“扩容”或“缩容”，目标值大小必须为16CU的整数倍。 − 弹性扩缩容定时任务：可以根据业务周期或使用情况，在不同的时间或周期内设置不同的队列大小，系统将定时自动进行“扩容”或“缩容”。目标值大小必须为16CU的整数倍。 − 修改网段：使用DLI增强型跨源时，DLI队列网段与数据源网段不能重合，可根据需要进行修改。 − 测试地址连通性：测试队列到指定地址是否可达，支持域名和ip，可指定端口。

根据ID查询 根据消息ID查询唯一消息，选择消费组后，可以查询到该消息是否被该消费组消费过。 点击“查看”，可以查询该消息的包体内容。 消费状态标志含义： （1）Toconsume：未消费。 （2）Consumed：已签收。 （3）Consuming：已拉取，未签收。 根据offset查询 根据指定队列指定偏移量查询唯一消息，选择订阅组后，可以查询到该消息是否被该订阅组消费过。 点击“查看”，可以查询该消息的包体内容。 基于Topic查询 死信队列查询

本节介绍了分布式消息服务RabbitMQ产品常见的名词解释。 Vhost 虚拟主机（Virtual Host），类似于Namespace命名空间的概念，逻辑隔离，每个用户里可以创建多个Vhost，每个Vhost可以创建若干个Exchange和Queue。 Queue 消息队列，每个消息都会被投入到一个或者多个Queue里。 Producer 消息生产者，即投递消息的程序。 Consumer 消息消费者，即接受消息的程序。 Connection TCP 连接，Producer 或 Consumer 与消息队列间的物理 TCP 连接。 Connection将应用与分布式消息服务RabbitMQ连接在一起。Connection会执行认证、IP解析、路由等底层网络任务。应用与分布式消息服务RabbitMQ建立Connection需要多个TCP报文交互，因而会消耗较多的网络资源和分布式消息服务RabbitMQ资源。大量的Connection会对分布式消息服务RabbitMQ造成巨大压力，甚至触发分布式消息服务RabbitMQ SYN洪水攻击防护，导致分布式消息服务RabbitMQ无响应，进而影响业务。 Channel 在客户端的每个物理TCP连接里，可建立多个Channel，每个Channel代表一个会话任务。 Channel是物理TCP连接中的虚拟连接。当应用通过Connection与分布式消息服务RabbitMQ建立连接后，所有的AMQP协议操作（例如创建队列、发送消息、接收消息等）都会通过Connection中的Channel完成。Channel可以复用Connection，即一个Connection下可以建立多个Channel。Channel不能脱离Connection独立存在，而必须存活在Connection中。当某个Connection断开时，该Connection下的所有Channel都会断开。当大量应用需要与分布式消息服务RabbitMQ建立多个连接时，建议您使用Channel来复用Connection，从而减少网络资源和分布式消息服务RabbitMQ资源消耗。 Connection和Channel的使用建议 保持Connection长连接，请勿频繁开启或关闭Connection。如果确实需要频繁开启或关闭连接，请使用Channel。 一个进程对应一个Connection，一个进程中的多个线程则分别对应一个Connection中的多个Channel。 Producer和Consumer分别使用不同的Connection进行消息发送和消费。

本文主要介绍消息队列 Kafka 发布者的最佳实践，从而帮助您更好的使用该产品。 文中的最佳实践基于消息队列 Kafka 的 Java 客户端；对于其它语言的客户端，其基本概念与思想是通用的，但实现细节可能有差异，仅供参考。 Kafka 发送示例代码片段 Key 和 Value Kafka 0.10.0.0 的消息字段只有两个：Key 和 Value。Key 是消息的标识，Value 即消息内容。为了便于追踪，重要消息最好都设置一个唯一的 Key。通过 Key 追踪某消息，打印发送日志和消费日志，了解该消息的发送和消费情况。 失败重试 在分布式环境下，由于网络等原因，偶尔的发送失败是常见的。导致这种失败的原因有可能是消息已经发送成功，但是 Ack 失败，也有可能是确实没发送成功。 消息队列 Kafka 是 VIP 网络架构，会主动掐掉空闲连接（30 秒没活动），也就是说，不是一直活跃的客户端会经常收到 “connection rest by peer” 这样的错误，因此建议都考虑重试消息发送。 异步发送 发送接口是异步的；如果你想得到发送的结果，可以调用metadataFuture.get(timeout, TimeUnit.MILLISECONDS)。 线程安全 Producer 是线程安全的，且可以往任何 Topic 发送消息。通常情况下，一个应用对应一个 Producer 就足够了。 Acks Acks的说明如下： acks0，表示无需服务端的 Response，性能较高，丢数据风险较大； acks1，服务端主节点写成功即返回 Response，性能中等，丢数据风险中等，主节点宕机可能导致数据丢失； acksall，服务端主节点写成功，且备节点同步成功，才返回 Response，性能较差，数据较为安全，主节点和备节点都宕机才会导致数据丢失。 一般建议选择 acks1，重要的服务可以设置 acksall。

本文主要介绍PahoPublisher监控 介绍APM采集的PahoPublisher监控指标的类别、名称、含义等信息。 表 PahoPublisher监控指标说明 指标类别 指标 指标名称 指标说明 单位 数据类型 默认聚合方式 ::::::: PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） uri serviceUri PahoPublisher连接的MQTT服务端uri ENUM LAST PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） msgType 报文类型 发送的报文类型 ENUM LAST PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） concurrentMax 最大并发 最大并发 INT MAX PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） errorCount 错误次数 错误次数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） errorTraceId 错误traceId 采集周期内发生错误的调用链对应的traceid STRING LAST PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） slowTraceId 慢traceId 采集周期内最慢的调用链对应的traceid STRING LAST PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） invokeCount 调用次数 调用次数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） lastError 错误信息 错误信息 STRING LAST PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） maxTime 最大响应时间 最大响应时间 INT MAX PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） totalTime 总响应时间 总响应时间 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range1 010ms 响应时间在010ms范围请求数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range2 10100ms 响应时间在10100ms范围请求数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range3 100500ms 响应时间在100500ms范围请求数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range4 5001000ms 响应时间在5001000ms范围请求数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range5 110s 响应时间在110s范围请求数 INT SUM PahoPublisher发送Publish报文监控（message，PahoPublisher发送Publish报文监控。） range6 10s以上 响应时间在10s以上请求数 INT SUM PahoPublisher发送Publish报文节点维度监控（uriMessage，PahoPublisher发送Publish报文节点维度监控。） uri serviceUri PahoPublisher连接的MQTT服务端uri ENUM LAST PahoPublisher发送Publish报文节点维度监控（uriMessage，PahoPublisher发送Publish报文节点维度监控。） errorCount 错误次数 错误次数 INT SUM PahoPublisher发送Publish报文节点维度监控（uriMessage，PahoPublisher发送Publish报文节点维度监控。） invokeCount 调用次数 调用次数 INT SUM PahoPublisher发送Publish报文节点维度监控（uriMessage，PahoPublisher发送Publish报文节点维度监控。） totalTime 总响应时间 总响应时间 INT SUM 异常（exception，PahoPublisher调用的异常信息统计。） exceptionType 异常类型 异常类型 ENUM LAST 异常（exception，PahoPublisher调用的异常信息统计。） causeType 异常类 发生异常的类 ENUM LAST 异常（exception，PahoPublisher调用的异常信息统计。） count 次数 该异常的发生次数 INT SUM 异常（exception，PahoPublisher调用的异常信息统计。） message 异常消息 该异常产生的异常消息 STRING LAST 异常（exception，PahoPublisher调用的异常信息统计。） stackTrace 异常堆栈 该异常产生的堆栈信息 CLOB LAST PahoPublisher主题维度监控（clientPublish，PahoPublisher主题维度监控。） clientId clientId clientId ENUM LAST PahoPublisher主题维度监控（clientPublish，PahoPublisher主题维度监控。） topic 主题 主题 ENUM LAST PahoPublisher主题维度监控（clientPublish，PahoPublisher主题维度监控。） msgSentCount 消息推送次数 消息推送次数 INT SUM PahoPublisher主题维度监控（clientPublish，PahoPublisher主题维度监控。） bytesSent 消息推送字节数 消息推送字节数 INT SUM PahoPublisher版本（version，PahoPublisher版本。） version 版本 版本 STRING LAST PahoPublisher主题汇总（total，PahoPublisher主题汇总信息统计。） msgSentCount 消息推送次数 总的消息推送次数 INT SUM PahoPublisher主题汇总（total，PahoPublisher主题汇总信息统计。） bytesSent 消息推送字节数 总消息推送字节数 INT SUM

介绍分布式消息服务RocketMQ的安全方案,包括支持TLS传输加密、权限控制、Topic资源访问权限控制等内容。 安全价值 RocketMQ的安全对用户有以下几个重要价值： 1. 保护数据安全：RocketMQ的安全机制可以保护消息的机密性和完整性，防止敏感数据泄露或被篡改。这对于处理包含个人信息、商业机密等敏感数据的应用程序非常重要。 2. 防止未经授权的访问：RocketMQ的访问控制功能可以限制对消息队列的访问权限，只有具有相应权限的用户才能发送和消费消息。这可以防止未经授权的用户访问和操作消息队列，保护系统的安全性。 3. 合规性要求：对于一些行业和法规要求较高的场景，如金融、医疗等，RocketMQ的安全特性可以帮助用户满足合规性要求，确保数据的安全和合规性。 4. 提供安全审计功能：RocketMQ的安全审计功能可以记录和追踪对消息队列的操作，包括发送、消费、订阅等。这可以帮助用户监控和检测潜在的安全风险，及时发现和应对安全事件。 5. 增强用户信任：通过提供安全性能和功能，RocketMQ可以增强用户对系统的信任感。用户可以放心地使用RocketMQ来处理重要的消息传输和处理任务，而不必担心数据的安全问题。 综上所述，RocketMQ的安全性对用户来说具有重要的价值，可以保护数据安全，防止未经授权的访问，满足合规性要求，提供安全审计功能，并增强用户对系统的信任感。

消息超过老化时间，消息仍存在的原因 问题现象： 消息超过设置的老化时间（如果Topic已经设置了老化时间，此时“配置参数”中的log.retention.hours值将不对此Topic生效。仅在Topic中未设置老化时间时，“配置参数”中的log.retention.hours值才会对此Topic生效。），消息仍存在。 可能原因1： Topic的每个分区都是由多个大小相同的segment文件组成，每个segment文件的大小为500MB，当segment文件存储的消息大小到达500MB后，才会新建下一个segment文件。Kafka删除消息是删除segment文件，而不是删除一条消息。Kafka要求至少保留一个segment文件用来存储消息，如果正在使用的segment文件中包含超过老化时间的消息，由于此时segment文件不会被删除，所以超过老化时间的消息也不会被删除。 处理方法： 等待segment文件被使用完，或者删除超过老化时间的消息所在的Topic。 可能原因2： Topic中存在一条create time为未来时间的消息（例如当前时间为1月1日，create time设置成了2月1日），此消息在72小时后，并不会被老化，导致在此消息后创建的其他消息都不会被老化。 处理方法： 删除create time为未来时间的消息所在的Topic。 Kafka实例是否支持延迟消息？ 不支持延迟消息。 如何查看堆积消息数？ 通过以下任意一种方法，查看堆积消息数。 在Kafka控制台的“消费组管理”页面，单击待查看堆积消息的消费组名称，进入消费组详情页。在“消费进度”页签，查看消费组中每个Topic的总堆积数。具体步骤，请参考查询消费组信息。 在Kafka控制台的“监控”页面的“消费组”页签中，“消费组”选择待查看堆积消息数的消费组名称，“队列”选择“全部队列”，“消费组可消费消息数”表示此消费组中所有Topic的堆积消息数之和。查看监控数据的具体步骤，请参考查看监控数据。 在云监控页面的“消费组”页签中，“消费组”选择待查看堆积消息数的消费组名称，“队列”选择“全部队列”，“消费组可消费消息数”表示此消费组中所有Topic的堆积消息数之和。查看监控数据的具体步骤，请参考查看监控数据。 在Kafka客户端，在“/{命令行工具所在目录}/kafka{version}/bin/”目录下，通过 kafkaconsumergroups.sh bootstrapserver {kafka连接地址} describe group {消费组} 命令查看消费组中每个Topic的堆积消息数。“LAG”表示每个Topic的总堆积数。 图 查看每个Topic的总堆积数 说明 如果Kafka实例开启SASL认证，则以上命令还需要增加SASL认证的“consumer.properties”配置文件参数： commandconfig {SASL认证的consumer.properties配置文件} ，“consumer.properties”配置文件参考开启SASL认证的Kafka命令行连接说明。

本章节介绍了如何修改分布式消息服务RocketMQ实例的Topic。 操作场景 Topic创建成功后，您可以根据业务需要修改以下参数：读队列个数、写队列个数、权限和添加关联代理。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 选择以下任意一种方法修改Topic参数。 在待修改的Topic所在行，单击“编辑”，弹出“编辑Topic”页面。 单击待修改的Topic名称，进入Topic详情页面。在页面右上角，单击“编辑”，弹出“编辑Topic”页面。 5. 修改如表1所示配置信息。 表 Topic参数说明 参数 说明 权限 Topic的权限，包括发布+订阅、发布和订阅。 关联代理 修改读队列个数或写队列个数。 lic读数读队列个数：Top据可用的总队列数。 写队列个数：Topic写数据可用的总队列数。 如果创建Topic时未关联所有代理，单击“添加关联代理”，可以在其他代理上创建Topic，并设置读队列个数和写队列个数。 6. 修改完成后，单击“确定”。

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

topicconfigs]( " "))。 图中⑤区域表示队列监控数据统计。 图中⑥区域表示队列分区信息，包括分区消息数(Latest Offset)，分区leader(Leader)，副本列表(Replicas)，同步副本列表(In Sync Replicas)。 图 队列详情页 6.查看消费组列表 导航栏中单击 Consumers ，即可查看当前集群中的消费组列表。 说明 只显示14天内有消费记录的消费组。 图 集群的消费组列表 7.查看消费组详情页 单击消费组名称可进入消费组详情页面，展示消费组消费的所有队列列表以及每个队列的可消费数（Total Lag）。 图 消费组详情页面 8.查看消费组队列详情页 单击队列名称，即可进入详情页面，查看消费组消费在队列中每个分区的消费状态。包括分区编号（Partition）、分区消息数（LogSize）、分区消费进度（Consumer Offset）、分区剩余可消费数（Lag）和最近消费该分区的消费者（Consumer Instance Owner）。 图 消费组队列详情页面

触发器名称 调用方式 详细介绍 定时触发器 同步调用 按照指定的时间间隔触发工作流执行。 HTTP触发器 同步调用 当通过HTTP请求触发工作流时，工作流会根据请求参数执行。 Kafka触发器 同步/异步调用 当消息队列中出现新消息时，触发工作流执行。 RocketMQ触发器 同步/异步调用 当消息队列中出现新消息时，触发工作流执行。 云原生网关触发器 同步调用 当通过云原生网关触发工作流时，工作流会根据请求参数执行。

为了进一步提升云服务的安全性，加强实例访问的默认保护机制，对新创建MQTT实例的默认安全组策略进行优化调整。 调整内容 自本公告发布之日起， 新开通MQTT实例使用的安全组未配置vpc内入方向安全组规则，将默认禁止同VPC内访问MQTT实例服务端口。此变更旨在遵循最小权限安全原则，降低因配置疏忽导致的安全风险。 影响说明 若您使用的安全组未进行额外配置，新建的实例将无法被同VPC内的云主机访问。 应对措施 如需恢复VPC内访问，请您手动配置安全组规则，具体操作如下： 在安全组入方向规则中添加以下策略： 授权策略：允许。 协议类型：TCP。 端口范围：MQTT实例服务端口。 源地址：您的VPC网段（如：192.168.0.0/16）。 配置完成后，同VPC内的云主机即可正常访问MQTT实例服务。 建议 为保障业务安全，建议您根据实际需求精确设置访问源IP或网段，避免开放不必要的访问权限。 此次调整仅影响新创建的实例，已有实例的访问策略保持不变，不受此次调整影响。感谢您对云服务安全的支持与理解。如有疑问，请联系技术支持。 特此公告。

本文主要介绍如何配置分布式消息服务RabbitMQ必须的监控告警。 本章节主要介绍部分监控指标的告警策略，以及配置操作。在实际业务中，建议按照表1告警策略，配置监控指标的告警规则。 表1 RabbitMQ实例配置告警的指标 指标名称 告警策略 指标说明 解决方案 内存高水位状态 告警阈值：原始值>1 连续触发次数：1 告警级别：致命 告警阈值为1表示触发内存高水位，会阻塞消息生产 加快消费 采用生产者确认的发送模式，并监控生产端消息生产速度和时长，当消息生产时长有明显增加时进行流控措施 磁盘高水位状态 告警阈值：原始值>1 连续触发次数：1 告警级别：致命 告警阈值为1表示触发磁盘高水位，会阻塞消息生产 减少惰性队列的消息堆积 减少持久化队列的消息堆积 删除队列 内存使用率 告警阈值：原始值>业务预期使用率（推荐30%） 连续触发次数：连续3~5个周期 告警级别：重要 该指标需要分别为每个节点设置内存使用率告警，避免触发内存高水位阻塞生产 加快消费 采用生产者确认的发送模式，并监控生产端消息生产速度和时长，当消息生产时长有明显增加时进行流控措施 CPU使用率 告警阈值：原始值>业务预期使用率（推荐70%） 连续触发次数：连续3~5个周期 告警级别：重要 该指标需要分别为每个节点设置CPU使用率告警，CPU使用率过高可能会影响生产速度 减少镜像队列个数 对于集群实例，建议扩容节点个数，然后进行节点间重平衡 可消费消息数 告警阈值：原始值>业务预期可消费消息数 连续触发次数：1 告警级别：重要 可消费消息数过多表示消息堆积 请参考消息堆积的解决办法 未确认消息数 告警阈值：原始值>业务预期未确认消息数 连续触发次数：1 告警级别：重要 未确认消息数过多可能会导致消息堆积 检查消费者是否异常 检查消费者逻辑是否消耗时间过长 连接数 告警阈值：原始值>业务预期连接数 连续触发次数：1 告警级别：重要 连接数突增可能是流量变大的预警 需检查业务是否正常，可参考其他告警 通道数 告警阈值：原始值>业务预期通道数 连续触发次数：1 告警级别：重要 通道数突增可能是流量变大的预警 需检查业务是否正常，可参考其他告警 Erlang进程数 告警阈值：原始值>业务预期进程数 连续触发次数：1 告警级别：重要 进程数突增可能是流量变大的预警 需检查业务是否正常，可参考其他告警

本章节介绍 Topic管理 。 操作场景 Topic创建成功后，查询Topic相关的配置和状态信息。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 单击需要查询的Topic名称，进入Topic详情页面。 在详情页上方可以查看Topic名称、关联代理、读队列个数、写队列个数和权限。 在详情页下方可以查看Topic在每个代理上的队列状态，包括队列ID、最小偏移量、最大偏移量和消息更新时间。还可以查看消费组消费此Topic的情况，包括消费组名称、最大重试次数和广播消费。 图1 Topic详情

本文主要介绍天翼云分布式消息服务RabbitMQ的产品定义。 天翼云分布式消息服务RabbitMQ完全兼容开源RabbitMQ，为您提供即开即用、消息特性丰富、灵活路由、高可用、监控和告警等特性，广泛应用于秒杀、流控、系统解耦等场景。 即开即用 天翼云分布式消息服务RabbitMQ提供单机和集群的消息实例，拥有丰富内存规格，您可以通过控制台直接下单购买并创建，无需单独准备服务器资源。 消息特性丰富 支持AMQP协议，支持普通消息、广播消息、死信、延迟消息等特性。 灵活路由 在RabbitMQ中，生产者将消息发送到交换器，由交换器将消息路由到队列中。交换器支持Direct、Topic、Headers和Fanout四种路由方式，同时支持交换机组合和自定义。 高可用 RabbitMQ集群提供镜像队列，可通过镜像在其他节点同步数据，单节点宕机时，仍可通过唯一的访问地址对外提供服务，数据不丢失。 监控和告警 支持对RabbitMQ实例状态进行监控，支持对集群每个代理的内存、CPU、网络流量等进行监控。如果集群或节点状态异常，将触发告警。

等保级别 套餐型号 套餐安全功能特性 可选安全组件及规格 二级等保 入门版 云防火墙（含WAF/防篡改模块） 云防火墙：100M/200M/300M/400M/500M/800M/1G/1.6G 二级等保 入门版 云日志审计 云日志审计：10/20/50/100/200/300/500资产 二级等保 入门版 主机漏扫 安全管理中心基础版：10/20/50/100/200/300/500资产 二级等保 入门版 安全管理中心 二级等保 基础版 云防火墙（含WAF/防篡改模块） 云防火墙：100M/200M/300M/400M/500M/800M/1G/1.6G 二级等保 基础版 云日志审计 云日志审计：10/20/50/100/200/300/500资产 二级等保 基础版 终端安全EDR（防病毒/补丁/基线/微隔离） 终端安全EDR：实际资产数 二级等保 基础版 云堡垒机 云堡垒机：10/20/50/100/200/300/500资产 二级等保 基础版 主机漏扫 安全管理中心基础版：10/20/50/100/200/300/500资产 二级等保 基础版 安全管理中心 三级等保 高级版 云防火墙（含WAF/防篡改模块） 云防火墙：100M/200M/300M/400M/500M/800M/1G/1.6G 三级等保 高级版 云日志审计 云日志审计：10/20/50/100/200/300/500资产 三级等保 高级版 终端安全EDR（防病毒/补丁/基线/微隔离） 终端安全EDR：实际资产数 三级等保 高级版 云堡垒机 云堡垒机：10/20/50/100/200/300/500资产 三级等保 高级版 主机漏扫 安全管理中心基础版：10/20/50/100/200/300/500资产 三级等保 高级版 安全管理中心 三级等保 高级版 云数据库审计 云数据库审计：100M/200M/400M/600M 三级等保 旗舰版 云防火墙（含WAF/防篡改模块） 云防火墙：100M/200M/300M/400M/500M/800M/1G/1.6G 三级等保 旗舰版 云日志审计 云日志审计：10/20/50/100/200/300/500资产 三级等保 旗舰版 终端安全EDR（防病毒/补丁/基线微隔离） 终端安全EDR：实际资产数 三级等保 旗舰版 云堡垒机 云堡垒机：10/20/50/100/200/300/500资产 三级等保 旗舰版 主机漏扫 安全管理中心高级版：10/20/50/100/200/300/500资产 三级等保 旗舰版 应用漏扫 三级等保 旗舰版 安全管理中心 三级等保 旗舰版 云数据库审计 云数据库审计：100M/200M/400M/600M

介绍分布式消息服务RabbitMQ虚拟主机权限管理内容。 场景描述 在RabbitMQ中，虚拟主机（Virtual Hosts, vhosts）是一种逻辑隔离机制，允许单个 RabbitMQ 实例划分为多个独立的环境。每个虚拟主机都有自己的交换机、队列、绑定和权限设置。虚拟主机的用户权限管理，支持对每个用户配置三种权限，分别是配置权限、写权限、读权限，用户在未配置这三种权限之前，无法访问对应的虚拟主机。三个权限的详情如下： 配置权限：允许用户创建、修改或删除交换器（Exchange）、队列（Queue）和绑定关系。 写权限：允许用户向交换器发布消息。 读权限：允许用户消费队列中的消息或获取队列状态。 权限配置 在RabbitMQ中，虚拟主机的用户权限配置，采用的是正则表达式匹配方式， 只有满足正则表达式的资源才有权限。比如配置写权限为："."，表示该用户可以向虚拟主机下全部的交换器发布消息。配置写权限为："^$”，表示该用户不允许向任何交换器发布消息。配置写权限为："^exchange."，表示该用户可以向虚拟主机下以"exchange"开头的交换器发布消息。 最佳实践 1. 登录管理控制台。 2. 进入RabbitMQ管理控制台。 3. 在实例列表页在操作列，目标实例行点击“管理”。 4. 进入具体的实例管理页面，点击Vhost管理，在vhost列表页，针对指定的vhost，点击vhost进入具体的权限管理页： 5. 在具体的权限管理页，点击新建，选择指定用户，配置权限为"."、"."、"."，则该用户拥有对虚拟主机下全部资源的配置权限、写权限和读权限： 6. 点击权限列表的操作列的“修改”或“删除”按钮，可以修改或删除对应用户的权限：

本文介绍如何配置MQTT开源Java SDK的客户端自动连接。 要配置MQTT开源Java SDK的客户端自动连接，你需要设置以下参数和逻辑： 1、 设置自动重连：首先，你需要设置客户端的自动重连功能。在MQTT开源Java SDK中，你可以使用setAutomaticReconnect(true)方法来启用自动重连功能。这将使客户端在连接丢失后自动尝试重新连接。 MqttClient client new MqttClient(brokerUrl, clientId); client.setAutomaticReconnect(true); 2、 设置重连间隔：接下来，你可以设置自动重连的时间间隔。这将确定客户端在连接丢失后尝试重新连接的时间间隔。在MQTT开源Java SDK中，你可以使用setReconnectDelay方法来设置重连间隔，单位为毫秒。 client.setReconnectDelay(5000); // 设置重连间隔为5秒 3、 处理连接丢失事件：当连接丢失时，你可以通过实现MqttCallback接口来处理连接丢失事件。在connectionLost方法中，你可以添加自定义的处理逻辑，例如记录日志或触发重新连接。 client.setCallback(new MqttCallback() { @Override public void connectionLost(Throwable cause) { // 处理连接丢失事件 // 添加自定义的处理逻辑，例如记录日志或触发重新连接 } // 其他回调方法... }); 4、 处理重新连接事件：当客户端尝试重新连接时，你可以在reconnectComplete方法中添加自定义的处理逻辑。这个方法会在客户端成功重新连接到MQTT Broker后被调用。 client.setCallback(new MqttCallback() { // 其他回调方法... @Override public void reconnectComplete(boolean reconnect) { // 处理重新连接完成事件 // 添加自定义的处理逻辑 } }); 通过以上配置和逻辑，你可以实现MQTT开源Java SDK的客户端自动连接功能。当连接丢失时，客户端将自动尝试重新连接，并在重新连接完成后触发相应的事件回调方法。你可以根据需要添加自定义的处理逻辑来处理连接丢失和重新连接事件。

专属云Kafka的产品服务与公有云Kafka的产品规格保持一致，由两种服务资源组成，队列规格、队列存储。 其中队列计算规格按消息队列基准带宽分为4种分别是：100MB/s、300MB/s、600MB/s、1200MB/s； 存储类型分2种，分别如下：高I/O、超高I/O。

本文主要介绍日志收集系统Flume接入Kafka。 最佳实践概述 场景描述 使用Flume+Kafka来完成实时流式日志处理，后面再连接上Storm/Spark Streaming等流式实时处理技术，从而完成日志实时解析的目标。如果Flume直接对接实时计算框架，当数据采集速度大于数据处理速度，很容易发生数据堆积或者数据丢失，而kafka可以当做一个消息缓存队列，可以把它理解为一个数据库，可以存放一段时间的数据。 因此数据从数据源（ HTTP、Log 文件、JMS、监听端口数据等）到flume再到Kafka进行消息缓存，数据一方面可以同步到HDFS做离线计算，另一方面可以做实时计算，可实现数据多分发。 技术架构图 暂无。 方案优势 把数据存储到 HDFS 或者 HBase 等下游存储模块或者计算模块时需要考虑各种复杂的场景，例如并发写入的量以及系统承载压力、网络延迟等问题。Flume 作为灵活的分布式系统具有多种接口，同时提供可定制化的管道。在生产处理环节中，当生产与处理速度不一致时，Kafka 可以充当缓存角色。Kafka 拥有 partition 结构以及采用 append 追加数据，使 Kafka 具有优秀的吞吐能力；同时其拥有 replication 结构，使 Kafka 具有很高的容错性。所以将 Flume 和 Kafka 结合起来，可以满足生产环境中绝大多数要求。 前提条件 需已购买Kafka实例、创建Topic，并且已成功消费消息。 确认准备 Apache Flume环境（1.6.0以上版本兼容 Kafka）。 确认 Kafka 的 Source、 Sink 组件已经在 Flume 中。 资源规划 本实践方案内容仅涉及Kafka专享版实例。 分布式消息服务 Figure 1 分布式消息服务 资源类型 配置项 配置明细 说明 :::: 企业中间件 DMS Kafka专享实例 需已购买kafka专享实例，创建好Topic，并成功消费消息。 方案正文

消息可以批量生产和消费 为提高消息发送和消息消费效率，推荐使用批量消息发送和消费。通常，默认消息消费为批量消费，而消息发送尽可能采用批量发送。同时批量方式可有效减少API调用次数，减少服务使用费用。 如下面两张示意图对比所示，消息批量生产与消费，可以减少API调用次数，节约资源。 图 消息批量生产（发送）与消费 说明 批量发送消息时，单次不能超过10条消息，总大小不能超过512KB。 批量生产（发送）消息可以灵活使用，在消息并发多的时候，批量发送，并发少时，单条发送。这样能够在减少调用次数的同时保证消息发送的实时性。 图 消息逐条生产（发送）与消费 此外，批量消费消息时，消费者应按照接收的顺序对消息进行处理、确认，当对某一条消息处理失败时，不再需要继续处理本批消息中的后续消息，直接对已正确处理的消息进行确认即可。 巧用消费组协助运维 用户使用DMS服务作为消息管理系统，查看队列的消息内容对于定位问题与调试服务是至关重要的。 当消息的生产和消费过程中遇到疑难问题时，通过创建不同消费组可以帮助定位分析问题或调试服务对接。用户可以创建一个新的消费组，对队列中的消息进行消费并分析消费过程，这样不会影响其他服务对消息的处理。