分布式消息服务KAFKA版-华为云

分布式消息服务KAFKA版-Kafka实例的连接地址默认有多少个？

Kafka实例的连接地址默认有多少个？ Kafka实例的连接地址个数和实例的代理个数有关，连接地址个数即为代理个数。每类实例规格对应的代理个数如下表所示。表1 Kafka集群实例规格规格名称代理个数范围单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.cluster.small 3~30 20000 100 15 2000 300~300000 40 kafka.2u4g.cluster 3~30 30000 250 20 2000 300~300000 100 kafka.4u8g.cluster 3~30 100000 500 100 4000 300~600000 200 kafka.8u16g.cluster 3~50 150000 1000 150 4000 300~1500000 375 kafka.12u24g.cluster 3~50 200000 1500 200 4000 300~1500000 625 kafka.16u32g.cluster 3~50 250000 2000 200 4000 300~1500000 750 表2 Kafka单机实例规格规格名称代理个数单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.single.small 1 20000 100 15 2000 100~10000 40 kafka.2u4g.single 1 30000 250 20 2000 100~10000 100 父主题：连接问题

分布式消息服务KAFKA版连接问题

分布式消息服务KAFKA版-为什么限制Topic的总分区数？

为什么限制Topic的总分区数？ Kafka以分区为粒度管理消息，分区多导致生产、存储、消费都碎片化，影响性能稳定性。在使用过程中，当Topic的总分区数达到上限后，用户就无法继续创建Topic。不同规格配置的Topic总分区数不同，如下表所示。表1 Kafka集群实例规格规格名称代理个数范围单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.cluster.small 3~30 20000 100 15 2000 300~300000 40 kafka.2u4g.cluster 3~30 30000 250 20 2000 300~300000 100 kafka.4u8g.cluster 3~30 100000 500 100 4000 300~600000 200 kafka.8u16g.cluster 3~50 150000 1000 150 4000 300~1500000 375 kafka.12u24g.cluster 3~50 200000 1500 200 4000 300~1500000 625 kafka.16u32g.cluster 3~50 250000 2000 200 4000 300~1500000 750 表2 Kafka单机实例规格规格名称代理个数单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.single.small 1 20000 100 15 2000 100~10000 40 kafka.2u4g.single 1 30000 250 20 2000 100~10000 100 父主题： Topic和分区问题

分布式消息服务KAFKA版

分布式消息服务KAFKA版-如何将Kafka实例中的数据同步到另一个Kafka实例中？

如何将Kafka实例中的数据同步到另一个Kafka实例中？ Kafka实例支持创建Kafka数据复制类型的Smart Connect任务时，首先开启Smart Connect，然后创建Kafka数据复制类型的Smart Connect任务，实现两个Kafka实例数据的同步，具体步骤请参考创建Smart Connect任务（Kafka数据复制）。 Kafka实例不支持创建Kafka数据复制类型的Smart Connect任务时，如果需要做Kafka实例迁移，可以同时向两个Kafka实例生产消息，源实例中的消息可继续消费，待源实例的消息数据全部被消费完或老化后，业务可迁移到新的Kafka实例。父主题：实例问题

分布式消息服务KAFKA版实例问题

分布式消息服务KAFKA版-Kafka实例的Topic数量是否有限制？

Kafka实例的Topic数量是否有限制？ Topic数量和Topic总分区数、每个Topic的分区数有关，Kafka实例对Topic总分区数设置了上限，当达到上限后，会导致用户无法继续创建Topic。不同规格配置的Topic总分区数不同，如下表所示。表1 Kafka集群实例规格规格名称代理个数范围单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.cluster.small 3~30 20000 100 15 2000 300~300000 40 kafka.2u4g.cluster 3~30 30000 250 20 2000 300~300000 100 kafka.4u8g.cluster 3~30 100000 500 100 4000 300~600000 200 kafka.8u16g.cluster 3~50 150000 1000 150 4000 300~1500000 375 kafka.12u24g.cluster 3~50 200000 1500 200 4000 300~1500000 625 kafka.16u32g.cluster 3~50 250000 2000 200 4000 300~1500000 750 表2 Kafka单机实例规格规格名称代理个数单个代理TPS 单个代理分区上限单个代理建议消费组数单个代理客户端总连接数上限存储空间范围（GB）单个代理流量规格（MB/s） kafka.2u4g.single.small 1 20000 100 15 2000 100~10000 40 kafka.2u4g.single 1 30000 250 20 2000 100~10000 100 父主题： Topic和分区问题

