云终端 云服务名称 区域 控制台 OpenAPI 系统策略 云服务名称 区域 支持IAM 支持企业项目 支持IAM 支持企业项目 系统策略 天翼云电脑（政企版） 全局 是 是 是 是 有 天翼云手机 全局 是 是 是 是 有 CDN与边缘 云服务类别 区域 控制台 OpenAPI 系统策略 云服务类别 区域 支持IAM 支持企业项目 支持IAM 支持企业项目 系统策略 边缘安全加速平台 全局 是 是 是 是 有 容器与中间件 云服务名称 区域 控制台 OpenAPI 系统策略 云服务名称 区域 支持IAM 支持企业项目 支持IAM 支持企业项目 系统策略 分布式消息服务Kafka 全局 是 是 是 是 有 分布式消息服务RabbitMQ 全局 是 是 是 是 有 分布式消息服务RocketMQ 全局 是 是 是 是 有 微服务云应用平台 全局 是 是 是 是 有 云容器引擎 全局 是 是 是 是 有 弹性容器实例 全局 是 是 是 是 有 函数计算 全局 是 是 是 是 有 云日志服务LTS 全局 是 是 是 是 有 容器镜像服务 全局 是 是 是 是 有 分布式消息服务MQTT 全局 是 是 是 是 有 应用服务网格 全局 是 是 是 是 有 微服务引擎 全局 是 是 是 是 有

本节介绍了分布式消息服务RabbitMQ产品常见操作类问题。 无法被路由的消息，去了哪里？ 如果没有任何设置，无法路由的消息会被直接丢弃。 无法路由的情况：Routing key不正确。 解决方案： 1.使用mandatorytrue配合ReturnListener，实现消息回发。 2.声明交换机时，指定备份交换机。 多个消费者监听一个队列时，消息如何分发？ 1.RoundRobin（轮询） 默认的策略，消费者轮流、平均地收到消息。 2.Fair dispatch（公平分发） 如果要实现根据消费者的处理能力来分发消息，给空闲地消费者发送更多消息，可以用basicQos(int prefetchcount)来设置。prefetchcount含义：当消费者有多条消息没有响应ACK时，不再给这个消费者发送消息。 消息在什么时候会变成Dead Letter(死信)？ 1.消息被拒绝并且没有设置重新入队：(NACK Reject ) && requeue false 2.消息过期(消息或者队列的TTL设置) 3.消息堆积，并且队列达到最大长度，先入队的消息编程DL。 解决方案：可以在声明队列时，指定一个Dead Letter Exchange，来实现Dead Letter的转发，保证消息不会丢失。 如何进行消息持久化？ 所谓持久化，就是RabbitMQ将内存中的数据（比如交换机、队列、消息等）固化到磁盘，以防止异常情况的发生时造成数据丢失。 持久化分类 说明 交换机持久化 在创建Exchange时设置durable参数参数。channel.exchangeDeclare(EXCHANGENAME, "direct", true); 队列持久化 同样也是设置设置durable参数。持久化的队列会存盘，在服务器重启的时候可以保证不丢失相关信息。channel.queueDeclare(QUEUENAME, true, false, false, null); 消息持久化 即使交换机、队列持久化不会因为重启丢失，但是存储在队列中的消息仍然会丢失。解决的办法就是设置消息的投递模式为2，即代表持久化(JAVA)。理论上，可以将所有的消息都设置为持久化，但是这会严重影响RabbitMQ性能，因为写入到磁盘的速度可比写入到内存的速度慢非常多。因此，在选择是否要持久化消息时，需要在可靠性和吞吐量之间做一个权衡。

本文主要介绍计费项。 计费说明 分布式消息服务Kafka的计费项由实例费用和存储空间费用组成。 表 分布式消息服务Kafka计费项 计费项 计费项说明 适用的计费模式 计费公式 实例费用 计费因子：代理规格和代理数量 包年/包月、按需计费 实例规格单价 购买时长 实例规格单价请参考产品价格。 存储空间费用 计费因子：云硬盘类型、容量 包年/包月、按需计费 存储规格单价 购买时长 存储规格单价请参考产品价格。 Smart Connect Smart Connect用于异构系统间的数据同步，开启Smart Connect后，设置Smart Connect任务可以实现Kafka实例和其他云服务（如OBS）之间的数据同步，或者两个Kafka实例之间的数据同步。 包年/包月、按需计费 开启Smart Connect后，实例会另外创建2个同规格、用于Smart Connect的代理，并收取相应的代理费用。例如：规格为kafka.4u8g.cluster的实例，会另外创建2个规格为kafka.4u8g的代理，用于Smart Connect，所以需要收取相应代理费用。 计费示例 计费场景 假如某用户于2023/03/18 15:30:00在“广州4”区域购买了一个按需计费的Kafka实例，规格配置如下： 规格：kafka.2u4g.cluster3 总存储空间：高I/O 300GB 用了一段时间后，用户发现Kafka实例当前规格无法满足业务需要，于2023/03/20 9:00:00升配为kafka.4u8g.cluster3。因为该Kafka实例打算长期使用下去，于2023/03/20 10:30:00将Kafka实例转为包年/包月计费，购买时长为1个月。那么在3~4月份，该Kafka实例总共产生多少费用呢？

