连接Kafka网络要求 客户端可以通过公网连接Kafka实例，也可以通过内网连接Kafka实例。使用内网连接时，注意以下几点： 如果客户端和Kafka实例部署在同一个VPC内，网络默认互通。 如果客户端和Kafka实例部署在不同VPC中，由于VPC之间逻辑隔离，客户端和Kafka实例不能直接通信，需要打通VPC之间的网络。 客户端连接Kafka实例的方式如表1所示。 表1 连接方式说明 连接方式 实现方式 参考文档 公网连接 客户端通过IPv4地址访问Kafka实例：在Kafka控制台开启公网访问，配置弹性公网IP，客户端通过弹性公网IP访问Kafka实例。 客户端通过IPv6地址访问Kafka实例：在Kafka控制台开启IPv6功能，将IPv6地址加入到共享带宽中，实现客户端通过公网访问Kafka实例。 配置Kafka实例的公网访问 使用DNAT通过端口映射方式，实现客户端通过公网连接Kafka实例。 使用DNAT实现公网访问Kafka 内网连接 客户端和Kafka实例部署在同一个VPC中，此时网络默认互通。 - 客户端和Kafka实例部署在同一个Region的不同VPC中，利用 VPC终端节点 在不同VPC间建立跨VPC的连接通道，实现跨VPC访问。 使用VPCEP实现跨VPC访问Kafka 客户端和Kafka实例部署在同一个Region的不同VPC中，利用VPC对等连接将两个VPC间的网络打通，实现跨VPC访问。 对等连接 客户端连接Kafka实例前，需要在Kafka实例所属安全组中配置如下安全组规则，否则会连接失败。 创建安全组后，系统默认添加入方向“允许安全组内的弹性云服务器彼此通信”规则和出方向“放通全部流量”规则，此时使用内网通过同一个VPC访问Kafka实例，无需添加表2的规则。 表2中的源地址以全网段放通为例，可根据实际安全需要修改为客户端的IP地址。 表2 安全组规则 方向 协议 类型 端口 源地址 说明 入方向 TCP IPv4 9094 0.0.0.0/0 通过公网访问Kafka实例（明文接入）。 入方向 TCP IPv4 9092 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（明文接入）。 使用对等连接跨VPC访问Kafka实例（明文接入）。 入方向 TCP IPv6 9192 ::/0 通过IPv6地址访问Kafka实例（关闭SSL加密）（内网和公网都适用）。 入方向 TCP IPv4 9095 0.0.0.0/0 通过公网访问Kafka实例（密文接入）。 入方向 TCP IPv4 9093 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（密文接入）。 使用对等连接跨VPC访问Kafka实例（密文接入）。 入方向 TCP IPv6 9193 ::/0 通过IPv6地址访问Kafka实例（开启SSL加密）（内网和公网都适用）。 入方向 TCP IPv4 9011 198.19.128.0/17 通过VPC终端节点实现跨VPC访问Kafka实例（密文接入和明文接入都适用）。 入方向 TCP IPv4 9011 0.0.0.0/0 使用DNAT访问Kafka实例（密文接入和明文接入都适用）。 父主题： 配置Kafka网络连接

