安装Kafka：

我们使用3台机器搭建Kafka集群：

192.168.121.132   h201

192.168.121.131   h202

192.168.121.130   h203

 

下载kafka_2.10-0.10.1.0.tgz

[hadoop@h201 ~]$ tar -zxvf kafka_2.10-0.10.1.0.tgz

[hadoop@h201 ~]$ cd kafka_2.10-0.10.1.0

修改配置文件kafka_2.10-0.10.1.0/config/server.properties，修改如下内容为：

zookeeper.connect=h201:2181,h202:2181,h203:2181/kafka

这里需要说明的是，默认Kafka会使用ZooKeeper默认的/路径，这样有关Kafka的ZooKeeper配置就会散落在根路径下面，如果你有其他的应用也在使用ZooKeeper集群，查看ZooKeeper中数据可能会不直观，所以强烈建议指定一个chroot路径，直接在zookeeper.connect配置项中指定。

使用如下命令连接到任意一台ZooKeeper服务器：

[hadoop@h201 ~]$ cd zookeeper-3.4.5-cdh5.5.2/

[hadoop@h201 zookeeper-3.4.5]$ bin/zkCli.sh

connect h201:2181

create /kafka

ls /kafka/brokers/topics

ls /kafka/brokers/ids

将kafka拷贝到其他两台机器：

[hadoop@h201 ~]$ scp -r kafka_2.10-0.10.1.0/ h202:/home/hadoop/

[hadoop@h201 ~]$ scp -r kafka_2.10-0.10.1.0/ h203:/home/hadoop/

 

修改server.properties中的broker.id，三台机器都修改

因为Kafka集群需要保证各个Broker的id在整个集群中必须唯一，需要调整这个配置项的值（如果在单机上，可以通过建立多个Broker进程来模拟分布式的Kafka集群，但也需要Broker的id唯一，还需要修改一些配置目录的信息）。

[hadoop@h201 ~]$ vi kafka_2.10-0.10.1.0/config/server.properties

broker.id=0

[hadoop@h202 ~]$ vi kafka_2.10-0.10.1.0/config/server.properties

broker.id=1

[hadoop@h203 ~]$ vi kafka_2.10-0.10.1.0/config/server.properties

broker.id=2

首先启动zookeeper集群，然后三台机器都执行以下命令：

kafka-server-start.sh -daemon /home/hadoop/kafka_2.10-0.10.1.0/config/server.properties

kafka-server-stop.sh -daemon /home/hadoop/kafka_2.10-0.10.1.0/config/server.properties

创建topic:

bin/kafka-topics.sh --create --zookeeper h201:2181/kafka --replication-factor 3 --partitions 5 --topic data

查看创建的topic:

kafka-topics.sh --describe --zookeeper h201:2181/kafka --topic data

kafka-topics.sh --zookeeper h201:2181/kafka --list

上面Leader、Replicas、Isr的含义如下：

Partition： 分区

Leader   ： 负责读写指定分区的节点

Replicas ： 复制该分区log的节点列表

Isr      ： "in-sync" replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader

删除topic：

在kafka集群的所有三台机器中修改配置文件server.properties

添加如下配置：delete.topic.enable=true

kafka-topics.sh --delete --zookeeper h201:2181/kafka --topic data

自带压力测试

kafka-console-producer.sh --broker-list h201:9092 --topic data

kafka-console-consumer.sh --zookeeper h201:2181/kafka --topic data --from-beginning

生产者：throughput吞吐量

[hadoop@h202 config]$ kafka-producer-perf-test.sh --num-records 100000 --record-size 3000 --topic data --producer-props bootstrap.servers=h201:9092,h202:9092,h203:9092 acks=all --record-size 1000 --throughput 1000

SLF4J: Class path contains multiple SLF4J bindings.