生产消息 生产者需要创建一个连接到RabbitMQ服务器，然后创建一个通道（Channel）来进行消息的发布。在发布消息之前，生产者通常需要先声明一个队列，以确保消息能够被正确地路由和接收。 一旦连接和通道建立完成，生产者可以使用basicPublish()方法将消息发布到指定的队列。在发布消息时，需要指定目标队列的名称、消息内容以及其他的属性。 发布消息后，RabbitMQ将会将消息存储在队列中，等待消费者来接收。消费者可以使用相同的客户端库来创建连接和通道，并使用basicConsume()方法来订阅队列并接收消息。一旦有消息到达队列，消费者就会收到消息并进行相应的处理。 通过使用RabbitMQ，生产者和消费者可以实现解耦，即它们可以独立地进行开发和部署。生产者可以按照自己的节奏和需求发布消息，而消费者可以根据自己的处理能力和负载来接收和处理消息。 代码示例： import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class RabbitmqProducer { // private final static String EXCHANGENAME "YOUR EXCHANGE NAME"; private final static String QUEUENAME "YOUR QUEUE NAME"; // private final static String ROUTINGKEY "YOUR ROUTING KEY"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { // 创建连接工厂 ConnectionFactory factory new ConnectionFactory(); // 设置主机ip factory.setHost("YOUR HOST IP"); // 设置amqp的端口号 factory.setPort(YOUR PORT); // 设置用户名密码 factory.setUsername("YOUR USER NAME"); factory.setPassword("YOUR USER PASSWORD"); // 设置Vhost，需要在控制台先创建 factory.setVirtualHost("YOUR VHOST"); //基于网络环境合理设置超时时间 factory.setConnectionTimeout(30 1000); factory.setHandshakeTimeout(30 1000); factory.setShutdownTimeout(0); // 创建一个连接 Connection connection factory.newConnection(); // 创建一个频道 Channel channel connection.createChannel(); // 发送方消息确认，channel.confirmSelect(); // 启用发送方事务机制，channel.txSelect(); // 指定一个队列 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)); // 使用自定义交换器，需要在管理台预先建好，并设置routing key // channel.basicPublish(EXCHANGENAME, ROUTINGKEY, null, message.getBytes(StandardCharsets.UTF8)); System.out.println(" [x] Sent '" + message + "'"); TimeUnit.MILLISECONDS.sleep(100); } //关闭频道和连接 channel.close(); connection.close(); } } 消息发送后，可以进入控制台，在实例列表的队列选项卡查看消息发送状态。

类别 指标名称 主机 cpu使用率 主机 系统15分钟负载 主机 内存空闲率 主机 磁盘存储空闲率 主机 磁盘inode空闲率 主机 磁盘IO使用率 主机 网卡每秒接收丢包数 主机 网卡每秒接收错误包 主机 网卡每秒发送丢包数 主机 网卡每秒发送错误包数 主机 节点指标采集器可用状态 主机 主机运行时间 主机 磁盘挂载状态 主机 磁盘是否只读异常 主机 网卡每秒接收的比特数 主机 网卡每秒发送的比特数 HDFS HDFS进程状态 HDFS HDFS NameNode主备状态 HDFS HDFS rpc端口处理平均耗时 HDFS HDFS rpc端口请求队列长度 HDFS HDFS存储使用率 HDFS PendingDeletionBlocks HDFS MissingBlocks HDFS HDFS每秒创建文件数 HDFS hdfs锁队列等待线程数 HDFS HDFS距离上次checkpoint时间 HDFS HDFS NameNode内存使用率 HDFS HDFS DataNode节点存活比例 HDFS 存储使用总量 HDFS 数据块总量 HDFS HDFS存储空间使用百分比 HDFS HDFS存在损坏的磁盘 HDFS HDFS NameNode处于SafeMode HDFS HDFS NameNode Jvm暂停超过阈值的次数 HDFS HDFS DataNode Jvm暂停超过阈值的次数 YARN YARN进程状态 YARN YARN ResourceManager主从状态 YARN pending作业数 YARN pending container数 YARN 心跳处理时延 YARN 每秒心跳次数 YARN YARN NodeManager存活节点比例 YARN YARN ResourceManager内存使用率 YARN YARN ResourceManager主备状态 YARN YARN ResourceManager Container 分配的平均时间 Hive Hive进程状态 Hive hiveserver2 10000端口连接数 Hive hiveserver2 10000端口close wait数 Hive hiveserver2 端口监听状态 Hive hiveserver2堆内存使用率 Hive hiveserver2 10000端口连接数 Hive hiveserver2 10000端口close wait数 Hive Hive metastore 端口监听状态 Hive hiveserver2堆内存使用率 ZooKeeper ZooKeeper进程状态 ZooKeeper ZooKeeper watch数量 ZooKeeper ZooKeeper活跃连接数 ZooKeeper ZooKeeper数据节点存储总量 ZooKeeper ZooKeeper leader重新选主 ZooKeeper ZooKeeper的平均处理时间 Kafka Kafka进程状态 Kafka Kafka离线目录数量 Kafka Kafka离线副本数 Kafka Kafka低于 min.insync.replicas 的分区数量 Kafka Kafka未保持同步的分区数量 Kafka Kafka活跃controller数量 Kafka Kafka ZooKeeper请求延迟 Kafka Kafka broker访问zookeeper延迟时间 Kafka Kafka Broker请求队列积压的请求数量 Kafka Kafka 过去1分钟内触发的controller选举次数 HBase HBase 进程状态 HBase HBase master主从状态 HBase HBase regionserver rit数 HBase HBase master堆内存使用率 HBase HBase regionserver节点存活比例 HBase HBase regionserver堆内存使用率 HBase get操作平均耗时 HBase 99%的get操作耗时 HBase 请求平均处理耗时 HBase 每秒读请求数 HBase 每秒写请求数 HBase HBase正在使用的handler数量 HBase HBase等待执行的rpc队列长度 Doris Doris进程状态 Doris 每分钟GC时间 Doris 平均每分钟查询错误数 Doris be存储资源使用率 Doris Doris 前端主机cpu平均使用率 Doris Doris 前端主机内存平均使用率 Doris Doris 后端主机cpu平均使用率 Doris Doris 后端主机内存平均使用率 Doris Doris 前端每秒查询数量 Doris Doris 前端每秒请求数量 Doris Doris INSERT 加载中作业峰值 Doris Doris load任务失败率 Doris Doris集群最大compaction score Doris Doris集群FE的Minor GC持续时长 Doris Doris集群FE的Old GC持续时长 Doris Doris集群BE CPU使用率 Doris Doris DB当前事务数 Elasticsearch ES进程状态 Elasticsearch ES集群状态 Elasticsearch ES 未分配shard数 Elasticsearch pending任务数量 Elasticsearch 文件系统剩余容量百分比 Elasticsearch JVM堆使用率 Elasticsearch rejected线程数 Trino Trino进程状态 Trino 健康节点比例 Trino JVM堆使用率 OpenLDAP openldap进程状态 OpenLDAP 389端口连接数 Flink Flink进程状态 Flume Flume进程状态 Spark Spark进程状态 Kerberos Kerberos进程状态 Ranger Ranger进程状态 Ranger Ranger JVM堆使用率 Kyuubi Kyuubi进程状态 Kyuubi Kyuubi JVM Deadlock线程数 Kyuubi Kyuubi 服务堆内存使用率 Kyuubi Kyuubi 服务存活率 Knox Knox进程状态 Kibana Kibana进程状态 Logstash Logstash进程状态 JeekeFS JeekeFS进程状态 KafkaUI KafkaUI进程状态 Amoro Amoro 进程状态 Tezui TezUI 进程状态 Tezui TezUI 端口监听状态 Hue Hue 进程状态 Hue Hue 端口监听状态 Hue Hue Python第0代不可回收对象的5分钟增量 Hue Hue Python第1代不可回收对象的5分钟增量 Hue Hue Python第2代不可回收对象的5分钟增量 Hue Hue Django 500错误响应的5分钟增量 DolphinScheduler DolphinScheduler 进程状态 DolphinScheduler DolphinScheduler 系统当前的CPU使用率 DolphinScheduler DolphinScheduler 系统的平均负荷（1分钟） DolphinScheduler DolphinScheduler 各JVM进程CPU使用率 DolphinScheduler DolphinScheduler 各服务GC后长期存活堆内存的使用率 DolphinScheduler DolphinScheduler 各服务垃圾回收时间占比 DolphinScheduler DSApiServer 未捕获异常个数 DolphinScheduler DSAlertServer 等待发送的告警数量 DolphinScheduler DSApiServer 响应状态码的接口个数 DolphinScheduler DSMasterServer 1小时过载次数 DolphinScheduler DolphinScheduler 正在运行的工作流实例数量 DolphinScheduler DSWorkerServer 1小时过载次数 DolphinScheduler DSWorkerServer 上等待提交的任务总数 DolphinScheduler DSWorkerServer 上正在执行的任务总数 DolphinScheduler DSWorkerServer 1小时提交队列全满次数 DolphinScheduler DSAlertServer 5分钟未捕获异常个数

本文主要介绍云日志服务的投递概述。 天翼云云日志服务可以将日志单元中的数据投递到天翼云对象存储ZOS与分布式消息服务kafka，以满足归档或者计算的需求。 转储至对象存储ZOS 您将云主机和云服务的日志数据上报至云日志服务后，可对日志存储时长进行设置，默认存储时间为365天，超出存储时间后，日志数据将会被自动清理。若您需要长期存储日志数据进行日志持久化，可使用日志转储功能，将日志数据转储至对象存储中进行长期保存。具体操作请查看转储至对象存储ZOS。 说明 日志转储的实时性较弱，通常为5分钟~30分钟。数据延迟依赖于存储系统。 关于对象存储更多内容，请参考对象存储产品文档。 转储至分布式消息服务kafka 云日志服务采集到数据后，支持将数据转储至对天翼云分布式消息服务kafka，用于实时流计算等场景。具体操作请查看转储至分布式消息服务kafka。

支持将分布式消息服务Kafka的重平衡日志接入LTS。重平衡日志记录了Rebalance的详情，包括Rebalance时间、原因和触发Rebalance的客户端等。重平衡日志存储在云日志服务中，由LTS提供查询功能。 接入操作说明 1. 登录分布式消息服务Kafka管理控制台。 2. 在实例列表中，点击目标Kafka实例名称。 3. 点击左侧菜单栏“智能运维”重平衡日志。点击【立即开启】按钮，即可将Kafka日志接入至云日志服务的grouprebalancelog日志项目中。 4. 等待若干分钟后，系统将自动生成可视化仪表大盘。 5. 点击页面上方“日志查询”页签，即可查询详细的重平衡日志信息。 6. 您也可以登录云日志服务控制台，在“grouprebalancelog”日志项目中，可查看所有已接入日志服务的Kafka重平衡日志的日志单元，并进行查询、分析或告警配置等操作。

本文主要介绍消息堆积对业务的影响及解决办法。 消息堆积对业务的影响 过量的消息堆积可能会引起内存或磁盘告警，从而造成所有connection阻塞，进而影响到其他队列的使用，导致整体服务质量的下降。 消息堆积产生的原因 1. 一般来说消息堆积是由于生产消息的速率远大于消费消息的速率所导致的。比如某个时间段消费端处理消息异常缓慢，发送一条消息只要3秒钟，而消费一条消息需要1分钟，每分钟发送20个消息，只有一个消息被消费端处理，这样队列中就会产生大量的消息堆积。 2. 消费者出现异常，生产者一直在发送消息，但是消费者不能消费，造成消息积压。 3. 消费者没有出现异常，但是消费者与队列间的订阅可能出现了异常，也会导致消息无法被消费从而造成堆积的情况。 4. 消费者正常，与队列间的订阅也正常，但是消费端的代码本身逻辑耗费时间长导致了消费能力降低，这时候就会出现1中的情况从而导致消息堆积。 解决消息堆积的办法 1. 生产速率较快，消费速率较慢 ：您可以通过以下方法解决。 增加消费者数量提高消费速率。 采用生产者确认的发送模式，并监控生产端消息生产速度和时长，当消息生产时长有明显增加时进行流控措施。 2. 消费者异常 ：建议排查消费者逻辑是不是有问题，优化程序。 3. 消费者与队列间的订阅异常 ：建议排查队列和消费者之间的订阅是否正常。 4. 消费端的代码本身逻辑耗费时间长 ：建议给消息设置过期时间，设置方法如下： 在生产消息时，设置消息过期时间。消息过期时间以expiration值体现。 在properties中设置expiration值，单位为毫秒(ms)。 AMQP.BasicProperties properties new AMQP.BasicProperties().builder() .deliveryMode(2) .contentEncoding("UTF8") .expiration("10000") .build(); String message "hello rabbitmq"; channel.basicPublish(exchange, routingKey, properties, message.getBytes(StandardCharsets.UTF8)); 在Web界面中设置expiration值，单位为毫秒(ms)。 登录Web界面，在“Exchanges”页签，单击Exchange名称，进入Exchange详情页。在“Publish message”区域，设置expiration值，如下图所示。 设置队列过期时间。队列过期时间以xmessagettl值体现。从消息进入队列开始计算，超过了配置的队列过期时间，消息会自动被删除。 在客户端代码中设置xmessagettl值，单位为毫秒(ms)。 Map arguments new HashMap (); arguments.put("xmessagettl", 10000); channel.queueDeclare(queueName, true, false, false, arguments); 在Web界面新建队列时，设置xmessagettl值，单位为毫秒(ms)。 登录Web界面，在“Exchanges”页签，新建队列时，设置xmessagettl值，如下图所示。

本节介绍分布式消息服务Kafka连接常见问题 接入地址出现不足三个ip+端口？ 集群三台机器正常运作的情况下，接入点会出现三个ip:port连起来，当出现不足三个时候，说明其中一台机器不正常工作（没出现在接入点的机器），请尽快联系管理人员查看不正常工作的节点，尽快恢复。 实例是否支持跨VPC访问？ 实例支持跨VPC访问，可以通过创建VPC对等连接，将两个VPC的网络打通，实现跨VPC访问实例。 如何通过代理访问Kafka实例 暂不支持通过代理访问Kafka实例。 原因：客户端通过bootstrap.servers随机访问服务器获取元数据，通过元数据里的IP地址进行后续的消息收发。配置代理后，第一步正常，但在第二步时，客户端仍然通过元数据中的IP地址进行访问，导致访问不通。因此，必须保证网络全连通。 生产者会建立多少个连接到Broker？ 每个生产者通常会建立2个到Broker的TCP连接，一个TCP连接用于更新元数据，一个TCP连接用于发送消息。 VPC实例和公网/VPC实例的区别是什么？ VPC实例：仅提供VPC接入点，只能通过VPC访问。 公网/VPC实例：提供公网接入点和VPC接入点，可以通过公网访问，也可以通过VPC访问。 Kafka实例是否支持公网访问？ Kafka实例支持公网访问。使用公网访问Kafka实例的具体操作，请参考设置公网ip。

Kafka实例规格参考 kafka.2u4g.cluster，三个代理 Kafka客户端连接数在3000以内，消费组个数在60个以内，业务TPS为100000以内时推荐选用。 kafka.4u8g.cluster，三个代理 Kafka客户端连接数在10000以内，消费组个数在300个以内，业务TPS为300000以内时推荐选用。 kafka.8u16g.cluster，三个代理 Kafka客户端连接数在20000以内，消费组个数在600个以内，业务TPS为600000以内时推荐选用。 kafka.12u24g.cluster，三个代理 Kafka客户端连接数在20000以内，消费组个数在600个以内，业务TPS为900000以内时推荐选用。 kafka.16u32g.cluster，三个代理 Kafka客户端连接数在20000以内，消费组个数在600个以内，业务TPS为1200000以内时推荐选用。 Kafka实例的存储空间估算参考 Kafka实例支持多副本存储，存储空间包含所有副本存储空间总和，您在创建Kafka实例，选择初始存储空间时，建议根据业务消息体积预估以及副本数量选择合适的存储空间。 例如：业务消息体积预估100GB，则磁盘容量最少应为100GB副本数 + 预留磁盘大小100GB。 Kafka实例支持对存储进行扩容，根据业务增长，随时扩容，节约成本。 Kafka实例Topic数量计算 Kafka实例对Topic分区数之和设置了上限，当达到上限之后，用户无法继续创建Topic。 所以，Topic数量和实例分区数上限、每个Topic的分区数有关，其中，每个Topic分区数可在创建Topic时设置，如下图，实例分区数上限参考上表。 图Topic的分区数 kafka.2u4g.cluster 3 broker实例的分区数上限为750。 如果该实例下每个Topic的分区个数都为3，则Topic个数为750/3250个。 如果该实例下每个Topic的分区个数都为1，则Topic个数为750/1750个。

在开通了专属云（计算独享型）基础上，再购买专属云Kafka，对产品各服务规格收取服务费。具体价格如下： 专属云Kafka中，队列实例各规格价格： 规格 标准资费(元/月) 100 MB/s 650 300 MB/s 1050 600 MB/s 2002.5 1200 MB/s 3635 对于专属云Kafka的存储部分，有2种可选。 1）选择公有云中云硬盘来承载存储，则此部分存储价格按公有云收费标准收取； 2）购买专属云（存储独享型）后，以专属存储承载Kafka存储，则收费服务费，价格如下： 专属存储部分服务费 存储类型 标准资费（元/GB/月） 高IO 0.3 超高IO 0.75

本文主要介绍 Kafka Manager问题。 登录Kafka Manager的帐号是否可以设置为只读帐号？ 不支持设置为只读帐号。 登录到Kafka Manager页面，为什么获取不到节点信息？ 问题现象： 正常登录Kafka Manager页面，但是页面无法显示节点信息，如下所示： 问题原因： 这是Kafka开源问题，需要联系客服，重启Kafka Manager。 Yikes! Insufficient partition balance when creating topic : projectmanprojectenterpriseproject Try again. 问题现象： 在Kafka Manager中无法创建topic，报错：Yikes! Insufficient partition balance when creating topic : projectmanprojectenterpriseproject Try again. 问题原因： Topic分区超过限制，不能继续创建Topic。 处理方法： 建议扩大实例规格，实例规格增加，分区数也会相应增加。 Kafka Manager能否查询到消息的正文？ Kafka Manager无法查询到消息的正文。 Kafka Manager WebUI的端口能否修改？ Kafka Manager WebUI的端口不支持修改。 在Kafka Manager上支持修改Topic的哪些属性？ 在Kafka Manager上，支持修改如下的Topic属性：max.message.bytes、segment.index.bytes、segment.jitter.ms、min.cleanable.dirty.ratio、retention.bytes、file.delete.delay.ms、compression.type、flush.ms、cleanup.policy、unclean.leader.election.enable、flush.messages、retention.ms、min.insync.replicas、delete.retention.ms、preallocate、index.interval.bytes、segment.bytes、segment.ms。 具体修改方法如下： 1. 登录Kafka Manager。 2. 单击“kafkacluster”，进入“Summary”界面。 3. 单击“Topic > List”，进入“Topics”界面。 4. 单击需要修改属性的Topic名称，进入Topic详情界面。 5. 单击“Update Config”，修改Topic属性。

本文主要介绍如何通过kafka导入日志。 本文主要介绍如何将天翼云分布式消息服务Kafka的数据导入到云日志服务，实现数据的查询分析、加工等操作。 注意 当前功能为试用阶段，仅在华北2资源池开放。 创建数据导入任务 1. 登录云日志服务控制台。 2. 左侧菜单栏点击“日志接入”，进入接入管理页面 3. 在“数据导入”模块中，点击“分布式消息服务Kafka数据导入” 4. 选择目标日志项目和日志单元，单击下一步。 5. 设置接入配置，配置如下参数，确认无误后，单击下一步。 参数 说明 接入配置名称 导入任务的唯一标识 Kafka 实例 选择kafka实例作为数据源 Topic 列表 选择Kafka主题，当前仅支持导入公有Topic 消费组 选择kafka中的消费组， 起始位置 选择开始导入数据的位置。 最早：从已有的第一条Kafka数据开始导入 最晚：从最近生成的Kafka数据开始导入 数据格式 待导入数据的格式。若您选择JSON字符串，导入任务会将原始数据解析为keyvalue对格式，只解析到第一层。 编码格式 选择导入数据的编码格式（即字符集），目前仅支持UTF8和GBK。 6. 创建索引。默认开启全文索引，您也可以根据需要手动创建字段索引用于字段查询。 7. 点击完成，即可完成导入任务创建。等待1分钟左右，在查询日志界面能查询到日志，则说明导入成功。

本文介绍Kafka消费端从服务端拉取不到消息或拉取消息缓慢原因及解决方案 问题现象 Topic中有消息并且Consumer未消费到最新的位置，出现消费端从服务端拉取不到消息或拉取消息缓慢的情况（特别是公网消费时）。 可能原因 消费流量达到网络带宽。 单个消息大小超过网络带宽。 Consumer每次拉取的消息量超过网络带宽。 说明 Consumer每次消息的拉取量受以下参数影响： max.poll.records：每次拉取的最多消息数。 fetch.max.bytes：每次拉取的最大总byte数。 max.partition.fetch.bytes：每个Partition每次拉取的最大总byte数。 解决方案 （1）登录分布式消息服务Kafka控制台查询消息。 如果能查询到消息，请继续尝试以下步骤。 （2）在实例详情页面，单击左侧导航栏的监控信息，查看消费流量是否已达到网络带宽。 如果消费流量已经达到网络带宽，您需要扩充网络带宽。 （3）检查Topic中是否存在单个消息的大小超过网络带宽。 如果存在单个消息的大小超过网络带宽，请提高网络带宽，或者减小单个消息的大小。 （4）检查Consumer每次拉取的消息量是否超过网络带宽。 说明 如果每次拉取的消息量超过网络带宽，您需要调整以下参数。 网络带宽>fetch.max.bytes 网络带宽>max.partition.fetch.bytes总订阅Partition数

指标ID 指标名称 指标含义 取值范围 测量对象 监控周期（原始指标） topicproducemsg 消息生产数 Topic一分钟收到的消息数单位：Count >0 RocketMQ实例队列 1分钟 topicconsumemsg 消息消费数 Topic一分钟被消费的消息数单位：Count >0 RocketMQ实例队列 1分钟 topicproducerate 消息生产速率 Topic每秒收到的消息数单位：Count/s >0 RocketMQ实例队列 1分钟 topicconsumerate 消息消费速率 Topic每秒被消费的消息数单位：Count/s >0 RocketMQ实例队列 1分钟 topicbytesinrate 生产流量 当前主题的生产流量单位：Byte/s说明：2022年5月16号及以后购买的实例，支持此监控项。 >0 RocketMQ实例队列 1分钟 topicbytesoutrate 消费流量 当前主题的消费流量单位：Byte/s说明：2022年5月16号及以后购买的实例，支持此监控项。 >0 RocketMQ实例队列 1分钟

本文主要介绍实例规格变更问题。 Kafka扩容/缩容会影响业务吗？ Kafka扩容带宽/代理个数/存储空间，都不会影响业务的使用。 Kafka扩容代理规格，在扩容过程中，节点会重启，可能造成闪断。若Topic为单副本，扩容期间无法对该Topic生产消息或消费消息。 表 变更实例规格的影响 变更配置类型 影响 基准带宽/代理数量 扩容基准带宽/代理数量不会影响原来的代理，业务也不受影响。 扩容基准带宽/代理数量时，系统会根据当前磁盘大小进行相应比例的存储空间扩容。例如扩容前实例的代理数为3，每个代理的磁盘大小为200GB，扩容后实例的代理数为10，此时代理的磁盘大小依旧为200GB，但是总磁盘大小为2000GB。 新创建的Topic才会分布在新代理上，原有Topic还分布在原有代理上，造成分区分布不均匀。通过修改Kafka分区平衡，实现将原有Topic分区的副本迁移到新代理上。 存储空间 扩容存储空间有次数限制，只能扩容20次。 扩容存储空间不会影响业务。 代理规格 若Topic为单副本，扩容/缩容期间无法对该Topic生产消息或消费消息，会造成业务中断。 若Topic为多副本，扩容/缩容代理规格不会造成服务中断，但会逐个节点重启，负载会转移到剩余节点上，建议您在业务低峰期扩容/缩容。 扩容/缩容代理规格的过程中，节点滚动重启造成分区Leader切换，会发生秒级连接闪断，在用户网络环境稳定的前提下，Leader切换时长一般为1分钟以内。多副本的Topic需要在生产客户端配置重试机制，方法如下： a. 生产客户端为Kafka开源客户端时，检查是否配置retries和retry.backoff.ms参数。建议参数值分别配置为：retries10，retry.backoff.ms1000。 b. 生产客户端为Flink客户端时，检查是否配置重启策略，配置重启策略可以参考如下代码。 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(20))); 若实例已创建的分区数总和大于待缩容规格的实例分区数上限，此时无法缩容。不同规格配置的实例分区数上限不同，具体请参见产品规格。例如：kafka.4u8g.cluster3的实例，已创建800个分区，您想把此实例的规格缩容为kafka.2u4g.cluster3，kafka.2u4g.cluster3规格的实例分区数上限为750，此时无法缩容。 变更代理规格的时长和代理数量有关，单个代理的变更时长一般在5~10分钟，代理数量越多，规格变更时间越长。