查看Kafka Manager 在进入Kafka Manager集群管理页面后，您可以查看Kafka集群的监控、代理等信息。 集群信息页 单击Clusters中的集群列表，即可进入集群信息页。如图3所示。 图中①区域表示功能导航栏。 Cluster：集群，统计集群列表和集群详情。 Brokers：代理，统计当前集群中各代理的状态信息。 Topic：队列，统计当前集群中的kafka队列。 Preferred Replica Election：强制进行一次队列leader的最优选举（不建议用户操作）。 Reassign Partitions：进行分区副本的重分配（不建议用户操作）。 Consumers：统计集群中的消费组状态。 图中②区域表示集群信息统计，包含集群的Topic数和集群的代理数。 图3 集群信息页 集群所有代理统计页 单击功能导航栏中的Brokers，即可进入代理统计页。如图4所示。 图中①区域代理列表，包含总的字节流入和字节流出。 图中②集群监控信息。 图4 所有代理统计页 具体代理统计页 单击id列表中具体的Broker，即可查看对应代理的统计信息。如图5所示。 图中①区域表示对应代理总的统计信息，包括队列数、分区数、分区leader数、消息速率占比、写入字节占比以及流出字节占比。 图中②区域表示代理监控信息。 图5 具体Broker信息 查看实例的Topic 在导航栏选择Topic，并在下拉列表中选择List。页面如图6所示，展示了队列列表以及分区数等。 列表中以“__”开头的队列为内部队列，严禁操作，否则可能导致业务问题。 图6 查看实例的Topic 队列详情页 单击具体的Topic名称，进入如图7所示页面。 图中①区域表示队列基本信息，包括副本数(Replication)，分区数(Number of Partitions)，消息数(Sum of partition offsets)等。 图中②区域表示代理与队列分区的对应关系。 图中③区域表示该队列的消费组列表。单击消费组名称可进入该消费组的详情页。 图中④区域表示队列的配置信息。详情参考Kafka队列官方配置文档(https://kafka.apache.org/documentation/#topicconfigs)。 图中⑤区域表示队列监控数据统计。 图中⑥区域表示队列分区信息，包括分区消息数(Latest Offset)，分区leader(Leader)，副本列表(Replicas)，同步副本列表(In Sync Replicas)。 图7 队列详情页 查看消费组列表 导航栏中单击Consumers，即可查看当前集群中的消费组列表。 只显示14天内有消费记录的消费组。 图8 集群的消费组列表 查看消费组详情页 单击消费组名称可进入消费组详情页面，展示消费组消费的所有队列列表以及每个队列的可消费数（Total Lag）。 图9 消费组详情页面 查看消费组队列详情页 单击队列名称，即可进入详情页面，查看消费组消费在队列中每个分区的消费状态。包括分区编号（Partition）、分区消息数（LogSize）、分区消费进度（Consumer Offset）、分区剩余可消费数（Lag）和最近消费该分区的消费者（Consumer Instance Owner）。 图10 消费组队列详情页面

登录Kafka Manager 创建一台与Kafka实例相同VPC和相同安全组的Windows服务器，详细步骤请参考购买并登录Windows弹性云服务器。 如果是已经开启了公网访问，该步骤为可选，在本地浏览器中即可访问，不需要单独的Windows弹性云服务器。 在实例详情信息页面，获取Kafka Manager地址。 未开启公网访问时，Kafka Manager地址为“Manager内网访问地址”。 图1 Kafka Manager内网访问地址 已开启公网访问时，Kafka Manager地址为“Manager公网访问地址”。 图2 Kafka Manager公网访问地址 在浏览器中输入Kafka Manager的地址，进入Kafka Manager登录页面。 如果是开启了公网访问，在本地浏览器输入Kafka Manager地址访问；如果没有开启公网访问，需要登录1的弹性云服务器，然后在浏览器输入Kafka Manager地址访问。 输入创建实例时设置的Kafka Manager用户名和密码，即可管理Kafka集群。

迁移准备 配置网络环境。 Kafka实例分内网地址以及公网地址两种网络连接方式。如果使用公网地址，则消息生成与消费客户端需要有公网访问权限，并配置如下安全组。 表1中的源地址以全网段放通为例，可根据实际安全需要修改为客户端的IP地址。 表1 安全组规则 方向 协议 端口 源地址 说明 入方向 TCP 9094 0.0.0.0/0 通过公网访问Kafka（明文接入）。 入方向 TCP 9095 0.0.0.0/0 通过公网访问Kafka（密文接入）。 创建目标Kafka实例。 目标Kafka的规格不能低于原业务使用的Kafka规格。具体请参考购买Kafka实例。 在目标Kafka实例中创建Topic。 在目标Kafka实例上创建与原Kafka实例相同配置的Topic，包括Topic名称、副本数、分区数、消息老化时间，以及是否同步复制和落盘等。具体请参考创建Kafka Topic。

迁移方案三：先迁消费，再迁生产 指首先通过Smart Connect同步两个Kafka的消息，其次将消费端迁移到新Kafka，最后将生产端迁移到新Kafka。 本方案依赖于Smart Connect，Smart Connect实时同步源端和目标端的数据，但是消费进度是通过批处理同步的，可能会导致源端和目标端每个分区的消费进度存在0-100之间的差异，存在少量重复消费问题。此场景适用于生产端不可停止，端到端有时延要求，但是可以兼容少量重复消费的业务。 创建Kafka数据复制的Smart Connect任务，用于同步两个Kafka的消息。具体步骤请参见配置Kafka间的数据复制。 在Kafka控制台的“消息查询”页面，查看两个Kafka的最新消息是否一致，确认两个Kafka的同步进度是否一致。具体步骤请参见查看Kafka消息。 是，执行3。 否，在监控页面查看两个Kafka的“Kafka每分钟同步数据量”是否正常，如果正常，先等待两个Kafka的同步进度一致，然后执行3。 将消费客户端的Kafka连接地址修改为新Kafka实例的连接地址。 重启消费业务，使得消费者从新Kafka实例中消费消息。 观察消费者是否能正常从新Kafka实例中获取数据。 修改生产客户端，Kafka连接地址改为新Kafka实例的连接地址。 重启生产客户端，将生产业务迁移到新Kafka实例中。 生产业务迁移后，观察连接新Kafka实例的消费业务是否正常。 迁移结束。

迁移方案二：同时消费，后迁生产 指消费者业务启用多个消费客户端，分别向原Kafka和新Kafka实例消费消息，然后将生产业务切到新Kafka实例，这样能确保所有消息都被及时消费。 本方案中消费业务会在一段时间内同时消费原Kafka和新Kafka实例。由于在迁移生产业务之前，已经有消费业务运行在新Kafka实例上，因此不会存在端到端时延的问题。但在迁移生产的开始阶段，同时消费原Kafka与新Kafka实例，会导致部分消息之间的生产顺序无法保证，存在消息乱序的问题。此场景适用于对端到端时延有要求，却对消息顺序不敏感的业务。 启动新的消费客户端，配置Kafka连接地址为新Kafka实例的连接地址，消费新Kafka实例中的数据。 原有消费客户端需继续运行，消费业务同时消费原Kafka与新Kafka实例的消息。 修改生产客户端，Kafka连接地址改为新Kafka实例的连接地址。 重启生产客户端，将生产业务迁移到新Kafka实例中。 生产业务迁移后，观察连接新Kafka实例的消费业务是否正常。 等待原Kafka中数据消费完毕，关闭原有消费业务客户端。 迁移结束。

迁移方案一：先迁生产，再迁消费 指先将生产消息的业务迁移到新的Kafka，原Kafka不会有新的消息生产。待原有Kafka实例的消息全部消费完成后，再将消费消息业务迁移到新的Kafka，开始消费新Kafka实例的消息。 本方案为业界通用的迁移方案，操作步骤简单，迁移过程由业务侧自主控制，整个过程中消息不会存在乱序问题，适用于对消息顺序有要求的场景。但是该方案中需要等待消费者业务直至消费完毕，存在一个时间差的问题，部分数据可能存在较大的端到端时延。 将生产客户端的Kafka连接地址修改为新Kafka实例的连接地址。 重启生产业务，使得生产者将新的消息发送到新Kafka实例中。 观察各消费组在原Kafka的消费进度，直到原Kafka中数据都已经被消费完毕。 将消费客户端的Kafka连接地址修改为新Kafka实例的连接地址。 重启消费业务，使得消费者从新Kafka实例中消费消息。 观察消费者是否能正常从新Kafka实例中获取数据。 迁移结束。

使用PEM格式证书访问Kafka实例 以下示例演示在Java客户端使用PEM格式证书访问Kafka实例。 参考Java客户端接入示例，连接Kafka实例生产消费消息。其中，生产消息配置文件和消费消息配置文件中的SASL信息修改为如下内容。 #SASL认证机制为“PLAIN”时，配置信息如下。 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="username" \ password="password"; #SASL认证机制为“SC RAM -SHA-512”时，配置信息如下。 sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="username" \ password="password"; #设置Kafka安全协议。 security.protocol=SASL_SSL #ssl truststore.location为SSL证书的存放路径，如下代码以Windows系统路径格式举例，您在使用时请根据实际运行环境修改路径格式。 ssl.truststore.location=E:\\temp\\client.pem #ssl truststore.password为服务器证书密码，使用PEM格式证书访问Kafka实例，无需配置ssl.truststore.password参数。 #ssl.truststore.password=dms@kafka #ssl.endpoint.identification.algorithm为证书 域名 校验开关，为空则表示关闭，这里需要保持关闭状态，必须设置为空。 ssl.endpoint.identification.algorithm= #添加ssl.truststore.type参数，指定客户端使用的证书类型是PEM。 ssl.truststore.type=PEM

测试结果 表2 测试结果 分区数 副本数 是否同步复制 batch.size 是否跨AZ生产 客户端消息生产速率 服务端CPU消耗（broker-0） 服务端CPU消耗（broker-1） 服务端CPU消耗（broker-2） 3 1 否 1KB 否 34128 58.10% 56.70% 53.30% 3 1 否 16KB 否 102399 24.10% 25.00% 23.30% 3 1 否 1KB 是 8523 17.20% 16.70% 18.80% 3 3 是 1KB 否 3981 60.00% 55.20% 50.00% 3 3 否 1KB 否 14468 86.70% 80.60% 86.20% 通过上表的测试结果，得出以下结论，仅供参考： 生产请求的batch.size变大16倍时，客户端消息生产速率增加，服务端CPU消耗减少。 同AZ生产和跨AZ生产相比，客户端消息生产速率增加，服务端CPU消耗也随之增加。 副本从1变成3时，客户端消息生产速率下降较多，服务端CPU消耗增加。 异步复制的Topic和同步复制的Topic相比，客户端消息生产速率增加，服务端CPU消耗也随之增加。

测试步骤 测试场景一：批处理大小 登录客户端服务器，进入“kafka_2.12-2.7.2/bin”目录下，执行以下脚本。 batch.size=1KB，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-01 --num-records 8000000 --record-size 1024 --throughput 102400 执行结果如下： 8000000 records sent, 34128.673632 records/sec (33.33 MB/sec), 879.91 ms avg latency, 4102.00 ms max latency, 697 ms 50th, 2524 ms 95th, 2888 ms 99th, 4012 ms 99.9th. 客户端消息生产速率=34128 batch.size=16KB，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=16384 linger.ms=0 --topic Topic-01 --num-records 100000000 --record-size 1024 --throughput 102400 执行结果如下： 100000000 records sent, 102399.318430 records/sec (100.00 MB/sec), 4.72 ms avg latency, 914.00 ms max latency, 1 ms 50th, 5 ms 95th, 162 ms 99th, 398 ms 99.9th. 客户端消息生产速率=102399 登录Kafka实例控制台，单击测试实例名称，进入实例详情页。 在左侧导航栏单击“监控”，进入监控页面。 在“节点”页签，查看服务端节点的CPU使用率。 图1 broker-0的CPU使用率（batch.size=1KB） CPU消耗=58.10% 图2 broker-0的CPU使用率（batch.size=16KB） CPU消耗=24.10% 图3 broker-1的CPU使用率（batch.size=1KB） CPU消耗=56.70% 图4 broker-1的CPU使用率（batch.size=16KB） CPU消耗=25% 图5 broker-2的CPU使用率（batch.size=1KB） CPU消耗=53.30% 图6 broker-2的CPU使用率（batch.size=16KB） CPU消耗=23.30% 测试场景二：是否跨AZ生产 登录客户端服务器，进入“kafka_2.12-2.7.2/bin”目录下，执行以下脚本。 客户端服务器和实例在相同的AZ中，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-01 --num-records 8000000 --record-size 1024 --throughput 102400 执行结果如下： 8000000 records sent, 34128.673632 records/sec (33.33 MB/sec), 879.91 ms avg latency, 4102.00 ms max latency, 697 ms 50th, 2524 ms 95th, 2888 ms 99th, 4012 ms 99.9th. 客户端消息生产速率=34128 客户端服务器和实例在不同的AZ中，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-01 --num-records 4000000 --record-size 1024 --throughput 102400 执行结果如下： 4000000 records sent, 8523.042044 records/sec (8.32 MB/sec), 3506.20 ms avg latency, 11883.00 ms max latency, 1817 ms 50th, 10621 ms 95th, 11177 ms 99th, 11860 ms 99.9th. 客户端消息生产速率=8523 登录Kafka实例控制台，单击测试实例名称，进入实例详情页。 在左侧导航栏单击“监控”，进入监控页面。 在“节点”页签，查看服务端节点的CPU使用率。 图7 broker-0的CPU使用率（客户端服务器和实例在相同的AZ中） CPU消耗=58.10% 图8 broker-0的CPU使用率（客户端服务器和实例在不同的AZ中） CPU消耗=17.20% 图9 broker-1的CPU使用率（客户端服务器和实例在相同的AZ中） CPU消耗=56.70% 图10 broker-1的CPU使用率（客户端服务器和实例在不同的AZ中） CPU消耗=16.70% 图11 broker-2的CPU使用率（客户端服务器和实例在相同的AZ中） CPU消耗=53.30% 图12 broker-2的CPU使用率（客户端服务器和实例在不同的AZ中） CPU消耗=18.80% 测试场景三：副本数 登录客户端服务器，进入“kafka_2.12-2.7.2/bin”目录下，执行以下脚本。 1副本，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-01 --num-records 8000000 --record-size 1024 --throughput 102400 执行结果如下： 8000000 records sent, 34128.673632 records/sec (33.33 MB/sec), 879.91 ms avg latency, 4102.00 ms max latency, 697 ms 50th, 2524 ms 95th, 2888 ms 99th, 4012 ms 99.9th. 客户端消息生产速率=34128 3副本，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-02 --num-records 4000000 --record-size 1024 --throughput 102400 执行结果如下： 4000000 records sent, 14468.325219 records/sec (14.13 MB/sec), 2069.99 ms avg latency, 7911.00 ms max latency, 846 ms 50th, 6190 ms 95th, 6935 ms 99th, 7879 ms 99.9th. 客户端消息生产速率=14468 登录Kafka实例控制台，单击测试实例名称，进入实例详情页。 在左侧导航栏单击“监控”，进入监控页面。 在“节点”页签，查看服务端节点的CPU使用率。 图13 broker-0的CPU使用率（1副本） CPU消耗=58.10% 图14 broker-0的CPU使用率（3副本） CPU消耗=86.70% 图15 broker-1的CPU使用率（1副本） CPU消耗=56.70% 图16 broker-1的CPU使用率（3副本） CPU消耗=80.60% 图17 broker-2的CPU使用率（1副本） CPU消耗=53.30% 图18 broker-2的CPU使用率（3副本） CPU消耗=86.20% 测试场景四：同步/异步复制 登录客户端服务器，进入“kafka_2.12-2.7.2/bin”目录下，执行以下脚本。 异步复制，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=1 batch.size=1024 linger.ms=0 --topic Topic-02 --num-records 4000000 --record-size 1024 --throughput 102400 执行结果如下： 4000000 records sent, 14468.325219 records/sec (14.13 MB/sec), 2069.99 ms avg latency, 7911.00 ms max latency, 846 ms 50th, 6190 ms 95th, 6935 ms 99th, 7879 ms 99.9th. 客户端消息生产速率=14468 同步复制，执行脚本如下： ./kafka-producer-perf-test.sh --producer-props bootstrap.servers=192.168.0.69:9092,192.168.0.42:9092,192.168.0.66:9092 acks=-1 batch.size=1024 linger.ms=0 --topic Topic-03 --num-records 1000000 --record-size 1024 --throughput 102400 执行结果如下： 1000000 records sent, 3981.937930 records/sec (3.89 MB/sec), 7356.98 ms avg latency, 19013.00 ms max latency, 6423 ms 50th, 14381 ms 95th, 18460 ms 99th, 18975 ms 99.9th. 客户端消息生产速率=3981 登录Kafka实例控制台，单击测试实例名称，进入实例详情页。 在左侧导航栏单击“监控”，进入监控页面。 在“节点”页面，查看服务端节点的CPU使用率。 图19 broker-0的CPU使用率（异步复制） CPU消耗=86.70% 图20 broker-0的CPU使用率（同步复制） CPU消耗=60% 图21 broker-1的CPU使用率（异步复制） CPU消耗=80.60% 图22 broker-1的CPU使用率（同步复制） CPU消耗=55.20% 图23 broker-2的CPU使用率（异步复制） CPU消耗=86.20% 图24 broker-2的CPU使用率（同步复制） CPU消耗=50%

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

测试环境 进行性能测试前，您需要先构建如下的测试环境： 购买一个Kafka实例，参数信息如下，购买方法请参考购买Kafka实例。 区域：华北-北京四 项目：华北-北京四 可用区：可用区1 实例名称：kafka-test 企业项目：default 版本：2.7 代理规格：kafka.2u4g.cluster 代理个数：3 存储空间：超高I/O，200GB 容量阈值策略：自动删除 虚拟私有云：选择虚拟私有云 安全组：选择安全组 内网访问：开启“明文接入” 公网访问：不开启 更多配置：不开启“Smart Connect”和“Kafka自动创建Topic” 购买完成后，在实例详情页获取Kafka实例的内网明文连接地址。 在购买的Kafka实例中，创建如下参数的3个Topic，具体步骤请参考创建Kafka Topic。 Topic-01：3分区1副本，异步复制 Topic-02：3分区3副本，异步复制 Topic-03：3分区3副本，同步复制 获取测试工具。 获取Kafka命令行工具2.7.2版本。 购买客户端服务器。 购买如下参数的2台E CS 服务器，具体步骤请参考购买弹性云服务器。 区域、可用区、虚拟私有云、子网、安全组与Kafka实例保持一致，规格为4U8G，Linux系统的ECS。 区域、虚拟私有云、子网、安全组与Kafka实例保持一致，“可用区”为“可用区2”，规格为4U8G，Linux系统的ECS。 购买完成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-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：每秒发送的消息数。

测试结果 测试场景一（实例是否开启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 测试场景二（同步/异步复制）：相同的实例（超高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 测试场景三（是否同步落盘）：相同的实例（超高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 测试场景四（不同磁盘类型）：相同的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 测试场景五（不同分区数）：相同的实例（超高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

测试环境 进行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台ECS服务器（区域、可用区、虚拟私有云、子网、安全组与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