1. 準備三臺服務器, 安裝jdk1.8 ,其中每一臺虛擬機的hosts文件中都需要配置如下的內容

192.168.72.141 node01
192.168.72.142 node02
192.168.72.143 node03

1. 安裝目錄

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

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

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

3)注意先把zookeeper集群已經安裝好并且已經啟動好

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

其中2.11是推薦版本.

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

1. 切換目錄上傳安裝包
   cd /export/software
   rz # 選擇對應安裝包上傳即可

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

3. 重命名(由于名稱太長)
   mv kafka_2.11-1.0.0 kafka

3.4 修改kafka的核心配置文件

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

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

#broker.id 標識了kafka集群中一個唯一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)聽的地址. 需要更改為當前虛擬機的ip地址, 保證其他主機都能連接
listeners = PLAINTEXT://當前虛擬機的ip地址:9092

# 存放生產者生產的數據 數據一般以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ā)到其他二臺主機

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

拷貝后, 需要修改每一臺的broker.id

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

修改每一臺的listeners的ip地址

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

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

mkdir -p /export/data/kafka

以上步驟做完后我們就可以啟動kafka的集群了
3.6 啟動集群

//先進入bin的目錄
	cd /export/servers/kafka/bin   
//在bin目錄下啟動
./kafka-server-start.sh /export/servers/kafka/config/server.properties 1>/dev/null 2>&1 &

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

//輸入jps命令查看
jps

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

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

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

1. 使用Kafka自帶一個命令行客戶端啟動一個生產者，生產數據

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

1. 使用Kafka自帶一個命令行客戶端啟動一個消費者，消費數據

//該消費語句，只能獲取最新的數據，要想歷史數據，需要添加選項--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. 查看某一個具體的Topic的詳細信息

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

1. 刪除topic

//	這個只會標識一下, 這個topic已經被刪除了
./kafka-topics.sh --delete --topic order --zookeeper node01:2181

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

原始數據, 必須手動刪除即可

如果在使用中, 需要刪除一個topic: 直接刪除zookeeper中對應topic的節(jié)點即可,
如果要刪除原始數據, 需要到對應目錄下手動刪除