参数 说明 触发事件类型 选择触发作业运行的事件类型。 “KAFKA” “KAFKA”触发事件类型的参数 连接名称 选择数据连接，需先在“管理中心”创建kafka数据连接。 Topic 选择需要发往kafka的消息Topic。 事件处理并发数 选择作业并行处理的数量，最大并发数为128。 事件检测间隔 配置时间间隔，检测通道下是否有新的消息。时间间隔单位可以配置为秒或分钟。 读取策略 选择数据的读取位置： 从上次位置读取：首次启动时，从最新的位置读取数据。后续启动时，则从前一次记录的位置读取数据。 从最新位置读取：每次启动都会从最新的位置读取数据。 失败策略 选择调度失败后的策略： 挂起 忽略失败，读取下一个

系统解耦 以电商秒杀、抢购等流量短时间内暴增场景为例，传统做法是，用户下单后，订单系统发送查询请求到库存系统，等待库存系统返回请求结果给订单系统。如果库存系统发生故障，订单系统获取不到数据，订单失败。这种情况下，订单系统和库存系统是两个高耦合的子系统。 图4 系统高耦合 引入RabbitMQ消息队列，当用户下单后，将消息写入到RabbitMQ消息队列中，然后返回用户下单成功。 库存系统订阅下单的消息，消费下单消息，然后进行库操作。即使库存系统出现故障，也不影响用户下单。 图5 系统解耦 高可用 普通队列，由于队列以及队列内容仅存储在单代理上，当该代理故障后，对应的队列不可用。 RabbitMQ引入镜像队列机制，镜像队列是开源RabbitMQ 2.6.0版本新增的一个功能，允许集群将队列镜像到其他代理上，当集群某一代理宕机后，队列能自动切换到镜像中的其他代理，保证服务的可用性。 RabbitMQ引入仲裁队列机制，仲裁队列是开源RabbitMQ 3.8版本新增的一个功能，提供队列复制的能力，当集群某一代理宕机后，队列依旧可以正常运行，保证服务的可用性。

