1.zookeeper

kafka3.0之前依賴于zookeeper

zookeeper是一個(gè)開源，分布式的架構(gòu)，提供協(xié)調(diào)服務(wù)（Apache項(xiàng)目）

基于觀察者模式涉及的分布式服務(wù)管理架構(gòu)

存儲(chǔ)和管理數(shù)據(jù)，分布式節(jié)點(diǎn)上的服務(wù)接受觀察者的注冊(cè)，一旦分布式節(jié)點(diǎn)上的數(shù)據(jù)如果發(fā)生變化，由zookeeper來負(fù)責(zé)通知分布式節(jié)點(diǎn)上的服務(wù)

zookeeper分為領(lǐng)導(dǎo)者，追隨者 leader follower組成的集群

只要有一半以上的集群存活，zookeeper集群就可以正常工作，適用于安裝奇數(shù)臺(tái)的服務(wù)集群

全局?jǐn)?shù)據(jù)一致，每個(gè)zookeeper每個(gè)節(jié)點(diǎn)都保存相同的數(shù)據(jù)，維護(hù)監(jiān)控服務(wù)的數(shù)據(jù)一致。（主要作用）

數(shù)據(jù)更新的原子性，要么都成功，要么都失敗

實(shí)時(shí)性，只要有變化，立刻同步。

zookeeper的應(yīng)用場景

1.統(tǒng)一命名服務(wù)，在分布式的環(huán)境下，對(duì)所有的應(yīng)用和服務(wù)進(jìn)行統(tǒng)一命名

2.統(tǒng)一配置管理，配置文件同步，kafka的配置文件被修改，可以快速同步到其他節(jié)點(diǎn)

3.統(tǒng)一集群管理，實(shí)時(shí)掌握所有節(jié)點(diǎn)的狀態(tài)

4.服務(wù)器動(dòng)態(tài)上下限

5.負(fù)載均衡，把訪問的服務(wù)器的數(shù)據(jù)，發(fā)送到訪問最少的服務(wù)器處理客戶端的請(qǐng)求

領(lǐng)導(dǎo)者和追隨者：zookeeper的選舉機(jī)制

三臺(tái)服務(wù)器：A B C

A 先啟動(dòng) 發(fā)起第一選舉，投票投給自己，只有一票，不滿半數(shù)，A的狀態(tài)是looking

B 啟動(dòng) 在發(fā)起一次選舉，A和B分別投自己一票，交換選票信息，myid A發(fā)現(xiàn)B的myid比A的大，A的這票會(huì)轉(zhuǎn)而投給B，A 0 B 2 沒有半數(shù)以上結(jié)果，A B 會(huì)進(jìn)入looking 、B有可能leader

C 啟動(dòng) MYID C的myid最大 A和B都會(huì)把票投給C A 0 B 0 C 3

C的狀態(tài)變?yōu)閘eader A B變?yōu)閒ollower

只有l(wèi)eader確定，后續(xù)的服務(wù)器都是追隨者

只有兩種情況會(huì)開啟選舉機(jī)制：

1.初始化的情況會(huì)產(chǎn)生選舉

2.服務(wù)器之間和eader丟失了連接狀態(tài)

特殊情況下：

leader已經(jīng)存在，建立連接即可

leader不存在，leader不存在

1.服務(wù)器ID大的勝出

2.EPOCH大，直接勝出

3.EPOCH相同，事務(wù)ID大的勝出

EPOCH每個(gè)leader任期的代號(hào)，沒有l(wèi)eader，大家的邏輯地址相同。每投完一次之后，數(shù)據(jù)時(shí)遞增

事務(wù)id表示服務(wù)器的每一次更新，每變更一次id變化一次‘

服務(wù)器ID

zookeeper 當(dāng)中所有機(jī)器。每臺(tái)機(jī)器不重復(fù)，和mysql保存一直

zookeeper+kafka（2.7.0）

kafka （3.4.1）

實(shí)現(xiàn)

zookeeper集群

192.168.233.10 zookeeper+kafka

192.168.233.20 zookeeper+kafka

192.168.233.30 zookeeper+kafka

所有

systemctl stop firewalld

setenforce 0

cd /opt

拖進(jìn)去 apache-zookeeper-3.5.7-bin.tar.gz kafka_2.13-2.7.0

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel

