。

Kafka的核心概念包括发布-订阅消息系统、持久化日志和流处理平台。

它主要用于构建实时数据管道和流处理应用#xff…kafka单机安装及性能测试

Apache

Kafka是一个分布式流处理平台最初由LinkedIn开发并于2011年开源随后成为Apache项目。

Kafka的核心概念包括发布-订阅消息系统、持久化日志和流处理平台。

它主要用于构建实时数据管道和流处理应用广泛应用于日志聚合、数据传输、实时监控和分析等场景。

Kafka具有高吞吐量、低延迟、扩展性强和容错性高等特点。

Kafka安装

下载地址https://kafka.apache.org/downloads

wget

https://archive.apache.org/dist/kafka/2.7.0/kafka_2.13-2.7.0.tgz

tar

修改zookeeper.properties的配置文件。

修改dataDir的参数配置其他的配置默认不变。

dataDir/usr/local/kafka2.7.0/zookeeper

sudo

/usr/local/kafka2.7.0/config/zookeeper.properties$

sudo

/usr/local/kafka2.7.0/zookeeper/

sudo

修改server.properties的配置文件。

修改listeners、host.name、log.dirs、zookeeper.connect、create.topics.enable和delete.topic.enble的参数配置没有的配置添加其他的配置默认不变。

sudo

/usr/local/kafka2.7.0/config/server.properties

########

listenersPLAINTEXT://172.16.0.9:9092

host.name172.16.0.9

log.dirs/usr/local/kafka2.7.0/logs

##########

zookeeper.connect172.16.0.9:2181

##########

启动Kafka,使用root用户操作。

分为两步先启动zookeeper再启动Kafka。

[rootxx]#

/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

[rootxx]#

/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

jps

使用kafka自带的性能测试脚本发起写入MQ消息和消费MQ消息的请求。

根据不同数量级的消息写入和消息消费测试结果评估kafka处理消息的能力。

2.1

对kafka节点进行MQ消息服务的压力测试关注Kafka消息写入的延迟时间是否满足需求。

脚本命令位于/usr/local/kafka2.7.0/bin

创建topic单机环境replication-factor设置为1。

上述server.properties中的auto.create.topics.enable设置为true可以自动创建主题。

sudo

bootstrap.servers172.16.0.9:9092

100000

bootstrap.servers172.16.0.9:9092

1000000

bootstrap.servers172.16.0.9:9092

10000000

throughput设置0-1测试producer的最大吞吐量。

优化参数compression.typesnappy使用snappy算法压缩消息。

sudo

bootstrap.servers172.16.0.9:9092

batch_size563840

(org.apache.kafka.clients.producer.ProducerConfig)

[2024-03-28

(org.apache.kafka.clients.producer.ProducerConfig)

[2024-03-28

(org.apache.kafka.clients.producer.ProducerConfig)

577921

以写入100w条MQ消息为例每秒平均向kafka写入了4.77MB的数据平均4999.725条消息/秒每次写入的平均延迟为0.51毫秒最大的延迟为481毫秒。

producer优化思路与优化参数

linger.ms这两个参数是配合起来使用的目的就是缓存更多的数据减少客户端发起请求的次数。

这两个参数根据实际情况调整注意要适量。

关闭数据发送确认机制适用于对数据完整性要求不高的场景比如日志丢几条无所谓那种指定数据发送时的压缩算法默认不压缩可选压缩算法gzip,snappy,lz4,zstd等

推荐一组优化参数

设置消息总数(单位w)设置单个消息大小单位字节设置每秒发送消息数实际写入消息数/秒95%的消息延迟单位ms10100020001999.761ms100100050004999.721ms1000100050004999.961ms

2.2

对Kafka节点进行MQ消息处理的压力测试验证Kafka的消息处理能力。

sudo

bootstrap.servers172.16.0.9:9092

sudo

bootstrap.servers172.16.0.9:9092

sudo

bootstrap.servers172.16.0.9:9092

sudo

以本例中消费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每秒消费消息数消费耗时单位s1095.3671089.63193984.96241.064100953.8287.3769301336.54723.31910009536.782360.874163831.1354156.664

Kafka