注意:
本次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命令。
思考: 虽然是本地的地址,但是可以思考,如果想消费其他服务器的消息。如果想向其他服务器生产消息
版权归原作者 强叔的自动化测试 所有, 如有侵权,请联系我们删除。