linux下安裝kafka,首先確保zookeeper已經安裝成功,想要安裝zookeeper，必須先在linux中安裝好jdk。
下面我們依次開始安裝jdk，zookeeper，kafka。
一：安裝jdk：
1.檢查linux環境是否安裝了jdk可以用java -version或者rpm -qa|grep jdk 。
2.下載jdk-8u221-linux-x64.tar.gz。可以在官網上下載，也可以從百度雲盤中下載，百度雲盤中有jdk，zookeeper，kafka安裝包
https://pan.baidu.com/s/1JjAS4VXRVzPkwuxylJnk4w 提取碼：79i1 。
3.下載到本地后用rz命令把安裝包傳到linux服務器的usr/local/java目錄下。
4.然后用解壓縮命令解壓縮tar -zxvf jdk-8u221-linux-x64.tar.gz。
5.配置jdk環境變量：編輯/etc/下的profile文件： vi /etc/profile
在文件末尾添加如下配置：
export JAVA_HOME=/usr/local/java/jdk1.8.0_162
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH
標紅色路徑為jdk解壓路徑，根據實際情況配置，保存配置文件退出。
6.使/etc/profile生效 ：source /etc/profile
7.測試jdk安裝是否成功： java -version
二：安裝zookeeper：
1.可以wget下載zookeeper-3.4.13.tar.gz，也可以下載到本機安裝包，然后用rz命令傳到服務器/usr/local/zookeeper路徑下。
2.解壓縮，用解壓縮命令，tar -zxvf zookeeper-3.4.13.tar.gz解壓。
3.編輯配置文件前看一下本機ip，及端口號占用。netstat -tunlp ，用於顯示tcp，udp的端口和進程等相關情況。netstat -tunlp|grep 端口號，用於查看指定端口號的進程情況。
hostname命令可以看本服務器ip地址。查看/tmp/zookeeper/data，/tmp/zookeeper/log目錄，如果沒有用mkdir命令創建。
10.0.66.50服務器：創建myid文件：server ID和myid對應（三個服務器分別寫入1，2，3）
echo 1 >> /tmp/zookeeper/data/myid
10.0.47.32服務器：創建myid文件：server ID和myid對應（三個服務器分別寫入1，2，3）
echo 2 >> /tmp/zookeeper/data/myid
10.0.66.65服務器：創建myid文件：server ID和myid對應（三個服務器分別寫入1，2，3）
echo 3 >> /tmp/zookeeper/data/myid
4.編輯配置文件：進入conf目錄：cd zookeeper-3.4.13/conf，
將zoo_sample.cfg這個文件復制為zoo.cfg (必須是這個文件名)：cp zoo_sample.cfg zoo.cfg，
進入zoo.cfg文件進行編輯：vim zoo.cfg，
按 i 進入編輯模式，修改以下內容（修改之前看端口號是否被占用）：
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper/data
dataLogDir=/tmp/zookeeper/log
clientPort=2181
server.1=10.0.66.50:2888:3888
server.2=10.0.47.32:2888:3888
server.3=10.0.66.65:2888:3888
參數說明
tickTime：指 zookeeper 服務器之間或客戶端與服務器之間維持心跳的時間間隔。
initLimit：用來指定zookeeper 集群中leader接受follower初始化連接時最長能忍受的心跳時間間隔數。若超過 指定個心跳的時間間隔leader還沒有收到follower的響應，表明follower連接失敗。
syncLimit：指定 leader與follower之間請求和應答時間最長不能超過多少個 tickTime 。
dataDir：快照日志的存儲路徑。
dataLogDir：事物日志的存儲路徑，如果不配置該路徑，當zk吞吐量較大的時，會嚴重影響zk的性能。
clientPort：客戶端連接 zookeeper 服務器的端口，默認為2181。
server.1 這個1是服務器的標識也可以是其他的數字，這個標識要寫到dataDir目錄下面myid文件里。
server.1=hostname:2888:3888，hostname后面的第一個端口2888主要用於leader和follower之間的通信，第二個端口3888主要用於選主。
5.配置zookeeper環境變量，首先打開profile文件，vim /etc/profile 按i進入編輯模式，在文件末尾添加zookeeper環境變量
#set zookeeper environment
export ZK_HOME=/usr/local/zookeeper/zookeeper-3.4.13/
export PATH=$ZK_HOME/bin:$PATH
保存文件后，讓該環境變量生效 source /etc/profile
6.另外兩台也一樣安裝
7.啟動zookeeper
'/usr/local/zookeeper/zookeeper-3.4.13/bin/zkServer.sh start'
8.查看zookeeper狀態
'/usr/local/zookeeper/zookeeper-3.4.13/bin/zkServer.sh status'
9.停止zookeeper
'/usr/local/zookeeper/zookeeper-3.4.13/bin/zkServer.sh stop'
三.安裝kafka：
1.可以wget下載kafka_2.11-2.1.1.tgz，也可以下載到本機安裝包，然后用rz命令傳到服務器/usr/local/kafka路徑下。
2.解壓縮，用解壓縮命令，tar -zxvf kafka_2.11-2.1.1.tgz解壓。
3.配置kafka環境變量，vim /etc/profile，按i進入編輯模式，在文件末尾添加kafka環境變量
#set kafka environment
export KAFKA_HOME=/usr/local/kafka/kafka_2.11-2.1.1
PATH=${KAFKA_HOME}/bin:$PATH
保存文件后，讓該環境變量生效
source /etc/profile
4.10.0.66.50服務器，修改配置文件config下面的server.properties
broker.id=1
listeners = PLAINTEXT://10.0.66.50:9092
log.dirs=/tmp/kafka/log
zookeeper.connect=10.0.66.50:2181,10.0.47.32:2181,10.0.66.65:2181
10.0.47.32服務器，修改配置文件server.properties
broker.id=2
listeners = PLAINTEXT://10.0.47.32:9092
log.dirs=/tmp/kafka/log
zookeeper.connect=10.0.66.50:2181,10.0.47.32:2181,10.0.66.65:2181
10.0.66.65服務器，修改配置文件server.properties
broker.id=3
listeners = PLAINTEXT://10.0.66.65:9092
log.dirs=/tmp/kafka/log
zookeeper.connect=10.0.66.50:2181,10.0.47.32:2181,10.0.66.65:2181
主要修改一下四個地方:
1) broker.id 需要保證每一台kafka都有一個獨立的broker
2) listeners = PLAINTEXT://當前虛擬機ip地址:9092
3) log.dirs 數據存放的目錄
4) zookeeper.connect zookeeper的連接地址信息
5.啟動kafka（要確保zookeeper已啟動），每台主機上分別啟動kafka。
/usr/local/kafka/kafka_2.11-2.1.1/bin/kafka-server-start.sh -daemon /usr/local/kafka/kafka_2.11-2.1.1/config/server.properties
6.創建一個名稱為test-topic-7的Topic，7個分區，並且復制因子為3，執行如下命令：
/usr/local/kafka/kafka_2.11-2.1.1/bin/kafka-topics.sh --create --zookeeper 10.0.66.50:12181,10.0.47.32:12181,10.0.66.65:12181 --replication-factor 3 --partitions 7 --topic test-topic-7
7.查看創建的topic信息，進入kafka的bin目錄下，執行如下命令：
/usr/local/kafka/kafka_2.11-2.1.1/bin/kafka-topics.sh --describe --zookeeper 10.0.66.50:12181,10.0.47.32:12181,10.0.66.65:12181 --topic test-topic-7
8.在其中一台服務器啟動生產者(producer)
/usr/local/kafka/kafka_2.11-2.1.1/bin/kafka-console-producer.sh --broker-list 10.0.66.50:9092,10.0.47.32:9092,10.0.66.65:9092 --topic test-topic-7
9.在任意議一台服務器啟動消費者(consumer)，接受消息
/usr/local/kafka/kafka_2.11-2.1.1/bin/kafka-console-consumer.sh --bootstrap-server 10.0.66.50:9092,10.0.47.32:9092,10.0.66.65:9092 --topic test-topic-7 --from-beginning
10.如果consumer服務器接收到producer服務器輸入的內容，則啟動集群成功。
執行生產者命令，在生產者命令窗口輸入消息。
執行消費者命令，在消費者命令窗口顯示消息。