SLF4J: Found binding in [jar:file:/home/hadoop/work/hadoop2.6jar/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: Found binding in [jar:file:/home/hadoop/kafka_2.10-0.10.1.0/libs/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

4399 records sent, 860.2 records/sec (0.82 MB/sec), 64.0 ms avg latency, 725.0 max latency.

2857 records sent, 513.4 records/sec (0.49 MB/sec), 237.0 ms avg latency, 3464.0 max latency.

4461 records sent, 892.0 records/sec (0.85 MB/sec), 3908.1 ms avg latency, 6209.0 max latency.

7809 records sent, 1561.8 records/sec (1.49 MB/sec), 2653.6 ms avg latency, 5225.0 max latency.

6177 records sent, 1235.2 records/sec (1.18 MB/sec), 961.4 ms avg latency, 2343.0 max latency.

4991 records sent, 976.7 records/sec (0.93 MB/sec), 218.8 ms avg latency, 1001.0 max latency.

4444 records sent, 843.1 records/sec (0.80 MB/sec), 390.9 ms avg latency, 2136.0 max latency.

5867 records sent, 1169.0 records/sec (1.11 MB/sec), 149.5 ms avg latency, 1086.0 max latency.

4713 records sent, 810.4 records/sec (0.77 MB/sec), 95.1 ms avg latency, 1304.0 max latency.

5235 records sent, 1046.8 records/sec (1.00 MB/sec), 1083.5 ms avg latency, 2910.0 max latency.

6041 records sent, 1122.9 records/sec (1.07 MB/sec), 411.3 ms avg latency, 1252.0 max latency.

5402 records sent, 1075.5 records/sec (1.03 MB/sec), 92.3 ms avg latency, 471.0 max latency.

5019 records sent, 1003.2 records/sec (0.96 MB/sec), 46.6 ms avg latency, 331.0 max latency.

4971 records sent, 994.2 records/sec (0.95 MB/sec), 142.3 ms avg latency, 677.0 max latency.

5026 records sent, 1005.0 records/sec (0.96 MB/sec), 76.9 ms avg latency, 555.0 max latency.

4461 records sent, 712.4 records/sec (0.68 MB/sec), 858.5 ms avg latency, 2328.0 max latency.

6240 records sent, 1247.8 records/sec (1.19 MB/sec), 596.2 ms avg latency, 2434.0 max latency.

5564 records sent, 1112.6 records/sec (1.06 MB/sec), 306.2 ms avg latency, 1089.0 max latency.

4998 records sent, 999.6 records/sec (0.95 MB/sec), 110.0 ms avg latency, 1215.0 max latency.

100000 records sent, 999.360409 records/sec (0.95 MB/sec), 692.04 ms avg latency, 6209.00 ms max latency, 156 ms 50th, 4165 ms 95th, 5337 ms 99th, 5761 ms 99.9th.

消费者：

[hadoop@h201 ~]$ kafka-consumer-perf-test.sh --zookeeper h201:2181/kafka --messages 100000 --topic data --threads 3

start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec

SLF4J: Class path contains multiple SLF4J bindings.

SLF4J: Found binding in [jar:file:/home/hadoop/work/hadoop2.6jar/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: Found binding in [jar:file:/home/hadoop/kafka_2.10-0.10.1.0/libs/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

2018-07-08 22:40:31:658, 2018-07-08 22:40:34:270, 9.8706, 3.7789, 3460, 1324.6554

1、 生产者测试结果及分析

结果分析：

1）kafka在批处理，多线程，不适用同步复制的情况下，吞吐率是比较高的，可以达26MB/s,消息数达17w条/s以上。

2）使用批处理或多线程对提升生产者吞吐率效果明显。

3）复制因子会对吞吐率产生较明显影响

使用同步复制时，复制因子会对吞吐率产生较明显的影响。复制因子为2比因子为1（即无复制）时，吞吐率下降40%左右。

4）使用sync方式，性能有明显下降。

使用Sync方式producer吞吐率会有明显下降

5)压缩与吞吐率

使用Gzip及Snappy方式压缩，吞吐率反而有下降，原因待分析。而Snappy方式吞吐率高于gzip方式。

6）分区数与吞吐率

分区数增加生产者吞吐率反而有所下降

 

 

2、消费者结果及分析

结果分析：

1）kafka consumer吞吐率在parition，threads较大的情况下，在测试场景下，最大吞吐率达到了34MB/s

2）复制因子，影响较小。replication factor并不会影响consumer的吞吐率测试， consumer从每个partition的leader读数据，而与replication factor无关。同样，consumer吞吐率也与同步复制还是异步复制无关。

3)线程数和partition与吞吐率关系

当分区数较大时，增加thread数可显著提升consumer的吞吐率。

但要注意在分区较大时线程数不改大于分区数，否则会出现No broker partitions consumed by consumer，对提升吞吐率也没有帮助。

4）批处理数对吞吐率影响

改变批处理数对吞吐率影响不大

5）压缩与吞吐率

压缩对吞吐率影响小。

 

附优化后的配置文件：

broker.id=1

listeners=PLAINTEXT://0.0.0.0:6667

advertised.listeners=PLAINTEXT://203.150.54.215:6667

port=6667

host.name=203.150.54.215

 

# Replication configurations

num.replica.fetchers=1

replica.fetch.max.bytes=1048576

replica.fetch.wait.max.ms=500

replica.high.watermark.checkpoint.interval.ms=5000

replica.socket.timeout.ms=30000

replica.socket.receive.buffer.bytes=65536

replica.lag.time.max.ms=10000

replica.lag.max.messages=4000

 

compression.codec:none

controller.socket.timeout.ms=30000

controller.message.queue.size=10

controlled.shutdown.enable=true

default.replication.factor:2

 

# Log configuration

num.partitions=1

num.recovery.threads.per.data.dir=1

message.max.bytes=1000000

auto.create.topics.enable=true

auto.leader.rebalance.enable=true

log.dirs=/mnt/kafka-logs/kafka00

log.index.interval.bytes=4096

log.index.size.max.bytes=10485760

log.retention.hours=72 #保留三天，也可以更短

log.flush.interval.ms=10000 #每间隔1秒钟时间，刷数据到磁盘

log.flush.interval.messages=20000 #log数据文件刷新策略

log.flush.scheduler.interval.ms=2000

log.roll.hours=72

log.retention.check.interval.ms=300000

log.segment.bytes=1073741824 #kafka启动时是单线程扫描目录(log.dir)下所有数据文件

 

# ZK configuration

zookeeper.connection.timeout.ms=6000

zookeeper.sync.time.ms=2000

zookeeper.connect=203.150.54.215:2181,203.150.54.216:2182,203.150.54.217:2183

 

# Socket server configuration

num.io.threads=5 #配置线程数量为cpu核数加1

num.network.threads=8 #配置线程数量为cpu核数2倍，最大不超过3倍

socket.request.max.bytes=104857600

socket.receive.buffer.bytes=1048576

socket.send.buffer.bytes=1048576

queued.max.requests=500

fetch.purgatory.purge.interval.requests=1000

producer.purgatory.purge.interval.requests=1000