分布式消息服务KAFKA版 Topic和分区问题

分布式消息服务KAFKA版-为什么Kafka实例常常发生Rebalance（重平衡），导致消息拉取失败？:解决方法/排查思路

解决方法/排查思路场景一：未能及时发送心跳请求排查思路：如果是Java语言客户端，建议排查GC日志是否存在长时间FullGC的情况。该场景可能会导致心跳线程阻塞，服务端检查心跳失败，进入Rebalance状态。解决方法：如果存在长时间FullGC的情况，需要您排查客户端相关问题（FullGC可能是客户端存在内存泄露）。场景二：消费者消费时间间隔过长排查思路：检查单条消息的处理时间是多久，处理max.poll.records条消息会不会超过max.poll.interval.ms时间。消息处理流程是否有网络行为，如写数据库、调用后端API等，在发生Rebalance的场景下消费者下游系统是否正常。解决方法：建议在消费者客户端将max.poll.records值减小，如果消息处理时间较长可以适当增加max.poll.interval.ms。

分布式消息服务KAFKA版消息问题

分布式消息服务KAFKA版-Kafka实例的SASL_SSL开关如何修改？

Kafka实例的SASL_SSL开关如何修改？在“华北-北京一”、“中东-利雅得”、“拉美-圣保罗一”和“拉美-圣地亚哥”区域，Kafka SASL_SSL开关不支持购买实例后修改，在购买时，请慎重选择，如果购买后需要修改，需要重新购买实例。除“华北-北京一”、“中东-利雅得”、“拉美-圣保罗一”和“拉美-圣地亚哥”以外的区域，集群版实例的SASL_SSL开关支持修改，具体步骤请参考修改实例的接入方式。单机实例不支持SASL_SSL功能。开启IPv6的实例不支持动态修改SASL_SSL开关。父主题：实例问题

分布式消息服务KAFKA版实例问题

分布式消息服务KAFKA版-删除Kafka消息:前提条件

前提条件删除消息前，请先在消费代码中设置“auto.offset.reset”参数。“auto.offset.reset”用来指定当Kafka中没有初始偏移量或者当前偏移量不存在（例如当前偏移量已被删除）时，消费者的消费策略。取值如下： latest：偏移量自动被重置到最晚偏移量。 earliest：偏移量自动被重置到最早偏移量。 none：向消费者抛出异常。如果将此配置设置为latest，新增分区时，生产者可能会在消费者重置初始偏移量之前开始向新增加的分区发送消息，从而导致部分消息丢失。

分布式消息服务KAFKA版管理消息

分布式消息服务KAFKA版-变更Kafka实例规格:变更实例规格的影响

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

分布式消息服务KAFKA版

分布式消息服务KAFKA版-查看Kafka消费者信息:查看消费者连接地址（Kafka Manager）

查看消费者连接地址（Kafka Manager）登录Kafka Manager。单击“kafka_cluster”，进入集群详情页。在顶部导航栏单击“Consumers”，进入消费组列表页面。图2 导航栏单击待查看消费者连接地址的消费组名称，进入消费组订阅的Topic列表页面。图3 消费组列表页面单击待查看消费者连接地址的Topic名称，进入Topic详情页。图4 消费组订阅的Topic列表页面在“Consumer Instance Owner”中，查看消费者连接地址。图5 Topic详情页

分布式消息服务KAFKA版管理消费组

分布式消息服务KAFKA版-测试Kafka实例TPS:测试脚本

