本文中包含了一套kafka叢集的部署、配置、除錯和壓測的技術方法。

在三個主機節點上進行部署。 server1：192.168.10.1 server2：192.168.10.2 server3：192.168.10.3 1、jdk7u80的安裝與配置 rpm -ivh jdk-7u80-linux-x64.rpm 配置環境變數：

more /etc/profile
JAVA_HOME=/usr/java/jdk1.7.0_80
PATH=\$JAVA_HOME/bin:\$PATH:.
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export PATH
export CLASSPATH
 

注：低版本jdk在執行kafka時存在bug。 2、系統iptables防火牆對區域網段開放以下埠

-A INPUT -s 192.168.10.0/24 -m state --state NEW -m tcp -p tcp --dport 2888 -j ACCEPT
-A INPUT -s 192.168.10.0/24 -m state --state NEW -m tcp -p tcp --dport 3888 -j ACCEPT
-A INPUT -s 192.168.10.0/24 -m state --state NEW -m tcp -p tcp --dport 9092 -j ACCEPT
-A INPUT -s 192.168.10.0/24 -m state --state NEW -m tcp -p tcp --dport 2181 -j ACCEPT
 

3、叢集節點間主機名解析配置

more /etc/hosts
192.168.10.1    server1
192.168.10.2    server2
192.168.10.3    server3

4、在三個主機節點上部署kafka-2.11如下

cd /data
unzip kafka_2.11-0.10.0.0.zip
mv kafka_2.11-0.10.0.0/ kafka

5、配置zookeeper叢集 注：以下除特別說明在哪個節點進行配置外，均需要修改三個主機節點。 因為該zookeeper是專服務於kafka的，所以直接把其資料目錄放置於/data/kafka/zookeeper，便於後續管理。 mkdir -p /data/kafka/zookeeper 編輯zookeeper配置檔案：

cd /data/kafka
vi config/zookeeper.properties

tickTime=2000
dataDir=/data/kafka/zookeeper
clientPort=2181
maxClientCnxns=0
initLimit=15
syncLimit=5
server.1=192.168.10.1:2888:3888
server.2=192.168.10.2:2888:3888
server.3=192.168.10.3:2888:3888

建立ServerID標識： 節點server1：echo "1" > /data/kafka/zookeeper/myid 節點server2：echo "2" > /data/kafka/zookeeper/myid 節點server3：echo "3" > /data/kafka/zookeeper/myid 注：這裡設定的myid取值需要和zookeeper.properties中“server.id”保持一致。 chmod +x zookeeper-server-start.sh zookeeper-server-stop.sh kafka-run-class.sh 修改zookeeper啟動指令碼如下，以便於管理：

$ more zookeeper-server-start.sh
#!/bin/bash
#if [ $# -lt 1 ];
#then
#    echo "USAGE: $0 [-daemon] zookeeper.properties"
#    exit 1
#fi
base_dir=$(dirname $0)
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"
fi
EXTRA_ARGS="-name zookeeper -loggc"
#COMMAND=$1
COMMAND="-daemon"
case $COMMAND in
  -daemon)
     EXTRA_ARGS="-daemon "$EXTRA_ARGS
     shift
     ;;
 *)
     ;;
esac
#exec $base_dir/kafka-run-class.sh $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain "[email protected]"
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain "../config/zookeeper.properties"

啟停zookeeper的方法：

cd /data/kafka/bin
./zookeeper-server-start.sh
./zookeeper-server-stop.sh

6、配置kafka叢集 配置/data/kafka/config/server.properties如下。三個主機節點上配置檔案中僅前面幾行的引數取值不同。

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
############################# Socket Server Settings #############################
listeners=PLAINTEXT://192.168.10.1:9092
port=9092
host.name=192.168.10.1
# The number of threads handling network requests
num.network.threads=8
# The number of threads doing disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=1048576
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=1048576
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
# The number of queued requests allowed before blocking the network threads
queued.max.requests=100
# The purge interval (in number of requests) of the fetch request purgatory
fetch.purgatory.purge.interval.requests=200
# The purge interval (in number of requests) of the producer request purgatory
producer.purgatory.purge.interval.requests=200

