Kafka集羣部署

部署ZooKeeper集羣

Kafka依賴ZooKeeper，因此須要先部署ZooKeeper集羣。

• 環境準備：

主機名	IP
test1	192.168.30.128
test2	192.168.30.129
test3	192.168.30.130

• 所有設置hosts：

vim /etc/hosts

192.168.30.128 test1
192.168.30.129 test2
192.168.30.130 test3

• 所有關閉selinux和firewalld：

setenforce 0 && sed -i 's/=enforcing/=disabled/g' /etc/selinux/config

systemctl stop firewalld && systemctl disable firewalld

• 所有安裝java環境：

tar zxf jdk-8u191-linux-x64.tar.gz && mv jdk1.8.0_191/ /usr/local/jdk

vim /etc/profile

JAVA_HOME=/usr/local/jdk
PATH=$PATH:$JAVA_HOME/bin:$JAVA_HOME/jre/bin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/libexport JAVA_HOME PATH CLASSPATHsource !$

java -versionln -s /usr/local/jdk/bin/java /usr/bin/java

• 所有下載ZooKeeper：

mkdir /software && cd /softwarewget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gztar zxf zookeeper-3.4.14.tar.gzmv zookeeper-3.4.14 /usr/local/zookeeper

• 所有建立數據、日誌目錄及當前節點ID：

mkdir /usr/local/zookeeper/datamkdir /usr/local/zookeeper/dataLogecho 1 > /usr/local/zookeeper/data/myid                 #集羣每臺機器此ID不一樣便可

建議test2修改/usr/local/zookeeper/data/myid爲2，test3上修改成3

• 所有修改配置：

cp /usr/local/zookeeper/conf/zoo_sample.cfg /usr/local/zookeeper/conf/zoo.cfg

vim /usr/local/zookeeper/conf/zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper/data
dataLogDir=/usr/local/zookeeper/dataLog
clientPort=2181
maxClientCnxns=1024#集羣節點信息，能夠是ip或主機名server.1=test1:2888:3888
server.2=test2:2888:3888
server.3=test3:2888:3888

tickTime：Zookeeper中最小時間單元的長度

initLimit：默認值爲10，表示是tickTime的10倍。Leader服務器等待Follower啓動，並完成數據同步的時間

syncLimit：默認值爲5，表示是tickTime的5倍。Leader服務器和Follower之間進行心跳檢測的最大延時時間

dataDir：Zookeeper服務器存儲快照文件的目錄

dataLogDir：ZooKeeper日誌目錄

clientPort：Zookeeper對外的服務端口，集羣中無須保持一致

maxClientCnxns：默認是60，從Socket層面限制單個客戶端與單臺服務器之間的併發鏈接數

• 所有時間同步：

yum install -y ntpdateecho "*/5 * * * * /usr/sbin/ntpdate time.windows.com &>/dev/null" >> /var/spool/cron/root

• 所有啓動ZooKeeper：

/usr/local/zookeeper/bin/zkServer.sh startnetstat -lntp |grep java                #檢查端口，擁有2888端口爲leader

爲了後續啓動方便，能夠作個軟連接

ln -s /usr/local/zookeeper/bin/zkServer.sh /usr/local/bin/zookeeper

• 測試鏈接ZooKeeper：

/usr/local/zookeeper/bin/zkCli.sh -server test1:2181[zk: test1:2181(CONNECTED) 0] ls /[zookeeper]             #當前只有zookeeper一個節點[zk: test1:2181(CONNECTED) 1] getAcl /zookeeper'world,'anyone: cdrwa

鏈接沒有問題

---

ZooKeeper常見用法

首先進入到ZooKeeper命令行模式下，能夠進行以下操做：

1. 查詢節點

ls /

1. 建立節點

create /test_node 'test node' #說明：節點名稱必須以/開頭，test_node爲節點名稱，'test node'爲具體數據#建立臨時節點create -e /test_node2 'ephemeral node'#建立順序節點，它會自動加上一堆數字create -s /s_node 'sequential node'#建立臨時順序節點create -e -s /e_s_node 'ephemeral and sequential node'

1. 查看節點狀態

stat /test_node

1. 查看節點數據內容

get /test_node

1. 設置節點數據

set /test_node 'update node data' 10#說明：最後面的數字是版本號

1. 刪除節點

delete /test_node#注意：若是 /test_node 下有子節點，則刪除會報錯。可以使用遞歸刪除命令rmr /test_node

1. 設置節點ACL

setAcl /test_node ip:192.168.30.129:cdrwa#cd權限用於控制子節點，rwa權限用於控制節點自己#c爲建立，d爲刪除；r爲讀，w爲寫，a爲admin

1. 獲取節點ACL

getAcl /test_node

更多命令參考這裏：https://blog.csdn.net/xyang81/article/details/53053642

---

部署Kafka集羣

• 所有下載Kafka：

cd /softwarewget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.1.1/kafka_2.11-2.1.1.tgztar zxf kafka_2.11-2.1.1.tgzmv kafka_2.11-2.1.1 /usr/local/kafka

• 所有修改配置：

mkdir /usr/local/kafka/logs

vim /usr/local/kafka/config/server.properties

broker.id=1                 #ID惟一，建議與zookeeper的myid對應port=9092
host.name=192.168.30.128               #本機ipnum.network.threads=3               #borker進行網絡處理的線程數num.io.threads=8                #borker進行I/O處理的線程數log.dirs=/usr/local/kafka/logs              #消息存放目錄，不是日誌目錄num.partitions=1                #每一個topic的默認分區數log.retention.hours=168             #消息過時時間，默認爲1周default.replication.factor=2                #kafka保存消息的副本數log.cleaner.enable=false                #是否啓用log壓縮，通常不用啓用，啓用的話能夠提升性能zookeeper.connect=test1:2181,test2:2181,test3:2181                #zookeeper主機名（ip）及端口

• 所有啓動Kafka：

/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.propertiesnetstat -lntp |grep 9092

• 測試：

將test1做爲生產者，test3做爲消費者

#test1上執行#建立一個topic lzx：一個分區，兩個副本/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper test1:2181 --replication-factor 2 --partitions 1 --topic lzx              #注意：factor大小不能超過broker的個數Created topic "lzx".#建立一個生產者（消息發佈者）/usr/local/kafka/bin/kafka-console-producer.sh --broker-list test1:9092 --topic lzx

#test3上執行#建立一個消費者（消息訂閱者）/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server test1:9092  --topic lzx --from-beginning              # --from-beginning表示從開始接收，不然只接收新產生的消息

能夠看到，消息生產和消費沒有問題，Kafka集羣部署完成。

---

Kafka經常使用命令

1. 查看topic

/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181

1. 查看topic test詳情

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

1. 刪除topic test(需設置參數delete.topic.enable=true)

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test

1. 生產者參數查看

/usr/local/kafka/bin/kafka-console-producer.sh

1. 消費者參數查看

/usr/local/kafka/bin/kafka-console-consumer.sh