测试脚本 ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=${连接地址} acks=1 batch.size=16384 linger.ms=10 --topic ${Topic名称} --num-records 10000000 --record-size 1024 --throughput -1 --producer.config ../config/producer.properties bootstrap.servers：购买Kafka实例后，获取的Kafka实例的地址。 acks：消息主从同步策略，acks=1表示异步复制消息，acks=-1表示同步复制消息。 batch.size：每次批量发送消息的大小（单位为字节）。 linger.ms：两次发送时间间隔。 topic：创建Topic中设置的Topic名称。 num-records：总共需要发送的消息数。 record-size：每条消息的大小。 throughput：每秒发送的消息数。

分布式消息服务KAFKA版测试实例性能

分布式消息服务KAFKA版-测试Kafka实例TPS:测试结果

测试结果测试场景一（实例是否开启SASL）：相同的Topic（30分区、3副本、异步复制、异步落盘），实例分为开启SASL和未开启SASL，测试结果如下：表3 测试结果实例规格磁盘类型代理数量 TPS（开启SASL） TPS（未开启SASL） kafka.2u4g.cluster 超高I/O 3 100000 280000 kafka.4u8g.cluster 超高I/O 3 170000 496000 kafka.8u16g.cluster 超高I/O 3 200000‬ 730000 kafka.12u24g.cluster 超高I/O 3 320000 790000 kafka.16u32g.cluster 超高I/O 3 360000 1000000 结论：在Topic相同的情况下，生产消息到规格相同、接入方式不同的Kafka实例，未开启SASL的实例TPS高于开启SASL的实例TPS。测试场景二（同步/异步复制）：相同的实例（超高I/O、3个代理、未开启SASL），不同复制机制的Topic，生产者进程数为3时，测试结果如下：表4 测试结果实例规格是否同步落盘副本数分区数 TPS（同步复制） TPS（异步复制） kafka.2u4g.cluster 否 3 30 100000 280000 kafka.4u8g.cluster 否 3 30 230000 496000 kafka.8u16g.cluster 否 3 30 342000 730000 kafka.12u24g.cluster 否 3 30 383000 790000 kafka.16u32g.cluster 否 3 30 485000 1000000 结论：生产消息到同一个Kafka实例的不同Topic中，Topic除了复制机制，其他参数相同，异步复制Topic的TPS高于同步复制Topic的TPS。测试场景三（是否同步落盘）：相同的实例（超高I/O、3个代理、未开启SASL），不同落盘机制的Topic，测试结果如下：表5 测试结果实例规格是否同步复制副本数分区数 TPS（同步落盘） TPS（异步落盘） kafka.2u4g.cluster 否 3 30 30000 280000 kafka.4u8g.cluster 否 3 30 32500 496000 kafka.8u16g.cluster 否 3 30 36100 730000 kafka.12u24g.cluster 否 3 30 37400 790000 kafka.16u32g.cluster 否 3 30 40400 1000000 结论：生产消息到同一个Kafka实例的不同Topic中，Topic除了落盘机制，其他参数相同，异步落盘Topic的TPS远远高于同步落盘Topic的TPS。测试场景四（不同磁盘类型）：相同的Topic（30分区、3副本、异步复制、异步落盘），不同磁盘类型的实例，测试结果如下：表6 测试结果实例规格代理数量是否开启SASL TPS（高I/O） TPS（超高I/O） kafka.2u4g.cluster 3 否 110000 250000 kafka.4u8g.cluster 3 否 135000 380000 kafka.8u16g.cluster 3 否 213000 480000 kafka.12u24g.cluster 3 否 240000 577000 kafka.16u32g.cluster 3 否 280000 840000 结论：在Topic相同的情况下，生产消息到规格相同、磁盘类型不同的Kafka实例，超高I/O的实例TPS高于高I/O的实例TPS。测试场景五（不同分区数）：相同的实例（超高I/O、3个代理、未开启SASL），不同分区数的Topic，测试结果如下：表7 测试结果实例规格是否同步落盘是否同步复制副本数 TPS（3分区） TPS（12分区） TPS（100分区） kafka.2u4g.cluster 否否 3 250000 260000 250000 kafka.4u8g.cluster 否否 3 330000 280000 260000 kafka.8u16g.cluster 否否 3 480000 410000 340000 kafka.12u24g.cluster 否否 3 570000 750000 520000 kafka.16u32g.cluster 否否 3 840000 1000000 630000 结论：生产消息到同一个Kafka实例的不同Topic中，Topic除了分区数量，其他参数相同。随着分区数的增加，Kafka的性能通常会随之增加，当分区数达到一定程度后，继续增加分区数可能会导致性能下降。