java -version

cd /opt

tar -xf apache-zookeeper-3.5.7-bin.tar.gz

mv apache-zookeeper-3.5.7-bin zookeeper

cd zookeeper

cd /opt/zookeeper/conf

cp zoo_sample.cfg zoo.cfg

vim zoo.cfg

tickTime=2000

#服務(wù)器與客戶端之間心跳時(shí)間，2秒檢測一次服務(wù)器和客戶端之間的通信

initLimit=10

#領(lǐng)導(dǎo)者和追隨者之間，最對(duì)幾次心跳數(shù)超時(shí) 10*2S 20S

syncLimit=5

#同步超時(shí)時(shí)間，領(lǐng)導(dǎo)者和追隨者之間，同步通信超時(shí)的時(shí)間。5*2s leader會(huì)認(rèn)為follower丟失，移除集群

16行

dataDir=/opt/zookeeper/data

dataLogDir=/opt/zookeeper/logs (需要改)

添加

server.1=192.168.233.10:3188:3288

server.2=192.168.233.20:3188:3288

server.3=192.168.233.30:3188:3288

##

server.1=192.168.233.10:3188:3288

1 定義每個(gè)zookeeper集群的初始myid

192.168.233.10 ：服務(wù)器的ip地址

3188：領(lǐng)導(dǎo)者和追隨者之間交換信息的端口（內(nèi)部通信的端口）

3288：一旦leader丟失響應(yīng)，開啟選舉，3288就是用來執(zhí)行選舉時(shí)的通信端口

##

wq!

mkdir /opt/zookeeper/data

mkdir /opt/zookeeper/logs

三臺(tái)分別創(chuàng)建

10：

cd ..

cd data/

echo 1 > myid

20：

cd ..

cd data/

echo 2 > myid

30:

cd ..

cd data/

echo 3 > myid

所有

vim /etc/init.d/zookeeper

#!/bin/bash

#chkconfig:2345 20 90

#description:Zookeeper Service Control Script

ZK_HOME='/opt/zookeeper'

case $1 in

start)

echo "---------- zookeeper 啟動(dòng) ------------"

$ZK_HOME/bin/zkServer.sh start

;;

stop)

echo "---------- zookeeper 停止 ------------"

$ZK_HOME/bin/zkServer.sh stop

;;

restart)

echo "---------- zookeeper 重啟 ------------"

$ZK_HOME/bin/zkServer.sh restart

;;

status)

echo "---------- zookeeper 狀態(tài) ------------"

$ZK_HOME/bin/zkServer.sh status

;;

*)

echo "Usage: $0 {start|stop|restart|status}"

esac

wq!

三臺(tái)服務(wù)器分別啟動(dòng)(一個(gè)一個(gè)起)

chmod +x /etc/init.d/zookeeper

chkconfig --add zookeeper

service zookeeper start

起完以后看狀態(tài)

service zookeeper status

2.kafka

為什么要引入消息隊(duì)列（MQ），首先他也是一個(gè)中間件。在高并發(fā)環(huán)境下，同步請(qǐng)求太多來不及處理，來不及處理的請(qǐng)求會(huì)阻塞，比方數(shù)據(jù)庫就會(huì)形成行鎖或者表鎖，請(qǐng)求線程滿了，超標(biāo)了，too many connection。整個(gè)系統(tǒng)雪崩

消息隊(duì)列的作用：異步處理請(qǐng)求 流量削峰，應(yīng)用解耦 可恢復(fù)性 緩沖機(jī)制

解耦：

耦合：在軟件系統(tǒng)當(dāng)中，修改一個(gè)組件需要修改所有其他組件，高度耦合

低度耦合：修改其中一個(gè)組件，對(duì)其他最賤影響不大，無需修改所有

A B C

只有通信保證，其他的修改不影響整個(gè)集群，每個(gè)組件可以獨(dú)立的擴(kuò)展，修改，降低組件之間的依賴性

依賴點(diǎn)就是接口約束，通過不同的端口，保證集群通信

可恢復(fù)性：系統(tǒng)當(dāng)中的有一部分組件消失，不影響整個(gè)系統(tǒng)，也就是在消息隊(duì)列當(dāng)中，即使有一個(gè)處理消息的進(jìn)程失敗，一旦恢復(fù)還可以重新加入到隊(duì)列當(dāng)中，繼續(xù)處理消息。