############################# Log Basics #############################
# A comma seperated list of directories under which to store log files
log.dirs=/data/kafka/kafka-logs
# The default number of log partitions per topic. 
num.partitions=24
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
num.recovery.threads.per.data.dir=2
# The maximum size of message that the server can receive
message.max.bytes=1000000
# Enable auto creation of topic on the server
auto.create.topics.enable=true
# The interval with which we add an entry to the offset index
log.index.interval.bytes=4096
# The maximum size in bytes of the offset index
log.index.size.max.bytes=10485760
# Allow to delete topics
delete.topic.enable=true
############################# Log Flush Policy #############################
# The number of messages to accept before forcing a flush of data to disk
log.flush.interval.messages=20000
# The maximum amount of time a message can sit in a log before we force a flush
log.flush.interval.ms=10000
# The frequency in ms that the log flusher checks whether any log needs to be flushed to disk
log.flush.scheduler.interval.ms=2000
############################# Log Retention Policy #############################
# The minimum age of a log file to be eligible for deletion
log.retention.hours=168
# A size-based retention policy for logs. 
log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
# The maximum time before a new log segment is rolled out (in hours)
log.roll.hours=168
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
zookeeper.connect=192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
# How far a ZK follower can be behind a ZK leader
zookeeper.sync.time.ms=2000

############################# Replication configurations ################
# default replication factors for automatically created topics
default.replication.factor=3
# Number of fetcher threads used to replicate messages from a source broker.
num.replica.fetchers=4
# The number of bytes of messages to attempt to fetch for each partition.
replica.fetch.max.bytes=1048576
# max wait time for each fetcher request issued by follower replicas. 
replica.fetch.wait.max.ms=500
# The frequency with which the high watermark is saved out to disk
replica.high.watermark.checkpoint.interval.ms=5000
# The socket timeout for network requests.
replica.socket.timeout.ms=30000
# The socket receive buffer for network requests
replica.socket.receive.buffer.bytes=65536
# If a follower hasn't sent any fetch requests or hasn't consumed up to the leaders log end offset for at least this time, the leader will remove the follower from isr
replica.lag.time.max.ms=10000
# The socket timeout for controller-to-broker channels
controller.socket.timeout.ms=30000
controller.message.queue.size=10

7、配置kafka生產者和消費者 修改/data/kafka/config下的producer.properties檔案 bootstrap.servers=192.168.10.1:9092,192.168.10.2:9092,192.168.10.3:9092 producer.type=async compression.type=snappy 修改/data/kafka/config下的comsumer.properties檔案 zookeeper.connect=192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181 8、kafka叢集服務啟停管理的配置

chmod +x bin/kafka-server-start.sh kafka-server-stop.sh
修改kafka-server-start.sh
more kafka-server-start.sh
#!/bin/bash
#if [ $# -lt 1 ];
#then
#    echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
#    exit 1
#fi
base_dir=$(dirname $0)

if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx5G -Xms5G"
fi

EXTRA_ARGS="-name kafkaServer -loggc"

#COMMAND=$1
COMMAND="-daemon"
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift
    ;;
  *)
    ;;
esac

#exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "[email protected]"
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "../config/server.properties"

啟停kafka服務的方法： cd /data/kafka/bin ./kafka-server-start.sh ./kafka-server-stop.sh 注：觀察和檢查/data/kafka/logs下的各個日誌檔案，以確認無報錯資訊且各項服務日誌輸出正確。 9、kafka叢集的常用管理命令 建議另外安排一個備機作為長期監控和測試kafka叢集的主機。程式部署在/data/kafka下。 建立了以下用於服務和效能監控的topic： ./kafka-topics.sh --create --zookeeper 192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181 --replication-factor 3 --partitions 6 --topic test-for-sys-monitor 檢視已建立了的topics列表： ./kafka-topics.sh --list --zookeeper 192.168.10.1:2181 檢視指定topic的詳情： ./kafka-topics.sh --describe --zookeeper 192.168.10.1:2181 --topic test-for-sys-monitor 使用終端生產者命令進行測試： ./kafka-console-producer.sh --broker-list 192.168.10.1:9092,192.168.10.2:9092,192.168.10.3:9092   --topic test-for-sys-monitor 使用終端消費者命令進行測試： ./kafka-console-consumer.sh --zookeeper 192.168.10.1:2181 --topic test-for-sys-monitor 注：以上測試，在生產者側輸入的資料，會在消費者側打印出來。 刪除topic的命令： ./kafka-topics.sh --delete --zookeeper 192.168.10.1:2181  --topic test-for-sys-monitor 注：kafka叢集中放開了刪除topic的功能，請謹慎使用。 壓測寫500萬條資料，每條1KB： ./kafka-producer-perf-test.sh --topic test-perf-20161220 --num-records 500000 --record-size 1000 --throughput 100000 --producer-props bootstrap.servers=192.168.10.1:9092,192.168.10.2:9092,192.168.10.3:9092 5000000 records sent, 53149.648149 records/sec (50.69 MB/sec), 569.30 ms avg latency, 2096.00 ms max latency, 8 ms 50th, 1759 ms 95th, 1874 ms 99th, 2044 ms 99.9th. 壓測讀6*100萬條資料：  ./kafka-consumer-perf-test.sh --zookeeper 192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181 --messages 1000000 --topic test-perf-20161220 --threads 6 start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec 2016-12-20 16:08:50:102, 2016-12-20 16:09:48:525, 5722.0459, 97.9417, 6000000, 102699.2794 注：壓測後，請及時刪除用於壓測的topic，因為所產生的kafka日誌資料很可觀。 登入zookeeper shell中檢視kafka建立的相關元資料：