本节介绍取消Kafka消费组和Topic的订阅关系。 消费者订阅Topic消费消息后，在消费组详情页可以查看消费进度。如果消费者不再消费某个Topic，您可以取消消费组和此Topic的订阅关系，避免由于此Topic的消息堆积，导致的告警误报。 约束与限制 已创建的1.1.0版本的Kafka实例可能不支持取消订阅功能，您可以通过升级实例的内核版本来解决此问题。 取消消费组和Topic的订阅关系会删除消费组对该Topic的消费位点信息，删除后无法恢复，请谨慎操作。 取消Kafka消费组和Topic的订阅关系 1. 登录Kafka控制台。 2. 在管理控制台左上角单击，选择Kafka实例所在的区域。 3. 单击Kafka实例的名称，进入实例详情页面。 4. 在左侧导航栏选择“实例管理 > 消费组管理”，进入消费组列表页面。 5. 单击待取消订阅关系的消费组名称，进入消费组详情页。 6. 在“消费进度”页签中，勾选待取消订阅关系的Topic，每次最多勾选50个，单击信息栏左上侧的“取消订阅”。 7. 在“取消订阅”对话框中，单击“确定”。 取消订阅成功后，在“消费进度”页签，不再显示取消订阅的Topic。

介绍分布式消息服务Kafka用户列表功能操作内容。 场景描述 Kafka用户管理的场景描述如下： 访问控制：Kafka用户管理可以用于实现访问控制，限制对Kafka集群和主题的访问权限。通过创建不同的用户和角色，并为其分配合适的权限，可以确保只有经过授权的用户能够访问和操作Kafka集群。 安全认证：Kafka用户管理可以用于实现安全认证，确保只有合法用户能够登录和使用Kafka集群。通过设置用户名和密码，以及使用TLS/SSL证书等安全机制，可以保护Kafka集群免受未经授权的访问和攻击。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“用户管理”后即可查看所有的用户信息。 （5）右上角输入用户名称，可查询对应用户。