緩沖機(jī)制：可以控制和優(yōu)化數(shù)據(jù)經(jīng)過系統(tǒng)的時(shí)間和速度，解決生產(chǎn)消息和消費(fèi)消息處理速度不一致的問題

峰值的處理能力：消息隊(duì)列在峰值情況之下，能夠頂住突發(fā)的訪問壓力。避免專門為了突然情況而對(duì)系統(tǒng)進(jìn)行修改

異步通信：允許用戶把一個(gè)消息放入隊(duì)列，但是不立即處理，等我想處理的時(shí)候在處理

消息隊(duì)列的模式：

點(diǎn)對(duì)點(diǎn) 一對(duì)一：消息的生產(chǎn)者發(fā)送消息到隊(duì)列中，消費(fèi)者從隊(duì)列中提取消息，消費(fèi)者提取完之后，隊(duì)列中被提取的消息將會(huì)被移除，后續(xù)消費(fèi)者不能再消費(fèi)隊(duì)列當(dāng)中的消息，消息隊(duì)列可以有多個(gè)消費(fèi)者，但是一個(gè)消息，只能由一個(gè)消費(fèi)者提取

RABBITMQ

發(fā)布/訂閱模式：一對(duì)多，又叫做觀察者模式。消費(fèi)者提取數(shù)據(jù)之后，隊(duì)列當(dāng)中的消息不會(huì)被清除

生產(chǎn)者發(fā)布一個(gè)消息到對(duì)象（主題）所有消費(fèi)者都是通過主題獲取消費(fèi)之后，隊(duì)列當(dāng)中的消息不會(huì)消除

主題：topic topic類似于一個(gè)數(shù)據(jù)流的管道，生產(chǎn)者把消息發(fā)布到主題，消息從主題當(dāng)中訂閱數(shù)據(jù)，主題可以分區(qū)，每個(gè)分區(qū)都要自己的偏移量

分區(qū)：partition 每個(gè)主題都可以分成多個(gè)分區(qū)，每個(gè)分區(qū)是數(shù)據(jù)的有序子集，分區(qū)可以運(yùn)行kafka進(jìn)行水平擴(kuò)展，以處理大量數(shù)據(jù)

消息在分鐘按照偏移量存儲(chǔ)，消費(fèi)者可以獨(dú)立每個(gè)分區(qū)的數(shù)據(jù)

偏移量：是每個(gè)消息在分區(qū)中唯一的標(biāo)識(shí)，消費(fèi)者可以通過偏移量來跟蹤獲取已讀或者未讀消息的位置，也可以提交偏移量來記錄已處理的信息

生產(chǎn)者：producer 生產(chǎn)者把數(shù)據(jù)發(fā)送kafka的主題當(dāng)中 負(fù)責(zé)寫入消息

消費(fèi)者：consumer從主題當(dāng)中讀取數(shù)據(jù)，消費(fèi)者可以是一個(gè)也可以是多個(gè)，每個(gè)消費(fèi)者有一個(gè)唯一的消費(fèi)者組ID，kafka實(shí)現(xiàn)負(fù)載均衡和容錯(cuò)性

經(jīng)紀(jì)人：Broker每個(gè)kafka節(jié)點(diǎn)都有一個(gè)Broker，每個(gè)Broker負(fù)責(zé)一臺(tái)kafka服務(wù)器，id唯一，存儲(chǔ)主題分區(qū)當(dāng)中的數(shù)據(jù)，處理生產(chǎn)和消費(fèi)者的請(qǐng)求，維護(hù)元數(shù)據(jù)（zookeeper）

zookeeper：zookeeper負(fù)責(zé)保存元數(shù)據(jù)，元數(shù)據(jù)就是topic的相關(guān)信息（發(fā)布在哪臺(tái)主機(jī)上，指定了多少分區(qū)，以及副本數(shù)，偏移量）

zookeeper自建一個(gè)主題：_consumer_offsets,

3.0之后不依賴zookeeper的核心 元數(shù)據(jù)由kafka節(jié)點(diǎn)自己管理

消費(fèi)的方式：

begining，從頭開始

實(shí)時(shí)更新

