赞
踩
Apache Kafka是一个分布式流处理平台,最初由LinkedIn开发,并于2011年开源,随后成为Apache项目。Kafka的核心概念包括发布-订阅消息系统、持久化日志和流处理平台。它主要用于构建实时数据管道和流处理应用,广泛应用于日志聚合、数据传输、实时监控和分析等场景。Kafka具有高吞吐量、低延迟、扩展性强和容错性高等特点。
安装kafka2.7.0:
下载地址:https://kafka.apache.org/downloads
# 下载 $ wget https://archive.apache.org/dist/kafka/2.7.0/kafka_2.13-2.7.0.tgz $ tar xf kafka_2.13-2.7.0.tgz $ sudo mv kafka_2.13-2.7.0/ /usr/local/kafka2.7.0/ # 修改zookeeper.properties的配置文件。修改dataDir的参数配置,其他的配置默认不变。dataDir=/usr/local/kafka2.7.0/zookeeper $ sudo vi /usr/local/kafka2.7.0/config/zookeeper.properties $ sudo mkdir -p /usr/local/kafka2.7.0/zookeeper/ $ sudo mkdir -p /usr/local/kafka2.7.0/logs/ # 修改server.properties的配置文件。修改listeners、host.name、log.dirs、zookeeper.connect、create.topics.enable和delete.topic.enble的参数配置,没有的配置添加,其他的配置默认不变。 $ sudo vi /usr/local/kafka2.7.0/config/server.properties ######## Socket Server Settings ######## listeners=PLAINTEXT://172.16.0.9:9092 host.name=172.16.0.9 ########### Log Basics ########### log.dirs=/usr/local/kafka2.7.0/logs ########## Zookeeper ########### zookeeper.connect=172.16.0.9:2181 ########## Group Coordinator Settings ######### auto.create.topics.enable=false delete.topic.enable=true # 启动Kafka,使用root用户操作。分为两步,先启动zookeeper,再启动Kafka。 [root@xx]# nohup /usr/local/kafka2.7.0/bin/zookeeper-server-start.sh /usr/local/kafka2.7.0/config/zookeeper.properties > /usr/local/kafka2.7.0/zookeeper-run.log 2>&1 & [root@xx]# sleep 10 [root@xx]# nohup /usr/local/kafka2.7.0/bin/kafka-server-start.sh /usr/local/kafka2.7.0/config/server.properties > /usr/local/kafka2.7.0/kafka-run.log 2>&1 & # 验证。jps查询输出如下择表示启动成功 # jps 101981 Kafka 101420 QuorumPeerMain #zookeeper 102575 Jps
使用kafka自带的性能测试脚本,发起写入MQ消息和消费MQ消息的请求。根据不同数量级的消息写入和消息消费测试结果,评估kafka处理消息的能力。
对kafka节点进行MQ消息服务的压力测试,关注Kafka消息写入的延迟时间是否满足需求。
# 脚本命令位于/usr/local/kafka2.7.0/bin # 创建topic,单机环境replication-factor设置为1。上述server.properties中的auto.create.topics.enable设置为true可以自动创建主题。 $ sudo ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 4 --topic test_perf # 删除topic:sudo ./kafka-topics.sh --delete --topic test_perf --zookeeper localhost:2181 # 查询topic:sudo ./kafka-topics.sh --list --zookeeper localhost:2181 # 指定吞吐量测试时延 $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 100000 --record-size 1000 --throughput 2000 --producer-props bootstrap.servers=172.16.0.9:9092 100000 records sent, 1999.760029 records/sec (1.91 MB/sec), 1.13 ms avg latency, 448.00 ms max latency, 0 ms 50th, 1 ms 95th, 17 ms 99th, 83 ms 99.9th. $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 1000000 --record-size 1000 --throughput 5000 --producer-props bootstrap.servers=172.16.0.9:9092 1000000 records sent, 4999.725015 records/sec (4.77 MB/sec), 0.51 ms avg latency, 481.00 ms max latency, 0 ms 50th, 1 ms 95th, 2 ms 99th, 53 ms 99.9th $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 10000000 --record-size 1000 --throughput 5000 --producer-props bootstrap.servers=172.16.0.9:9092 10000000 records sent, 4999.985000 records/sec (4.77 MB/sec), 0.35 ms avg latency, 424.00 ms max latency, 0 ms 50th, 1 ms 95th, 1 ms 99th, 5 ms 99.9th. # throughput设置0-1,测试producer的最大吞吐量。 # 优化参数:compression.type=snappy,使用snappy算法压缩消息。 $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 10000000 --record-size 1000 --throughput -1 --producer-props bootstrap.servers=172.16.0.9:9092 batch_size=563840 linger_ms=30000 acks=0 compression_type=snappy [2024-03-28 16:57:00,757] WARN The configuration 'batch_size' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig) [2024-03-28 16:57:00,757] WARN The configuration 'compression_type' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig) [2024-03-28 16:57:00,757] WARN The configuration 'linger_ms' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig) 577921 records sent, 115584.2 records/sec (110.23 MB/sec), 239.9 ms avg latency, 491.0 ms max latency. 646464 records sent, 128854.7 records/sec (122.89 MB/sec), 247.7 ms avg latency, 604.0 ms max latency. 313216 records sent, 62418.5 records/sec (59.53 MB/sec), 514.5 ms avg latency, 854.0 ms max latency. 206016 records sent, 41137.4 records/sec (39.23 MB/sec), 724.7 ms avg latency, 1781.0 ms max latency. ... 301184 records sent, 59949.0 records/sec (57.17 MB/sec), 545.7 ms avg latency, 725.0 ms max latency. 10000000 records sent, 62655.463870 records/sec (59.75 MB/sec), 494.30 ms avg latency, 5370.00 ms max latency, 506 ms 50th, 775 ms 95th, 1149 ms 99th, 5221 ms 99.9th.
结果解析:
以写入100w条MQ消息为例,每秒平均向kafka写入了4.77MB的数据,平均4999.725条消息/秒,每次写入的平均延迟为0.51毫秒,最大的延迟为481毫秒。
producer优化思路与优化参数
- 优化思路
- 适当调大 batch.size和 linger.ms:这两个参数是配合起来使用的,目的就是缓存更多的数据,减少客户端发起请求的次数。这两个参数根据实际情况调整,注意要适量。
- 关闭数据发送确认机制:适用于对数据完整性要求不高的场景,比如日志,丢几条无所谓那种
- 指定数据发送时的压缩算法:默认不压缩,可选压缩算法gzip,snappy,lz4,zstd等
- 推荐一组优化参数
- batch_size=563840: 默认值是 16384
- linger_ms=30000: 默认值是 0
- acks=0: 默认值是 1
- compression_type=“gzip”: 默认值是 None
结果汇总:
设置消息总数(单位:w) | 设置单个消息大小(单位:字节) | 设置每秒发送消息数 | 实际写入消息数/秒 | 95%的消息延迟(单位:ms) |
---|---|---|---|---|
10 | 1000 | 2000 | 1999.76 | 1ms |
100 | 1000 | 5000 | 4999.72 | 1ms |
1000 | 1000 | 5000 | 4999.96 | 1ms |
对Kafka节点进行MQ消息处理的压力测试,验证Kafka的消息处理能力。
# 消费10w消息压测结果。先写入10w消息,然后消费 $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 100000 --record-size 1000 --throughput 2000 --producer-props bootstrap.servers=172.16.0.9:9092 $ sudo ./kafka-consumer-perf-test.sh --broker-list 172.16.0.9:9092 --topic test_perf --fetch-size 1048576 --messages 100000 start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec 2024-03-27 14:14:36:989, 2024-03-27 14:14:38:053, 95.3674, 89.6310, 100000, 93984.9624, 1711520077451, -1711520076387, -0.0000, -0.0001 # 消费100w消息压测结果。先写入100w消息,然后消费 $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 1000000 --record-size 1000 --throughput 5000 --producer-props bootstrap.servers=172.16.0.9:9092 $ sudo ./kafka-consumer-perf-test.sh --broker-list 172.16.0.9:9092 --topic test_perf --fetch-size 1048576 --messages 1000000 --threads 1 start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec 2024-03-27 14:20:11:235, 2024-03-27 14:20:14:554, 953.8040, 287.3769, 1000136, 301336.5472, 1711520411703, -1711520408384, -0.0000, -0.0006 # 消费1000w消息压测结果。先写入1000w消息,然后消费 $ sudo ./kafka-producer-perf-test.sh --topic test_perf --num-records 10000000 --record-size 1000 --throughput 5000 --producer-props bootstrap.servers=172.16.0.9:9092 $ sudo ./kafka-consumer-perf-test.sh --broker-list 172.16.0.9:9092 --topic test_perf --fetch-size 1048576 --messages 10000000 --threads 1 start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec 2024-03-27 14:56:24:937, 2024-03-27 14:59:01:601, 9536.7823, 60.8741, 10000041, 63831.1354, 1716562585422, -1716562428758, -0.0000, -0.0058
结果解析:
以本例中消费100w条MQ消息为例总共消费了953.8M的数据,每秒消费数据大小为287.377M,总共消费了1000136条消息,每秒消费301336.547条消息。
参数解释:
start.time:测试开始的时间,通常以时间戳形式表示,标志着性能测试或监控的开始时刻。
end.time:测试结束的时间,通常以时间戳形式表示,标志着性能测试或监控的结束时刻。
data.consumed.in.MB:在测试期间消费者从Kafka主题中消费的数据总量,以MB(兆字节)为单位。这个参数表示消费者在指定时间段内消费了多少数据。
MB.sec:每秒消费的数据量,以MB(兆字节)为单位。它表示消费者的吞吐量,即每秒能够消费的数据量。
data.consumed.in.nMsg:在测试期间消费者从Kafka主题中消费的消息总数。这个参数表示消费者在指定时间段内消费了多少条消息。
nMsg.sec:每秒消费的消息数。它表示消费者的吞吐量,即每秒能够消费的消息数量。
rebalance.time.ms:在测试期间由于消费者组重新平衡所花费的总时间,以毫秒为单位。消费者组重新平衡是指消费者组内的消费者发生变动(如新增或移除消费者)时,Kafka需要重新分配分区给各个消费者的过程。
fetch.time.ms:在测试期间用于从Kafka获取消息的总时间,以毫秒为单位。这个参数表示消费者花在从Kafka拉取消息上的总时间。
fetch.MB.sec:每秒从Kafka获取的数据量,以MB(兆字节)为单位。这个参数表示消费者在拉取消息时的吞吐量。
fetch.nMsg.sec:每秒从Kafka获取的消息数。这个参数表示消费者在拉取消息时的吞吐量。
这些参数可以帮助评估Kafka消费者在不同负载下的性能,找出可能的瓶颈,并进行相应的优化。
结果汇总:
消费消息总数(单位:w) | 共消费数据(单位:M) | 每秒消费数据(单位:M) | 每秒消费消息数 | 消费耗时(单位:s) |
---|---|---|---|---|
10 | 95.367 | 1089.631 | 93984.9624 | 1.064 |
100 | 953.8 | 287.3769 | 301336.5472 | 3.319 |
1000 | 9536.7823 | 60.8741 | 63831.1354 | 156.664 |
参考:
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。