0


如何在JMeter中玩懂KAFKA

  • 注意:

    本次JMeter做角色Producer,如何生产消息的。而在第二章节就会详细讲解如何搭建KAFKA集群,通过JMeter把消息发送到搭建的KAFKA上。

在JMeter配置Producer的5步走:

在安装KAFKA插件时,请一定要注意和对应的JMeter版本相对应,否则会运行失败。

以下的是JMeter5.0以上的方法:

1. 下载第三方插件源码:

https://github.com/BrightTag/kafkameterhttps://github.com/BrightTag/kafkameter

  • 2. 解压源码,使用IDE的maven -> package进行打包

打包成功会在target目录下,生成 kafkameter-0.2.0.jar和 original-kafkameter-0.2.0.jar

  • 3. 复制target/的jar到$JMeter_HOME/lib/ext

  • 4. 重新打开JMeter

  • 5. 添加Java请求取样器

Class name:

co.signal.kafkameter.KafkaProducerSampler就是引用kafkameter-0.2.0.jar里面的方法,如果没有导入jar包到./lib/ext下,就不会有这个classname

以下属性是必需的。

  • kafka_brokers: 以逗号分隔的主机列表,格式为(主机名:端口)
  • kafka_topic: kafa的topic
  • kafka_key: 消息的分区键。
  • kafka_message: kafka发送的消息体

如果有需要,也可以编辑下面的内容:

  • kafka_message_serializer: the Kafka client value.serializer property.
  • kafka_key_serializer: the Kafka client key.serializer property.
  • kafka_ssl_keystore: the keystore certificate file (with path).
  • kafka_ssl_truststore_password: the truststore certificate file password.
  • kafka_ssl_truststore: the truststore certificate file (with path).
  • kafka_ssl_keystore_password: the keystore certificate file password.
  • kafka_use_ssl: set to 'true' if you want to use SSL to connect to kafka.
  • kafka_compression_type: the compression type, for example gzip. Leave the field empty for no encryption.
  • kafka_partition: the partition number to use. Use a fixed value or calculate it on a per thread basis. Example: ${__jexl(${__threadNum} % 10)} This would spread the threads across 10 partitions. Works best if the thread numbers are a multiple of the total partitions. If you leave this field empty, the Kafka library will calculate a value based on your key.

producer 发送消息到 broker 时,会根据分区算法选择将其存储到哪一个 partition。其路由机制为:

  • 指定了 kafka_partition,则直接使用;
  • 未指定 kafka_partition 但指定 kafka_key,通过对 kafka_key 的 value 进行hash 选出一个 patition
  • kafka_partition 和 kafka_key 都未指定,使用轮询选出一个 patition。

在windows搭建简单的KAFKA环境

在windows搭建简单的KAFKA环境,理解一下其中的大致原理,加深理解。

1. 将要搭建ZeeKeeper、kafka的大致架构图

2. zeekeeper和kafka简介

ZooKeeper(动物管理员) 是一个开源的分布式协调服务,kafka基于ZooKeeper部署。

kafka是用于构建实时数据管道和流应用程序。具有横向扩展,容错,wicked fast(变态快)等优点。

3. 搭建准备

  • 下载kafka

从Kafka官网 http://kafka.apache.org/downloads 下载Kafka安装包。(要下载Binary downloads这个类型,不要下载源文件,方便使用)

本次使用 kafka_2.11-2.4.0.tgz

  • 下载ZooKeeper

从zookeeper官网 http://zookeeper.apache.org/releases.html 下载zookeeper安装包。

本次使用 apache-zookeeper-3.5.9-bin.tar.gz

  • 下载JDK

从官网 http://www.java.com/download/ 下载(建议下载Oracle官方发布的Java),,配置环境变量。

本次使用JDK1.8

4. 搭建步骤

(1)部署ZooKeeper

a、解压apache-zookeeper-3.5.9-bin.tar.gz

b、在目录apache-zookeeper-3.5.9-bin\conf重命名zoo_sample.cfg为zoo.cfg

c、打开zoo.cfg文件,修改dataDir的值路径,路径可以自己定义。data的文件夹需要先创建。例如下面的dataDir=F:/kafka/zookeeper/data

... ...
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
# dataDir=/tmp/zookeeper
dataDir=F:/kafka/zookeeper/data
# the port at which the clients will connect
clientPort=2181
... ...

d、到此,ZooKeeper安装完毕

(2)部署kafka

a、新建两个文件夹,一个是kafka,另外一个是kafka2