./zookeeper-shell.sh 192.168.10.3:2181
Connecting to 192.168.10.3:2181
Welcome to ZooKeeper!
JLine support is disabled
WATCHER::
WatchedEvent state:SyncConnected type:None path:null
ls /
     [consumers, config, controller, isr_change_notification, brokers, admin, zookeeper, controller_epoch]
get /brokers/ids/1
{"jmx_port":-1,"timestamp":"1482223274389","endpoints":["PLAINTEXT://192.168.10.1:9092"],"host":"192.168.10.1","version":3,"port":9092}
cZxid = 0x900000009
ctime = Tue Dec 20 16:41:14 CST 2016
mZxid = 0x900000009
mtime = Tue Dec 20 16:41:14 CST 2016
pZxid = 0x900000009
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1591b61880f0000
dataLength = 137
numChildren = 0
get /brokers/ids/2
{"jmx_port":-1,"timestamp":"1482223307625","endpoints":["PLAINTEXT://192.168.10.2:9092"],"host":"192.168.10.2","version":3,"port":9092}
cZxid = 0x900000013
ctime = Tue Dec 20 16:41:47 CST 2016
mZxid = 0x900000013
mtime = Tue Dec 20 16:41:47 CST 2016
pZxid = 0x900000013
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1591b61880f0001
dataLength = 137
numChildren = 0
get /brokers/ids/3
{"jmx_port":-1,"timestamp":"1482223315746","endpoints":["PLAINTEXT://192.168.10.3:9092"],"host":"192.168.10.3","version":3,"port":9092}
cZxid = 0x900000020
ctime = Tue Dec 20 16:41:55 CST 2016
mZxid = 0x900000020
mtime = Tue Dec 20 16:41:55 CST 2016
pZxid = 0x900000020
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x3591b618ffe0001
dataLength = 137
numChildren = 0
quit
     Quitting...

改變topic的分割槽數量 ./kafka-topics.sh --alter --zookeeper 192.168.10.1:2181 --topic test_topic --partitions 4 增刪改topic的配置引數 ./kafka-topics.sh --alter --zookeeper 192.168.10.1:2181 --topic test_topic--config key=value ./kafka-topics.sh —alter --zookeeper 192.168.10.1:2181 --topic test_topic--deleteConfig key 檢視topic的分割槽、副本狀態，需要關注各個分割槽的負載分佈是否均衡（即Leader角色的分佈）：

./kafka-topics.sh --describe --zookeeper 192.168.10.1:2181 --topic test-for-sys-monitor
Topic:test-for-sys-monitor    PartitionCount:6    ReplicationFactor:3    Configs:
    Topic: test-for-sys-monitor    Partition: 0    Leader: 1    Replicas: 1,3,2    Isr: 1,3,2
    Topic: test-for-sys-monitor    Partition: 1    Leader: 2    Replicas: 2,1,3    Isr: 2,1,3
    Topic: test-for-sys-monitor    Partition: 2    Leader: 3    Replicas: 3,2,1    Isr: 3,2,1
    Topic: test-for-sys-monitor    Partition: 3    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
    Topic: test-for-sys-monitor    Partition: 4    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    Topic: test-for-sys-monitor    Partition: 5    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2

執行leader分佈的再平衡： ./kafka-preferred-replica-election.sh --zookeeper 192.168.10.1:2181 注：kafka本身是會自動做leader分佈再平衡工作的，但不會是發現問題後立即執行，會有半小時的延遲。 注：很多配置引數，除非明確理解其作用，否則不必配置，因為kafka自身是對這些引數設定了預設值的，這些預設設定已經是一個比較好的選擇。