2.1 環(huán)境準(zhǔn)備
2.1.1 集群規(guī)劃
bigdata11? ? bigdata12? ? bigdata13
zk? ? ? ? ? ? ? zk? ? ? ? ? ? ? zk
kafka? ? ? ? kafka? ? ? ? ? kafka
2.1.2 jar包下載
http://kafka.apache.org/downloads
2.1.3 虛擬機(jī)準(zhǔn)備
1）準(zhǔn)備3臺(tái)虛擬機(jī)
2）配置ip地址
3）配置主機(jī)名稱
4）3臺(tái)主機(jī)分別關(guān)閉防火墻
chkconfig iptables off
chkconfig iptables off
chkconfig iptables off
2.1.4 安裝jdk
2.1.5 安裝Zookeeper
0）集群規(guī)劃
在bigdata11、bigdata12和bigdata13三個(gè)節(jié)點(diǎn)上部署Zookeeper廓握。
1）解壓安裝
（1）解壓zookeeper安裝包到/opt/module/目錄下
tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
（2）在/opt/module/zookeeper-3.4.10/這個(gè)目錄下創(chuàng)建zkData
mkdir -p zkData
（3）重命名/opt/module/zookeeper-3.4.10/conf這個(gè)目錄下的zoo_sample.cfg為zoo.cfg
mv zoo_sample.cfg zoo.cfg
2）配置zoo.cfg文件
（1）具體配置
dataDir=/opt/module/zookeeper-3.4.10/zkData
增加如下配置
#######################cluster##########################
server.2=bigdata11:2888:3888
server.3=bigdata12:2888:3888
server.4=bigdata13:2888:3888
（2）配置參數(shù)解讀
Server.A=B:C:D敦锌。
A是一個(gè)數(shù)字，表示這個(gè)是第幾號(hào)服務(wù)器郊艘；
B是這個(gè)服務(wù)器的ip地址光戈；
C是這個(gè)服務(wù)器與集群中的Leader服務(wù)器交換信息的端口杂抽；
D是萬(wàn)一集群中的Leader服務(wù)器掛了，需要一個(gè)端口來(lái)重新進(jìn)行選舉生音，選出一個(gè)新的Leader宁否，而這個(gè)端
口就是用來(lái)執(zhí)行選舉時(shí)服務(wù)器相互通信的端口。
集群模式下配置一個(gè)文件myid缀遍，這個(gè)文件在dataDir目錄下慕匠，這個(gè)文件里面有一個(gè)數(shù)據(jù)就是A的值，Zookeeper啟動(dòng)時(shí)讀取此文件域醇，拿到里面的數(shù)據(jù)與zoo.cfg里面的配置信息比較從而判斷到底是哪個(gè)server台谊。
3）集群操作
（1）在/opt/module/zookeeper-3.4.10/zkData目錄下創(chuàng)建一個(gè)myid的文件
touch myid
添加myid文件，注意一定要在linux里面創(chuàng)建譬挚，在notepad++里面很可能亂碼
（2）編輯myid文件
vi myid
在文件中添加與server對(duì)應(yīng)的編號(hào)：如2
（3）拷貝配置好的zookeeper到其他機(jī)器上
scp -r zookeeper-3.4.10/ root@bigdata12.itstar.com:/opt/module/
scp -r zookeeper-3.4.10/ root@bigdata13.itstar.com:/opt/module/
并分別修改myid文件中內(nèi)容為3锅铅、4
（4）分別啟動(dòng)zookeeper
[root@bigdata11 zookeeper-3.4.10]# bin/zkServer.sh start
[root@bigdata12 zookeeper-3.4.10]# bin/zkServer.sh start
[root@bigdata13 zookeeper-3.4.10]# bin/zkServer.sh start
（5）查看狀態(tài)
[root@bigdata11 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
[root@bigdata12 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: leader
[root@bigdata13 zookeeper-3.4.5]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
2.2 Kafka集群部署
1）解壓安裝包
2）修改解壓后的文件名稱
3）在/opt/module/kafka目錄下創(chuàng)建logs文件夾
4）修改配置文件
輸入以下內(nèi)容：
5）配置環(huán)境變量
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
[itstar@bigdata11 software]$ tar -zxvf kafka_2.11-0.11.0.2.tgz -C /opt/module/
[itstar@bigdata11 module]$ mv kafka_2.11-0.11.0.2/ kafka
[itstar@bigdata11 kafka]$ mkdir logs
[itstar@bigdata11 kafka]$ cd config/
[itstar@bigdata11 config]$ vi server.properties
#broker的全局唯一編號(hào)，不能重復(fù)
broker.id=0
#是否允許刪除topic
delete.topic.enable=true
#處理網(wǎng)絡(luò)請(qǐng)求的線程數(shù)量
num.network.threads=3
#用來(lái)處理磁盤IO的線程數(shù)量
num.io.threads=8
#發(fā)送套接字的緩沖區(qū)大小
socket.send.buffer.bytes=102400
#接收套接字的緩沖區(qū)大小
socket.receive.buffer.bytes=102400
#請(qǐng)求套接字的最大緩沖區(qū)大小
socket.request.max.bytes=104857600
#kafka運(yùn)行日志存放的路徑
log.dirs=/opt/module/kafka/logs
#topic在當(dāng)前broker上的分區(qū)個(gè)數(shù)
num.partitions=1
#用來(lái)恢復(fù)和清理data下數(shù)據(jù)的線程數(shù)量
num.recovery.threads.per.data.dir=1
#segment文件保留的最長(zhǎng)時(shí)間减宣，超時(shí)將被刪除
log.retention.hours=168
#配置連接Zookeeper集群地址
zookeeper.connect=bigdata11:2181,bigdata12:2181,bigdata13:2181
5）配置環(huán)境變量
[root@bigdata11 module]# vi /etc/profile
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[root@bigdata11 module]# source /etc/profile
6）分發(fā)安裝包
[root@bigdata11 etc]# xsync profile
[itstar@bigdata11 module]$ xsync kafka/
7）分別在bigdata12和bigdata13上修改配置文件/opt/module/kafka/confifig/server.properties中的
broker.id=1盐须、broker.id=2
注：broker.id不得重復(fù)
8）啟動(dòng)集群
依次在bigdata11、bigdata12漆腌、bigdata13節(jié)點(diǎn)上啟動(dòng)kafka(加上& ,是在后臺(tái)啟動(dòng))
[itstar@bigdata11 kafka]$ bin/kafka-server-start.sh config/server.properties &
[itstar@bigdata12 kafka]$ bin/kafka-server-start.sh config/server.properties &
[itstar@bigdata13 kafka]$ bin/kafka-server-start.sh config/server.properties &
9）關(guān)閉集群
[itstar@bigdata11 kafka]$ bin/kafka-server-stop.sh stop
[itstar@bigdata12 kafka]$ bin/kafka-server-stop.sh stop
[itstar@bigdata13 kafka]$ bin/kafka-server-stop.sh stop
2.3 Kafka命令行操作
1）查看當(dāng)前服務(wù)器中的所有topic
[itstar@bigdata11 kafka]$ bin/kafka-topics.sh --zookeeper bigdata13:2181 --list
2）創(chuàng)建topic
[itstar@bigdata11 kafka]$ bin/kafka-topics.sh --zookeeper bigdata13:2181 --
create --replication-factor 3 --partitions 1 --topic first
選項(xiàng)說(shuō)明：
--topic 定義topic名
--replication-factor 定義副本數(shù)
--partitions 定義分區(qū)數(shù)
3）刪除topic
[itstar@bigdata11 kafka]$ bin/kafka-topics.sh --zookeeper bigdata11:2181 --
delete --topic first
需要server.properties中設(shè)置delete.topic.enable=true否則只是標(biāo)記刪除或者直接重啟。
4）發(fā)送消息
[itstar@bigdata11 kafka]$ bin/kafka-console-producer.sh --broker-list
bigdata11:9092 --topic first
>hello world
>itstar itstar
5）消費(fèi)消息
[itstar@bigdata12 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server
node3:9092 --from-beginning --topic first
--from-beginning：會(huì)把first主題中以往所有的數(shù)據(jù)都讀取出來(lái)。根據(jù)業(yè)務(wù)場(chǎng)景選擇是否增加該配置捌治。
6）查看某個(gè)Topic的詳情
[itstar@bigdata11 kafka]$ bin/kafka-topics.sh --zookeeper bigdata11:2181 --
describe --topic first
2.4 Kafka配置信息
2.4.1 Broker配置信息
2.4.2 Producer配置信息
2.4.3 Consumer配置信息
