1. 準(zhǔn)備三臺(tái)服務(wù)器, 安裝jdk1.8 ,其中每一臺(tái)虛擬機(jī)的hosts文件中都需要配置如下的內(nèi)容

192.168.72.141 node01
192.168.72.142 node02
192.168.72.143 node03

1. 安裝目錄

安裝包存放的目錄：/export/software
安裝程序存放的目錄：/export/servers
數(shù)據(jù)目錄：/export/data
日志目錄：/export/logs

創(chuàng)建各級(jí)目錄命令:

mkdir -p /export/servers/
mkdir -p /export/software/
mkdir -p /export/data/
mkdir -p /export/logs/ 

3)注意先把zookeeper集群已經(jīng)安裝好并且已經(jīng)啟動(dòng)好

下面開(kāi)始安裝kafka:
3.2 下載安裝包
由于kafka是scala語(yǔ)言編寫(xiě)的，基于scala的多個(gè)版本，kafka發(fā)布了多個(gè)版本。

其中2.11是推薦版本.

3.3 上傳安裝包并解壓
使用 rz 命令將安裝包上傳至 /export/software

1. 切換目錄上傳安裝包
   cd /export/software
   rz # 選擇對(duì)應(yīng)安裝包上傳即可

2. 解壓安裝包到指定目錄下
   tar -zxvf kafka_2.11-1.0.0.tgz -C /export/servers/
   cd /export/servers/

3. 重命名(由于名稱(chēng)太長(zhǎng))
   mv kafka_2.11-1.0.0 kafka

3.4 修改kafka的核心配置文件

cd /export/servers/kafka/config/
vi   server.properties

主要修改一下四個(gè)地方:
	1) broker.id            需要保證每一臺(tái)kafka都有一個(gè)獨(dú)立的broker
	2) listeners = PLAINTEXT://當(dāng)前虛擬機(jī)ip地址:9092
	3) log.dirs             數(shù)據(jù)存放的目錄  
	4) zookeeper.connect    zookeeper的連接地址信息 
    

#broker.id 標(biāo)識(shí)了kafka集群中一個(gè)唯一broker。   
broker.id=0   
num.network.threads=3   
num.io.threads=8   
socket.send.buffer.bytes=102400   
socket.receive.buffer.bytes=102400   
socket.request.max.bytes=104857600  

#listeners : 表示的監(jiān)聽(tīng)的地址. 需要更改為當(dāng)前虛擬機(jī)的ip地址, 保證其他主機(jī)都能連接
listeners = PLAINTEXT://當(dāng)前虛擬機(jī)的ip地址:9092

# 存放生產(chǎn)者生產(chǎn)的數(shù)據(jù) 數(shù)據(jù)一般以topic的方式存放    
log.dirs=/export/data/kafka           
num.partitions=1   
num.recovery.threads.per.data.dir=1   
offsets.topic.replication.factor=1   
transaction.state.log.replication.factor=1   
transaction.state.log.min.isr=1   
log.retention.hours=168   
log.segment.bytes=1073741824   
log.retention.check.interval.ms=300000     

# zk的信息   
zookeeper.connect=node01:2181,node02:2181,node03:2181   
zookeeper.connection.timeout.ms=6000   
group.initial.rebalance.delay.ms=0  

3.5 將配置好的kafka分發(fā)到其他二臺(tái)主機(jī)

cd /export/servers
scp -r kafka/ node02:$PWD  
scp -r kafka/ node03:$PWD

拷貝后, 需要修改每一臺(tái)的broker.id

ip為61的服務(wù)器: broker.id=0
ip為62的服務(wù)器: broker.id=1
ip為63的服務(wù)器: broker.id=2

修改每一臺(tái)的listeners的ip地址

ip為141的服務(wù)器: listeners=PLAINTEXT://192.168.72.61:9092
ip為142的服務(wù)器: listeners=PLAINTEXT://192.168.72.62:9092
ip為143的服務(wù)器: listeners=PLAINTEXT://192.168.72.63:9092

修改完ip后立馬 創(chuàng)建存放數(shù)據(jù)的目錄：
在三臺(tái)的服務(wù)器執(zhí)行創(chuàng)建數(shù)據(jù)文件的命令

mkdir -p /export/data/kafka

以上步驟做完后我們就可以啟動(dòng)kafka的集群了
3.6 啟動(dòng)集群

//先進(jìn)入bin的目錄
	cd /export/servers/kafka/bin   
//在bin目錄下啟動(dòng)
./kafka-server-start.sh /export/servers/kafka/config/server.properties 1>/dev/null 2>&1 &

可以在三臺(tái)都執(zhí)行這個(gè)這個(gè)步驟，也可以先在一臺(tái)虛擬機(jī)上執(zhí)行
然后查看啟動(dòng)的狀態(tài)：

//輸入jps命令查看
jps

以下是kafka的基本使用和操作：

1. 創(chuàng)建一個(gè)topic:

./kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 1 --topic order

1. 使用Kafka自帶一個(gè)命令行客戶(hù)端啟動(dòng)一個(gè)生產(chǎn)者，生產(chǎn)數(shù)據(jù)

./kafka-console-producer.sh --broker-list node01:9092 --topic order

1. 使用Kafka自帶一個(gè)命令行客戶(hù)端啟動(dòng)一個(gè)消費(fèi)者，消費(fèi)數(shù)據(jù)

//該消費(fèi)語(yǔ)句，只能獲取最新的數(shù)據(jù)，要想歷史數(shù)據(jù)，需要添加選項(xiàng)--from-beginning
./kafka-console-consumer.sh --bootstrap-server node01:9092  --topic order
如：bin/kafka-console-consumer.sh --bootstrap-server node01:9092 --from-beginning --topic order

1. 查看有哪些topic

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

1. 查看某一個(gè)具體的Topic的詳細(xì)信息

./kafka-topics.sh --describe --topic order --zookeeper node01:2181

1. 刪除topic

//	這個(gè)只會(huì)標(biāo)識(shí)一下, 這個(gè)topic已經(jīng)被刪除了
./kafka-topics.sh --delete --topic order --zookeeper node01:2181

注意：徹底刪除一個(gè)topic，需要在server.properties中配置delete.topic.enable=true，否則只是標(biāo)記刪除
配置完成之后，需要重啟kafka服務(wù)。
徹底刪除, 描述將元數(shù)據(jù)信息徹底刪除, 不會(huì)刪除原始數(shù)據(jù)

原始數(shù)據(jù), 必須手動(dòng)刪除即可

如果在使用中, 需要?jiǎng)h除一個(gè)topic: 直接刪除zookeeper中對(duì)應(yīng)topic的節(jié)點(diǎn)即可,
如果要?jiǎng)h除原始數(shù)據(jù), 需要到對(duì)應(yīng)目錄下手動(dòng)刪除