继7月价格下调后，天翼云分布式消息服务Kafka收到许多新老用户的积极反馈，为进一步回馈用户支持，我们决定增加包年折扣优惠和进行第二轮降价！ 9月25日起订购分布式消息服务Kafka产品将享受1年83折 优惠，1011月将陆续开放2年7折，3年5折优惠选项，具体开放时间敬请关注产品公告。 针对现网主推的Intel、海光、鲲鹏计算增强型规格产品单价进行全面下调，相对7月价格最高降幅可达30%，详细降价情况如下： 注意 通用型规格已经调整为白名单特性，如需了解相关资费请联系技术支持。 本次包年优惠及降价适用于产品新资费范围，具体支持资源池请参阅 主机类型 规格名称 计费单位 实例包月标准价格 降幅 主机类型 规格名称 计费单位 20240925前价格 新价格 降幅 Intel计算增强型 kafka.2u4g.cluster 元/代理 630 441 30% Intel计算增强型 kafka.4u8g.cluster 元/代理 1260 1008 20% Intel计算增强型 kafka.8u16g.cluster 元/代理 2430 2187 10% Intel计算增强型 kafka.12u24g.cluster 元/代理 3690 3321 10% Intel计算增强型 kafka.16u32g.cluster 元/代理 4500 4050 10% Intel计算增强型 kafka.24u48g.cluster 元/代理 7560 6804 10% Intel计算增强型 kafka.32u64g.cluster 元/代理 10080 9072 10% Intel计算增强型 kafka.48u96g.cluster 元/代理 15120 13608 10% Intel计算增强型 kafka.64u128g.cluster 元/代理 20160 18144 10% 海光计算增强型 kafka.hg.2u4g.cluster 元/代理 781.2 546.84 30% 海光计算增强型 kafka.hg.4u8g.cluster 元/代理 1562.4 1249.92 20% 海光计算增强型 kafka.hg.8u16g.cluster 元/代理 3013.2 2410.56 20% 海光计算增强型 kafka.hg.16u32g.cluster 元/代理 5580 5022 10% 海光计算增强型 kafka.hg.32u64g.cluster 元/代理 12499.2 11249.28 10% 鲲鹏计算增强型 kafka.kp.2u4g.cluster 元/代理 894.6 626.22 30% 鲲鹏计算增强型 kafka.kp.4u8g.cluster 元/代理 1789.2 1431.36 20% 鲲鹏计算增强型 kafka.kp.8u16g.cluster 元/代理 3450.6 3105.54 10% 鲲鹏计算增强型 kafka.kp.16u32g.cluster 元/代理 6390 5751 10% 鲲鹏计算增强型 kafka.kp.32u64g.cluster 元/代理 14313.6 12882.24 10%