指定位置，用代碼編寫

kafka的工作流程：

生產(chǎn)者向主題里面發(fā)送數(shù)據(jù)，主題里的分區(qū)保存數(shù)據(jù)，消費(fèi)者根據(jù)消費(fèi)方式來消費(fèi)數(shù)據(jù)。

生產(chǎn)者寫入的topic的數(shù)據(jù)時(shí)持久化，默認(rèn)7個(gè)小時(shí)

至少一次語義：只要消費(fèi)者進(jìn)入，確保消息至少被消費(fèi)一次。

實(shí)驗(yàn)

所有

cd /opt

tar -xf kafka_2.13-2.7.0.tgz

mv kafka_2.13-2.7.0 kafka

vim /etc/profile

export KAFKA_HOME=/opt/kafka

export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

cd /opt/kafka/config

cp server.properties server.properties.bak

10:

vim server.properties

21行：

65行 修改

130行 修改

zookeeper.connect=192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

配置zookeeper的集群

20：

所有：

vim /etc/init.d/kafka

#!/bin/bash

#chkconfig:2345 22 88

#description:Kafka Service Control Script

KAFKA_HOME='/opt/kafka'

case $1 in

start)

echo "---------- Kafka 啟動(dòng) ------------"

${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties

;;

stop)

echo "---------- Kafka 停止 ------------"

${KAFKA_HOME}/bin/kafka-server-stop.sh

;;

restart)

$0 stop

$0 start

;;

status)

echo "---------- Kafka 狀態(tài) ------------"

count=$(ps -ef | grep kafka | egrep -cv "grep|$$")

if [ "$count" -eq 0 ];then

echo "kafka is not running"

else

echo "kafka is running"

fi

;;

*)

echo "Usage: $0 {start|stop|restart|status}"

esac

wq!

chmod +x /etc/init.d/kafka

chkconfig --add kafka

service kafka start

netstat -antp | grep 9092

所有：

cd /opt/kafka/bin

20:

kafka-topics.sh --create --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --replication-factor 2 --partitions 3 --topic test1

kafka-topics.sh --create --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --replication-factor 2 --partitions 3 --topic test1

##

創(chuàng)建主題

1.在kafka的bin目錄下創(chuàng)建，是所有kafka可執(zhí)行命令的文件

2. --zookeeper指定的是zookeeper的地址和端口，保存kafka的元數(shù)據(jù)

3. --replication-factor 2 定義每個(gè)分區(qū)的副本數(shù)

4.partition 3 指定主題的分區(qū)數(shù)

5. --topic test1 指定主題的名稱

10:

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --topic test1

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --topic test3

所有

vim /etc/hosts

192.168.233.10 test1

192.168.233.20 test2

192.168.233.30 test3

記得改主機(jī)名

20:

kafka-console-producer.sh --broker-list 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-producer.sh --broker-list 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

30:

從頭開始

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3 --from-beginning

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1 --from-beginning

實(shí)時(shí)消息

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

20：創(chuàng)建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi1

kafka-topics.sh --create --zookeeper 192.168.66.16:2181 --partitions 1 --replication-factor 1 --topic guoqi1

30:創(chuàng)建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi2

kafka-topics.sh --create --zookeeper 192.168.66.17:2181 --partitions 1 --replication-factor 1 --topic guoqi2

10:連接

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi2

kafka-console-consumer.sh --bootstrap-server 192.168.66.16:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.66.17:9092 --topic guoqi2

20:發(fā)送

kafka-console-producer.sh --broker-list 192.168.233.20:9092 --topic guoqi1

kafka-console-producer.sh --broker-list 192.168.66.16:9092 --topic guoqi1

30:發(fā)送

kafka-console-producer.sh --broker-list 192.168.233.30:9092 --topic guoqi2

kafka-console-producer.sh --broker-list 192.168.66.17:9092 --topic guoqi2

如何修改分區(qū)數(shù)

20：kafka-topics.sh --zookeeper 192.168.233.20:2181 --alter --topic guoqi1 --partitons 3

刪除

kafka-topics.sh --delete --zookeeper 192.168.233.20:2181 --topic guoqi1

查看元數(shù)據(jù)

cd /opt/zookeeper/bin

./zkCli.sh -server 192.168.233.10:2181