分布式消息服务KAFKA版测试实例性能

分布式消息服务KAFKA版-测试Kafka实例TPS:测试环境

测试环境进行TPS测试前，您需要先构建如下的测试环境：购买如表1所示实例，购买步骤请参考购买Kafka实例。表1 实例参数名称代理数量规格是否开启SASL 磁盘类型 kafka-01 3 kafka.2u4g.cluster 是超高I/O kafka-02 3 kafka.4u8g.cluster 是超高I/O kafka-03 3 kafka.8u16g.cluster 是超高I/O kafka-04 3 kafka.12u24g.cluster 是超高I/O kafka-05 3 kafka.16u32g.cluster 是超高I/O kafka-06 3 kafka.2u4g.cluster 否超高I/O kafka-07 3 kafka.4u8g.cluster 否超高I/O kafka-08 3 kafka.8u16g.cluster 否超高I/O kafka-09 3 kafka.12u24g.cluster 否超高I/O kafka-10 3 kafka.16u32g.cluster 否超高I/O kafka-11 3 kafka.2u4g.cluster 否高I/O kafka-12 3 kafka.4u8g.cluster 否高I/O kafka-13 3 kafka.8u16g.cluster 否高I/O kafka-14 3 kafka.12u24g.cluster 否高I/O kafka-15 3 kafka.16u32g.cluster 否高I/O 购买完成后，在实例详情页获取Kafka实例的内网明文连接地址。购买实例后，创建如表2所示Topic，创建步骤请参考创建Kafka Topic。表2 Topic参数名称是否同步复制是否同步落盘副本数分区数 topic-01 否否 3 30 topic-02 是否 3 30 topic-03 否是 3 30 topic-04 否否 3 3 topic-05 否否 3 12 topic-06 否否 3 100 获取测试工具。获取Kafka命令行工具2.7.2版本。购买客户端服务器。购买1台E CS 服务器（区域、可用区、虚拟私有云、子网、安全组与Kafka实例保持一致，Linux系统），具体步骤请参考购买弹性云服务器。购买完成ECS后，需要在ECS中完成以下配置：安装Java JDK，并配置JAVA_HOME与PATH环境变量。 export JAVA_HOME=/root/jdk1.8.0_231 export PATH=$JAVA_HOME/bin:$PATH 下载Kafka命令行工具2.7.2版本，并解压。 tar -zxf kafka_2.12-2.7.2.tgz

分布式消息服务KAFKA版测试实例性能

分布式消息服务KAFKA版-购买Kafka实例:准备实例依赖资源