本文介绍事件源中的请求参数。 请求参数 参数 类型 是否必选 示例值 描述 eventBusName String 是 test 事件总线的名称。 eventSourceName String 是 source1 事件源的名称。 desc String 否 test 事件源描述。 type String 是 simple 事件源类型，更多信息请参考下文关于事件源的格式。 eventSourceParams Arrays of Object 是 事件源参数，详见表 EventSourceParams，更多信息请参考下文关于事件源的格式。 表 EventSourceParams 参数 是否必填 参数类型 说明 resourceKey 是 String 事件源参数名。 value 是 String 事件源参数值。 form 是 String 事件源参数格式。 template 否 String 事件源参数模板样式。 分布式消息服务Kafka 当事件源的事件提供方选择分布式消息服务Kafka时，请求参数中的type为kafka，eventSourceParams 中的resourceKey 字段含义如下表所示： resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务Kafka实例ID。 无 topic 是 CONSTANT 主题名称。 无 groupName 是 CONSTANT 消费组名。 无 initialOffset 是 CONSTANT 消息位点。 最新位点：latest 最早位点：earliest 无 dataFormat 否 CONSTANT 数据格式,不填时默认为JSON编码格式： JSON格式编码：JSON 文本格式编码：Text 二进制格式编码：Binary 无 请求示例如下： plaintext { "eventSourceName": "source", "eventBusName": "test", "desc": "dddesc", "type": "kafka", "eventSourceParams": [ { "resourceKey": "instanceId", "value": "b16b04cf0776xxxxxfa31a001a5c48b6", "form": "constant" }, { "resourceKey": "topic", "value": "ebsource", "form": "constant" }, { "resourceKey": "groupName", "value": "ebgroup", "form": "constant" }, { "resourceKey": "initialOffset", "value": "latest", "form": "constant" }, { "resourceKey": "dataFormat", "value": "JSON", "form": "CONSTANT" } ] }

场景 弹性存储集群 普通集群 业务数据量30M/s左右 3副本主题，生产消费1:1，分区100个，消息保留24小时（约2531GB业务数据） 规格： kafka.2u4g.cluster，单节点超高IO 200GB，节点数 3 配置： 开启弹性存储，分片大小128M，消息本地保留时长为1小时，消息保留24小时 性能： 单节点带宽峰值100MB/s（性能瓶颈为节点带宽） 成本： 规格费用：1323元/月 本地存储费用：720元/月 远程存储费用：300元/月 规格： kafka.2u4g.cluster，单节点超高IO 3300GB，节点数 3 配置： 未开启弹性存储，消息保留24小时 性能： 单节点带宽峰值100MB/s（性能瓶颈为节点带宽） 成本： 规格费用：1323元/月 存储费用：11880 元/月