b、解压 kafka_2.11-2.4.0.tgz,解压内容一份放置在kafka文件夹下,另外一份放置在kafka2文件夹下

c、分别在kafka和kafka2文件夹下,新建kafka-logs文件夹,给下面配置使用

d、修改kafka目录中kafka_2.11-2.4.0\config\server.properties文件内容。样例如下:

    1、新增broker.list=localhost:9093,localhost:9092

    2、新增port=9092

    3、新增host.name=localhost
... ...
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
broker.list=localhost:9093,localhost:9092
#listener and port
port=9092
#broker 绑定的主机名称(IP) 如果不设置将绑定所有的接口。
host.name=localhost
############################# Socket Server Settings #############################
... ...

e、修改kafka2目录中kafka_2.11-2.4.0\config\server.properties文件内容。样例如下:

    1、修改broker.id=1

    2、新增broker.list=localhost:9093,localhost:9092

    3、新增port=9093

    4、新增host.name=localhost
... ...
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
broker.list=localhost:9093,localhost:9092
#listener and port
port=9093
#broker 绑定的主机名称(IP) 如果不设置将绑定所有的接口。
host.name=localhost
############################# Socket Server Settings #############################
... ...

f、到此kafka安装完毕

5. 启动服务

(1)启动ZooKeeper(必须先起来)

进入zookeeper的bin目录,执行zkServer.cmd(打开命令窗口)

(2)启动kafka和kafka2

进入kafka和kafka2目录中kafka_2.11-2.4.0\bin\windows (执行命令都需要进入该目录,下面的操作步骤就省略这个说明),执行命令

kafka-server-start.bat ....\config\server.properties

6. 创建TOPIC

进入kafka或kafka2目录中kafka_2.11-2.4.0\bin\windows,执行命令,创建一个topic为Testfactor2partitions2

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 2 --partitions 2 --topic Testfactor2partitions2

现在我们一共有两台broker,如果--replication-factor 多于2的话,就会报异常“ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 4 larger than available brokers: 2.”

再创建一个topic为Testfactor2partitions99

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 2 --partitions 99 --topic Testfactor2partitions99

查询toipc:进入kafka或kafka2目录中kafka_2.11-2.4.0\bin\windows,执行命令

kafka-topics.bat --list --zookeeper localhost:2181

现在情况如下:

7.使用JMeter发起Producer调试

(1)消费Testfactor2partitions2主题消息

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic Testfactor2partitions2 --from-beginning

(2)消费Testfactor2partitions99主题消息

kafka-console-consumer.bat --bootstrap-server localhost:9093 --topic Testfactor2partitions99 --from-beginning

(3)Jmerter的java请求体生产Testfactor2partitions2主题消息

kafka_brokers=localhost:9092,localhost:9093(kafka服务器)

kafka_topic=Testfactor2partitions2

kafka_message=Producer1(TOPIC): Testfactor2partitions2

如下截图:

点击JMeter的发送,那么a步骤的消费者就会消费这条消息了。如下截图:

如果传中文就会有乱码,需要另外配置处理。

(4)Jmerter的java请求体生产Testfactor2partitions99主题消息

kafka_brokers=localhost:9092,localhost:9093(kafka服务器)

kafka_topic=Testfactor2partitions99

kafka_message=Producer1(TOPIC): Testfactor2partitions99

如下截图:

(5)注意点

如果java请求体的kafka_partition字段传的值,大于创建主题时的--partitions xx时,就会报异常“java.lang.IllegalArgumentException: Invalid partition given with record: 100 is not in the range [0...99].”

总结

    搭建简单的kafka环境,并且也实现生产消费消息场景。zookeeper会把kafka信息存储到data文件内,kafka会把topic消息信息存储到kafka-logs内。

    如果要格式化kafka,就必须得把zookeeper目录下的data文件内的文件删掉,还要把kafka和kafka2目录下的kafka-logs文件内的文件删掉。

    如果停掉kafka时,除了关闭命令窗口外,也可以执行kafka-server-stop.bat命令。

思考: 虽然是本地的地址,但是可以思考,如果想消费其他服务器的消息。如果想向其他服务器生产消息

标签: jmeter kafka 分布式

本文转载自: https://blog.csdn.net/qq_38209830/article/details/138368644
版权归原作者 强叔的自动化测试 所有, 如有侵权,请联系我们删除。

“如何在JMeter中玩懂KAFKA”的评论:

还没有评论