准备实例依赖资源创建Kafka实例前，请提前准备好如表1所示资源。表1 Kafka实例依赖资源资源名称要求创建指导 VPC和子网 Kafka实例可以使用当前账号下已创建的VPC和子网，也可以使用新创建的VPC和子网，还可以使用共享VPC和子网，请根据实际需要进行配置。共享VPC是基于资源访问管理（Resource Access Manager，简称 RAM ）服务的机制，VPC的所有者可以将VPC内的子网共享给一个或者多个账号使用。通过共享VPC功能，可以简化网络配置，帮助您统一配置和运维多个账号下的资源，有助于提升资源的管控效率，降低运维成本。有关VPC子网共享的更多信息，请参见共享VPC。在创建VPC和子网时应注意：创建的VPC与Kafka实例在相同的区域。在创建VPC和子网时应注意：创建的VPC与Kafka实例在相同的区域。子网开启IPv6后，Kafka实例支持IPv6功能。Kafka实例开启IPv6后，客户端可以使用IPv6地址连接实例。创建VPC和子网的操作指导请参考创建虚拟私有云和子网，若需要在已有VPC上创建和使用新的子网，请参考为虚拟私有云创建新的子网。安全组不同的Kafka实例可以重复使用相同的安全组，也可以使用不同的安全组，请根据实际需要进行配置。连接Kafka实例前，请根据连接方式配置对应的安全组，具体请参考表2。创建安全组的操作指导请参考创建安全组，为安全组添加规则的操作指导请参考添加安全组规则。弹性IP地址如果客户端使用公网连接Kafka实例，请提前创建弹性IP地址。在创建弹性IP地址时，应注意如下要求：创建的弹性IP地址与Kafka实例在相同的区域。弹性IP地址的数量必须与Kafka实例的代理个数相同。 Kafka控制台无法识别开启IPv6转换功能的弹性IP地址。创建弹性IP地址的操作指导请参考申请弹性公网IP。

分布式消息服务KAFKA版

分布式消息服务KAFKA版-Kafka客户端使用建议:producer使用建议

producer使用建议同步复制客户端需要配合使用：acks=all 配置发送失败重试：retries=3 发送优化：对于时延敏感的信息，设置linger.ms=0。对于时延不敏感的信息，设置linger.ms在100~1000之间。生产端的JVM内存要足够，避免内存不足导致发送阻塞。时间戳设置为与当地时间一致，避免时间戳为未来时间导致消息无法老化。尽量复用producer，不要频繁创建producer。当producer开启幂等时（生产者客户端3.0及之后的版本默认开启幂等），生产消息会在服务端创建生产者状态对象，若频繁创建producer，会导致服务端创建大量生产者状态对象后无法及时回收，服务端内存占用飙升，进而导致服务端性能急剧下降。如果不需要使用幂等功能，请将“enable.idempotence”设置为“false”。

分布式消息服务KAFKA版配置Kafka客户端

分布式消息服务KAFKA版-Kafka客户端使用建议:consumer使用建议

consumer使用建议 consumer的owner线程需确保不会异常退出，避免客户端无法发起消费请求，阻塞消费。确保处理完消息后再做消息commit，避免业务消息处理失败，无法重新拉取处理失败的消息。通常不建议对每条消息都进行commit，如果对每条消息都进行了commit，会导致OFFSET_COMMIT请求过多，进而导致CPU使用率过高。例如：如果一个消费请求拉取1000条消息，每条都commit，则commit请求TPS是消费的1000倍，消息体越小，这个比例越大。建议隔一定条数或时间，批量commit，或打开enable.auto.commit，这样设置会存在一个缺点，即在客户端故障时，可能丢失一部分缓存的消费进度，导致重复消费。请根据业务实际情况，设置批量commit。 consumer不能频繁加入和退出group，频繁加入和退出，会导致consumer频繁做rebalance，阻塞消费。同一消费组内consumer数量不能超过该消费组订阅的分区总数，否则会有consumer拉取不到消息。 consumer需周期poll，维持和server的心跳，避免心跳超时，导致consumer频繁加入和退出，阻塞消费。 consumer拉取的消息本地缓存应有大小限制，避免OOM（Out of Memory）。 consumer session设置为30秒，session.timeout.ms=30000。 Kafka不能保证消费重复的消息，业务侧需保证消息处理的幂等性。消费线程退出要调用consumer的close方法，避免同一个组的其他消费者阻塞session.timeout.ms的时间。消费组名称开头不使用特殊字符（如#），使用特殊字符可能会导致云监控无法展示此消费组的监控数据。

分布式消息服务KAFKA版配置Kafka客户端

云服务器内容精选

分布式消息服务KAFKA版

7*24

备案

专业服务

退订

建议反馈

售前咨询热线