本节主要介绍分布式消息服务Kafka的计费项 分布式消息服务Kafka的计费项由实例费用和存储费用组成，具体费用详情可参考产品资费。 计费项 计费项说明 使用的计费模式 计费公式 实例费用 计费因子：代理规格和代理数量 包年/包月、按需计费 实例规格单价 代理数量 购买时长 存储空间费用 计费因子：云硬盘类型、容量大小 包年/包月、按需计费 云硬盘规格单价 单个代理存储大小 代理数量 购买时长

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

同步方向 数据流向 规格选择 升级规格 入云 MySQL>MySQL 支持 仅支持单AZ任务升级 入云 MySQL>PostgreSQL 支持 仅支持单AZ任务升级 入云 MySQL>GaussDB分布式版 支持 仅支持单AZ任务升级 入云 MySQL>GaussDB集中式版 支持 仅支持单AZ任务升级 入云 MySQL>TaurusDB 支持 仅支持单AZ任务升级 入云 PostgreSQL>PostgreSQL 支持 仅支持单AZ任务升级 入云 DRDS>MySQL 支持 不支持 入云 DRDS>DRDS 支持 不支持 入云 Oracle>PostgreSQL 支持 仅支持单AZ任务升级 入云 Oracle>MySQL 支持 仅支持单AZ任务升级 入云 Oracle>TaurusDB 支持 仅支持单AZ任务升级 入云 Oracle>GaussDB集中式版 支持 仅支持单AZ任务升级 入云 Oracle>GaussDB分布式版 支持 仅支持单AZ任务升级 入云 Oracle>DRDS 支持 仅支持单AZ任务升级 入云 Microsoft SQL Server>Microsoft SQL Server 不支持 不支持 入云 MongoDB>DDS 支持 仅支持单AZ任务升级 入云 MariaDB>MariaDB 支持 仅支持单AZ任务升级 入云 TaurusDB>TaurusDB 支持 仅支持单AZ任务升级 出云 MySQL>MySQL 支持 仅支持单AZ任务升级 出云 MySQL>kafka 支持 仅支持单AZ任务升级 出云 MySQL>CSS/ES 支持 仅支持单AZ任务升级 出云 MySQL>Oracle 支持 仅支持单AZ任务升级 出云 DRDS>MySQL 支持 不支持 出云 DRDS>Oracle 支持 不支持 出云 DRDS>Kafka 支持 不支持 出云 DDS>MongoDB 支持 不支持 出云 DDS>Kafka 支持 不支持 出云 PostgreSQL>Kafka 支持 仅支持单AZ任务升级 出云 GaussDB集中式版>MySQL 支持 仅支持单AZ任务升级 出云 GaussDB集中式版>Oracle 支持 仅支持单AZ任务升级 出云 GaussDB集中式版>Kafka 支持 仅支持单AZ任务升级 出云 GaussDB集中式版>GaussDB分布式版 支持 仅支持单AZ任务升级 出云 GaussDB集中式版>GaussDB集中式版 支持 仅支持单AZ任务升级 出云 GaussDB分布式版>MySQL 支持 不支持 出云 GaussDB分布式版>Oracle 支持 不支持 出云 GaussDB分布式版>Kafka 支持 不支持 出云 GaussDB分布式版>GaussDB分布式版 支持 不支持 出云 TaurusDB>MySQL 支持 仅支持单AZ任务升级 出云 TaurusDB>Kafka 支持 仅支持单AZ任务升级 出云 TaurusDB>CSS/ES 支持 仅支持单AZ任务升级 出云 TaurusDB>Oracle 支持 仅支持单AZ任务升级 自建自建 MySQL>Kafka 支持 仅支持单AZ任务升级 自建自建 MySQL>CSS/ES 支持 仅支持单AZ任务升级 自建自建 MySQL>GaussDB集中式版 支持 仅支持单AZ任务升级 自建自建 MySQL>GaussDB分布式版 支持 仅支持单AZ任务升级 自建自建 Oracle>Kafka 支持 仅支持单AZ任务升级 自建自建 PostgreSQL>Kafka 支持 仅支持单AZ任务升级 自建自建 GaussDB集中式版>MySQL 支持 仅支持单AZ任务升级 自建自建 GaussDB集中式版>Oracle 支持 仅支持单AZ任务升级 自建自建 GaussDB集中式版>Kafka 支持 仅支持单AZ任务升级 自建自建 GaussDB分布式版>Kafka 支持 不支持

本文介绍事件流中事件源的元数据参数。 请求参数 参数 类型 是否必选 描述 eventStreamName String 是 事件流名。 streamSource Object 是 事件源项。详见表StreamSource，更多信息请参考下文事件源的格式。 desc String 否 事件源描述。 filterPattern String 是 事件过滤项。 transform Object 是 事件转换项。 streamSink Object 是 事件目标项。详见事件目标参数。 表 StreamSource 参数 类型 是否必选 描述 type String 是 事件源类型。 params Array of Objects 是 事件源参数，详见表Params。 表 Params 参数 是否必填 参数类型 说明 resourceKey 是 String 事件源参数名。 value 是 String 事件源参数值。 form 是 String 事件源参数格式。 template 否 String 事件源参数模板样式。 分布式消息服务Kafka 当事件源的事件提供方选择分布式消息服务Kafka时，请求参数中的type为kafka，params 中的resourceKey 字段含义如下表所示： resourceKey 是否必传 form value template instanceId 是 CONSTANT 分布式消息服务Kafka实例ID。 无 topic 是 CONSTANT 主题名称。 无 groupName 是 CONSTANT 消费组名。 无 initialOffset 是 CONSTANT 消息位点。 最新位点：latest 最早位点：earliest 无 dataFormat 否 CONSTANT 数据格式,不填时默认为JSON编码格式： JSON格式编码：JSON 文本格式编码：Text 二进制格式编码：Binary 无 事件源部分的参数示例如下： plaintext { "type": "kafka", "params": [ { "resourceKey": "instanceId", "value": "b16b04cf0776xxxxxfa31a001a5c48b6", "form": "constant" }, { "resourceKey": "topic", "value": "ebsource", "form": "constant" }, { "resourceKey": "groupName", "value": "ebgroup", "form": "constant" }, { "resourceKey": "initialOffset", "value": "latest", "form": "constant" }, { "resourceKey": "dataFormat", "value": "JSON", "form": "CONSTANT" } ] }

