kafka单机安装及性能测试

news/2024/10/21 9:27:12/

kafka_0">kafka单机安装及性能测试

Apache Kafka是一个分布式流处理平台，最初由LinkedIn开发，并于2011年开源，随后成为Apache项目。Kafka的核心概念包括发布-订阅消息系统、持久化日志和流处理平台。它主要用于构建实时数据管道和流处理应用，广泛应用于日志聚合、数据传输、实时监控和分析等场景。Kafka具有高吞吐量、低延迟、扩展性强和容错性高等特点。

1. 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

2. Kafka性能测试

使用kafka自带的性能测试脚本，发起写入MQ消息和消费MQ消息的请求。根据不同数量级的消息写入和消息消费测试结果，评估kafka处理消息的能力。

2.1 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优化思路与优化参数

  1. 优化思路
  • 适当调大 batch.size和 linger.ms：这两个参数是配合起来使用的，目的就是缓存更多的数据，减少客户端发起请求的次数。这两个参数根据实际情况调整，注意要适量。
  • 关闭数据发送确认机制：适用于对数据完整性要求不高的场景，比如日志，丢几条无所谓那种
  • 指定数据发送时的压缩算法：默认不压缩，可选压缩算法gzip,snappy,lz4,zstd等
  1. 推荐一组优化参数
  • batch_size=563840: 默认值是 16384
  • linger_ms=30000: 默认值是 0
  • acks=0: 默认值是 1
  • compression_type=“gzip”: 默认值是 None

结果汇总：

设置消息总数(单位：w)设置单个消息大小（单位：字节）设置每秒发送消息数实际写入消息数/秒95%的消息延迟（单位：ms）
10100020001999.761ms
100100050004999.721ms
1000100050004999.961ms

2.2 Kafka消费消息压力测试

对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条消息。

参数解释：

  1. start.time：测试开始的时间，通常以时间戳形式表示，标志着性能测试或监控的开始时刻。

  2. end.time：测试结束的时间，通常以时间戳形式表示，标志着性能测试或监控的结束时刻。

  3. data.consumed.in.MB：在测试期间消费者从Kafka主题中消费的数据总量，以MB（兆字节）为单位。这个参数表示消费者在指定时间段内消费了多少数据。

  4. MB.sec：每秒消费的数据量，以MB（兆字节）为单位。它表示消费者的吞吐量，即每秒能够消费的数据量。

  5. data.consumed.in.nMsg：在测试期间消费者从Kafka主题中消费的消息总数。这个参数表示消费者在指定时间段内消费了多少条消息。

  6. nMsg.sec：每秒消费的消息数。它表示消费者的吞吐量，即每秒能够消费的消息数量。

  7. rebalance.time.ms：在测试期间由于消费者组重新平衡所花费的总时间，以毫秒为单位。消费者组重新平衡是指消费者组内的消费者发生变动（如新增或移除消费者）时，Kafka需要重新分配分区给各个消费者的过程。

  8. fetch.time.ms：在测试期间用于从Kafka获取消息的总时间，以毫秒为单位。这个参数表示消费者花在从Kafka拉取消息上的总时间。

  9. fetch.MB.sec：每秒从Kafka获取的数据量，以MB（兆字节）为单位。这个参数表示消费者在拉取消息时的吞吐量。

  10. fetch.nMsg.sec：每秒从Kafka获取的消息数。这个参数表示消费者在拉取消息时的吞吐量。

这些参数可以帮助评估Kafka消费者在不同负载下的性能，找出可能的瓶颈，并进行相应的优化。

结果汇总：

消费消息总数(单位：w)共消费数据（单位：M）每秒消费数据（单位：M）每秒消费消息数消费耗时（单位：s）
1095.3671089.63193984.96241.064
100953.8287.3769301336.54723.319
10009536.782360.874163831.1354156.664

参考：

  • Kafka压力测试(自带测试脚本)(单机版)
  • 如何做 Kafka 的性能测试

http://www.ppmy.cn/news/1464897.html

相关文章

通过el-tree自定义渲染网页版工作目录，实现鼠标悬浮显示完整名称、用icon区分文件和文件夹等需求

通过el-tree自定义渲染网页版工作目录，实现鼠标悬浮显示完整名称、用icon区分文件和文件夹等需求

目录 一、通过el-tree自定义渲染网页版工作目录 1.1、需求介绍 1.2、使用el-tree生成文档目录 1.2.1、官方基础用法 ①效果 ②代码： 1.2.2、自定义文档目录（实现鼠标悬浮显示完整名称、用icon区分文件和文件夹） ①效果（直接效…
阅读更多...
golang判断字符串是否base64编码的字符串 可准确判断是或否

golang判断字符串是否base64编码的字符串 可准确判断是或否

非常简单的判断一个字符串是否base64编码的方法，就是使用正则base64自身的解码转码来判断，如果失败则一定不是base64编码，否则是base64编码： // 使用正则自身的解码转码来判断字符串是否是base64，可以判断一定不是。 f…
阅读更多...
lua函数执行和虚拟机指令

lua函数执行和虚拟机指令

Stack based vs Register based VM 可直接参考 Stack based vs Register based VM lua函数调用 先看一下lua函数的结构： /* ** Function Prototypes */ typedef struct Proto {CommonHeader;TValue *k; /* constants used by the function */Instruction *code;…
阅读更多...
Linux基础指令用户管理002

Linux基础指令用户管理002

继Linux基础指令用户管理001我们讲述了创建用户和删除用户，我们讲一下如何设置用户密码以及修改用户信息。 操作系统 CentOS Stream 9 设置用户密码 我们使用passwd指令passwd name [rootlocalhost ~]# passwd wg 更改用户 wg 的密码 。 新的密码： …
阅读更多...
反编译 Trino Dockerfile

反编译 Trino Dockerfile

文章目录 反编译 Trino Dockerfile反编译Dockerfile命令反编译后Dockerfile内容获取 Trino 启动脚本卸载 反编译 Trino Dockerfile 反编译Dockerfile命令 alias dfimage"docker run -v /var/run/docker.sock:/var/run/docker.sock --rm ghcr.io/laniksj/dfimage" d…
阅读更多...
AWS安全性身份和合规性之Amazon Macie

AWS安全性身份和合规性之Amazon Macie

Amazon Macie是一项数据安全和数据隐私服务，它利用机器学习（ML）和模式匹配来发现和保护敏感数据。可帮助客户发现、分类和保护其敏感数据，以及监控其数据存储库的安全性。 应用场景： 敏感数据发现 一家金融服务公司…
阅读更多...
B/S架构＋java语言＋Mysqladr数 据 库ADR药物不良反应监测系统源码 ADR药物不良反应监测系统有哪些作用？

B/S架构＋java语言＋Mysqladr数 据 库ADR药物不良反应监测系统源码 ADR药物不良反应监测系统有哪些作用？

B/S架构＋java语言＋Mysqladr数 据 库ADR药物不良反应监测系统源码 ADR药物不良反应监测系统有哪些作用？ 药物不良反应(ADR)是指在合格药物以正常用量和用法用于预防、诊断、治疗疾病或调节生理功能时所发生的意外的、与防治目的无关的、不利或…
阅读更多...
42、Flink 关于窗口状态大小的考量

42、Flink 关于窗口状态大小的考量

关于状态大小的考量 窗口可以被定义在很长的时间段上（比如几天、几周或几个月）并且积累下很大的状态，当估算窗口计算的储存需求时，注意如下： Flink 会为一个元素在它所属的每一个窗口中都创建一个副本。 因此&#x…
阅读更多...
最新文章

Copyright @ 2022~2023