柚子快報激活碼778899分享：Kafka

http://yzkb.51969.com/

目錄

Kafka是什么

Kafka的特性

Kafka系統(tǒng)架構

topic

partition

broker

consumer group

offset

zookeeper

部署安裝kafka

假設一個程序員在維護兩個服務器，一個是A另一個是B，A每秒發(fā)出200個消息，B每秒只能處理100個消息，此時B是接收不過來的，常規(guī)情況下要么“開源”要么“節(jié)流”，減少A發(fā)出的消息數(shù)量，或者增加B的處理速度，最簡單直接的就是再增設一臺C服務器用來幫助B進行負載均衡，但這樣做成本太高，有什么辦法可以讓B獨自處理來自A的消息而且不會導致自身因為超出負荷而癱瘓呢？

沒有什么是增加中間層解決不了的，我們在兩個服務器之間添加一個消息隊列，A發(fā)送的消息首先進入消息隊列，然后由消息隊列發(fā)送至B服務器，常用的消息隊列就是Kafka

Kafka是什么

Kafka就是一個消息隊列，所以如果問kafka是什么，就相當于問消息隊列是什么，消息隊列就是通過將消息放在內存中的隊列中，可以保護消息不被丟失。但是，這會帶來性能問題，因此可以將隊列挪到一個單獨的進程中，從而解決性能問題。此外，還可以通過分區(qū)、分片、添加副本來擴展消息隊列，提高性能和可用性。

Kafka的特性

高吞吐量、低延遲

可擴展性，kafka集群支持熱擴展

持久化、可靠性，消息唄持久化到本地磁盤，并且支持數(shù)據(jù)備份防止數(shù)據(jù)丟失

容錯性，允許集群中的某個或某系節(jié)點下線的情況存在

高并發(fā)，支持數(shù)千個客戶端同時讀寫

Kafka系統(tǒng)架構

Kafka通過將消息放入隊列中，減輕了消息處理端的負載，將消息放入消息隊列的一端稱為生產(chǎn)者，處理消息的一段稱為消費者。但是話又說回來，消息隊列說到底也只是一個緩沖區(qū)，只能作為緩兵之計，如果生產(chǎn)的消息在很長一段時間甚至幾年內都處于居高不下的狀態(tài)，那該花錢還是得花錢，為了提升性能和擴展性，可以增加消費者和生產(chǎn)者的數(shù)量，并通過分區(qū)和副本來保證高可用性。

topic

隨著生產(chǎn)者和消費者的增多，慢慢的消息隊列開始變得擁堵，就會出現(xiàn)多個生產(chǎn)者或消費者爭搶一個消息隊列的情況，這時候該如何做呢？面多加水，水多加面唄，我們再增加一條消息隊列不就可以了？不多再多增一條未免有些麻煩了，在消息隊列內部就直接進行分類，每一類是一個topic，然后根據(jù)topic新增隊列的數(shù)量， 生產(chǎn)者根據(jù) topic 主題的不同，將消息投遞到不同的隊列中，消費者根據(jù)訂閱的消息的類型，從相應的隊列中讀取消息。

partition

每個topic代表不同類型或主題的消息，自然有多有少，像系統(tǒng)進程允許之類的日志，消息是非常多的，就導致這個topic 處理的請求也過多，來不及處理，此時可將 topic 拆分成好幾段，每段就是一個partition分區(qū)，每個消費者負責一個partition，這樣每個消費者負責一個partition分區(qū)，大大降低了消息隊列堵塞的可能性

每個 partition 有多個副本（Replica），其中有且僅有一個作為 Leader，Leader 是當前負責數(shù)據(jù)的讀寫的 partition。Follower 跟隨 Leader，所有寫請求都通過 Leader 路由，數(shù)據(jù)變更會廣播給所有 Follower， Follower 與 Leader 保持數(shù)據(jù)同步。Follower 只負責備份，不負責數(shù)據(jù)的讀寫。 如果 Leader 故障，則從 Follower 中選舉出一個新的 Leader。 當 Follower 掛掉、卡住或者同步太慢，Leader 會把這個 Follower 從 ISR（Leader 維護的一個和 Leader 保持同步的 Follower 集合） 列表中刪除，重新創(chuàng)建一個 Follower。

broker

當然，如此多的partition分布在一臺機器上，肯定會造成機器負載，將partition分布在不同的機器上，可以大大減少這方面問題，每臺機器就是一個broker，一個broker 中有多個topic，每個topic被劃分為不同的partition

當 broker 存儲 topic 的數(shù)據(jù)時，如果某 topic 有 N 個 partition，集群有 N 個 broker，那么每個 broker 存儲該 topic 的一個 partition；如果某 topic 有 N 個 partition，集群有 (N+M) 個 broker，那么其中有 N 個 broker 存儲 topic 的 一個 partition， 剩下的 M 個 broker 不存儲該 topic 的 partition 數(shù)據(jù)；如果某 topic 有 N 個 partition，集群中 broker 數(shù)目少于 N 個，那么一個 broker 存儲該 topic 的 一個或多個 partition，實際情況中，避免broker數(shù)少于partition，這種情況容易導致 Kafka 集群數(shù)據(jù)不均衡