本文主要介绍分布式消息服务RabbitMQ的队列迁移最佳实践。 在RabbitMQ集群上，队列在各个节点分布不均衡会导致部分节点压力过大，无法更有效的利用集群。这可能是扩容节点、删除队列等原因导致的。 设置队列负载均衡的方法如下： 删除队列重建 通过Policy修改master节点方式 删除队列重建 1. 登录RabbitMQ WebUI页面。 2. 在“Overview”页签中，单击“Download broker definitions”，导出元数据。 3. 停止生产，等待数据消费完，然后删除原有队列。 在“Overview”页签中，确认数据是否已消费完。 可消费消息数（Ready）和未确认的消息数（Unacked）都为0时，说明消费完成。 等数据消费完后，删除原有队列。 在“Queues”页签，单击需要删除的队列名称，进入队列详情页面。 单机“Delete Queue”，删除队列。 4. 在“Overview”页签中，上传2中导出的元数据。 在“Overview”页签中，单击“选择文件”，选择2中导出的元数据。 单击“Upload broker definitions”，上传元数据。 上传成功后，显示如下信息。 实例会自动将队列均衡创建在各个节点上，在“Queues”页签中查看队列分布详情。

验证数据是否同步 1、 在目标集群中查看Topic列表，确认是否有源集群Topic。 说明 目标集群中的Topic名称和源集群相比，多了前缀（如A.），这属于正常情况，是MirrorMaker 2为了防止Topic循环备份进行的设置。 2、 在源集群生产并消费消息，在目标集群查看消费进度，确认数据是否已从源集群同步到了目标集群。 如果目标集群为Kafka实例的话，在分布式消息服务Kafka控制台的“消费组管理 > 消费进度”中，查看消费进度。

本章节主要介绍DataArts Studio的获取Rest Client算子返回值操作。 Rest Client算子可以执行RESTful请求。 本教程主要介绍如何获取Rest Client的返回值，包含以下两个使用场景举例。 通过“响应消息体解析为传递参数定义”获取返回值。 通过EL表达式获取返回值。 通过“响应消息体解析为传递参数定义”获取返回值 如下图所示，第一个Rest Client调用了MRS服务查询集群列表的API，下图为API返回值的JSON消息体。 使用场景：需要获取集群列表中第一个集群的cluster Id，然后作为参数传递给后面的节点使用。 关键配置：在第一个Rest Client的“响应消息体解析为传递参数定义”配置中，配置clusterIdclusters[0].clusterId，后续的Rest Client节点就可以用${clusterId}的方式引用到集群列表中的第一个集群的cluster Id。 Rest Clien作业样例1 JSON消息体 通过EL表达式获取返回值 Rest Client算子可与EL表达式相配合，根据具体的场景选择不同的EL表达式来实现更丰富的用法。您可以参考本教程，根据您的实际业务需要，开发您自己的作业。EL表达式用法可参考表达式概述。 如下图所示，Rest Client调用了MRS服务查询集群列表的API，然后执行Kafka Client发送消息。 使用场景：Kafka Client发送字符串消息，消息内容为集群列表中第一个集群的cluster Id。 关键配置：在Kafka Client中使用如下EL表达式获取Rest API返回消息体中的特定字段：

态势感知（专业版）数据监控功能支持监控态势感知（专业版）管道上下游的生产速率、生产量、消费总速率等指标，您可以根据监控判断业务运行状态。 相关概念 生产者：是用来构建并传输数据到服务端的逻辑概念，负责把数据放入消息队列。 订阅器：用于订阅态势感知（专业版）管道消息，一个管道可由多个订阅器进行订阅，态势感知（专业版）通过订阅器进行消息分发。 消费者：是用来接收并处理数据的运行实体，负责通过订阅器把态势感知（专业版）管道中的消息进行消费并处理。 消息队列：是数据存储和传输的实际容器。 查看监控指标 1. 登录管理控制台。 2. 单击页面左上方的，选择“安全 > 态势感知（专业版）”，进入态势感知（专业版）管理页面。 3. 在左侧导航栏选择“工作空间 > 空间管理”，并在工作空间列表中，单击目标工作空间名称，进入目标工作空间管理页面。 4. 在左侧导航栏选择“日志审计 > 安全数据”，进入安全分析页面。 5. 在左侧数据空间导航栏中，单击数据空间名称，展开数据管道列后，单击目标管道名称后的“更多 > 监控”，进入管道监控页面。 6. 在数据管道的监控页面，查看监控指标。 总览：显示当前管道中生产者、管道、订阅器、消费者之间生产速率等信息。 生产者：显示生产者的“当前生产TPS”、“当前生产速率”、“当前生产量”、“当前消息存储大小”等相关指标信息。 管道：显示当前管道指定时间（近2/6/12/24小时、近7天或自定义）内的“管道存储的消息大小(MB)”、“生产到管道的消息大小(MB)”、“生产到管道的消息数量(条)”、“从管道消费的消息大小(MB)”、“从管道消费的消息数量(条)”、“未确认的消息大小(MB)”、“管道的生产速率(条/秒)”、“管道的消费速率(条/秒)”、“每条消息大小平均值(KB)”、“未卸载的消息大小(B)”等相关指标信息。 订阅器：显示当前订阅器指定时间（近2/6/12/24小时、近7天或自定义）内的“订阅器消费总速率(条/秒)”、“订阅器消费的数据大小(B)”、“订阅器消费的数据数量(条)”、和“活跃消费者”等相关指标信息。

本文介绍如何在事件总线EventBridge管理控制台中添加自建Apache Kafka类型的自定义事件源。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 创建自定义总线。 您已部署Apache Kafka集群，并在您的天翼云VPC内或公共网络可访问集群接入点。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件总线。 3. 在事件总线页面，单击目标总线名称。 4. 在左侧导航栏，单击事件源。 5. 在事件源 页面，单击添加事件源。 6. 在添加自定义事件源 面板，输入名称 和描述 ，事件提供方 选择自建Apache Kafka，配置相关参数，然后单击确认，如图1所示。 图1 创建事件源时事件提供方选择自建Apache Kafka 参数说明 参数 说明 示例 名称 事件源名。 source1 接入点 Apache Kafka集群broker接入点，由IP与端口号拼接而成，以逗号分隔。 172.17.0.25:9092,192.17.0.26:9092,172.17.0.27:9092 Topic topic名称。 topic1 Group ID 填入当前Apache Kafka集群中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 网络配置 根据业务场景选择对应配置。 专有网络 公网网络 专有网络 VPC 网络配置选择专有网络时必填，选择集群所在的VPC。 vpc 子网 网络配置选择专有网络时必填，选择集群所在的子网。 subnet 认证模式 选择认证模式。 PLAINTEXT SASLPLAINTEXT 用户名：填写SASL用户名 密码：填写SASL密码 SASL鉴权方式：可选PLAIN和SCRAMSHA512 PLAINTEXT 消费位点 消费消息的位点。 最新位点 最早位点 最新位点 数据格式 消息value数据编码格式。 JSON Text Binary JSON