consumer group

消費者組，由多個 consumer 組成，消費者組內的消費者共同消費一個或多個topic，消費者組之間互不干擾

offset

消息隊列如是工作，假如有一天broker掛了，那豈不是數(shù)據(jù)全部都丟失了？所以，不能僅僅將數(shù)據(jù)放在內存中，還要將其持久化到磁盤中，這樣數(shù)據(jù)的安全性便有了保障，但是新的問題出現(xiàn)了，如果broker掛了，之后進行了數(shù)據(jù)恢復，消費者再次去拉取消息，應該從何處開始。這里就引入了一個新的概念 offset（偏移量），記錄消費者讀取數(shù)據(jù)的位置，以便下次直接讀取。當然，還有一個問題，這樣一直持久化下去，磁盤空間會越來越少，所以要對這部分持久化的數(shù)據(jù)進行定期清理，默認周期時 7*24小時

zookeeper

到這里，整個體系已經(jīng)由原來的AB兩臺機器中間的消息隊列，擴展到相當多的組件了，而且每個組件都有自己的數(shù)據(jù)和狀態(tài)，所以，就需要有一個組件來統(tǒng)一維護和管理這些組件的狀態(tài)信息，zookeeper就是扮演著這個角色，掌握kafka的狀態(tài)信息

部署安裝kafka

在官網(wǎng)中下載kafka壓縮包或者安裝包 放在/opt下

cd /opt

tar?zxvf kafka_2.13-2.7.1.tgz

mv kafka_2.13-2.7.1 /usr/local/kafka

cd /usr/local/kafka/config/

將配置文件備份

cp server.properties{,.bak}

vim server.properties

broker.id=0? ? ? ? ? ????????????????#21行，broker的全局唯一編號，不重復，因此其他機器上不能為0

listeners=PLAINTEXT://172.16.233.105:9092? ?????????????????#31行，指定堅挺的ip和端口號

num.network.threads=3? ? ? ? ? ? ? ? ? ? ?#42行，broker 處理網(wǎng)絡請求的線程數(shù)量

num.io.threads=8? ? ? ? ? ? ? ? ? ? ? ? ? ? ?#45行，用來處理磁盤IO的線程數(shù)量，數(shù)值應該大于硬盤數(shù)

socket.send.buffer.bytes=102400 ? ? ????????????????? #48行，發(fā)送套接字的緩沖區(qū)大小

socket.receive.buffer.bytes=102400?????????????? ? ? #51行，接收套接字的緩沖區(qū)大小

socket.request.max.bytes=104857600? ? ? ? ? ? ? ?#54行，請求套接字的緩沖區(qū)大小

log.dirs=/usr/local/kafka/logs? ? ? ? ? ? ? ? ? ?#60行，kafka運行日志存放的路徑，也是數(shù)據(jù)存放的路徑

num.partitions=1? ? ? ? ? ? ? #65行，topic在當前broker上的默認分區(qū)數(shù)，topic創(chuàng)建時指定參數(shù)會覆蓋

num.recovery.threads.per.data.dir=1? ? ?#69行，用來恢復和清理data下數(shù)據(jù)的線程數(shù)量

log.retention.hours=168? ? ?#103行，segment 文件保留的最長時間，單位h，默認7天刪除

log.segment.bytes=1073741824? ?#110行，一個segment文件最大的大小，默認為1G

zookeeper.connect=172.16.233.105:2181,172.16.233.106:2181,172.16.233.107:2181

#123行，配置連接zookeeper集群地址

保存退出

修改環(huán)境變量

vim /etc/profile

export KAFKA_HOME=/usr/local/kafka

export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

配置 zookeeper 啟動腳本

vim /etc/init.d/kafka

#!/bin/bash

#chkconfig:2345 22 88

#description:kafka Service Control Script

KAFKA_HOME='/usr/local/kafka'

case $1 in

start)

? ? ? ? echo "-------------- kafka 啟 動 -------------------"

? ? ? ? ${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

保存退出

賦權

chmod +x /etc/init.d/kafka

設置開機自啟

chkconfig --add kafka

啟動kafka

service kafka start

kafka已經(jīng)安裝部署完成

下面是kafka操作命令

創(chuàng)建topic

kafka-topics.sh --create --zookeeper

172.16.233.105:2181,172.16.233.106:2181,172.16.233.107:2181 --replication-factor 2 --partitions 3 --topic test

查看當前服務器中的所有 topic

172.16.233.105:2181,172.16.233.106:2181,172.16.233.107:2181

柚子快報激活碼778899分享：Kafka

http://yzkb.51969.com/

推薦鏈接