1、Kafka簡介

1.1、消息隊列

1.1.1、為什么要有消息隊列？

1.1.2、消息隊列

• 消息 Message
  網(wǎng)絡(luò)中的兩臺計算機或者兩個通訊設(shè)備之間傳遞的數(shù)據(jù)。例如說：文本、音樂、視頻等內(nèi)容。
• 隊列 Queue
  一種特殊的線性表（數(shù)據(jù)元素首尾相接），特殊之處在于只允許在首部刪除元素和在尾部追加元素(FIFO)。
  入隊、出隊。
• 消息隊列 MQ
  消息+隊列，保存消息的隊列。消息的傳輸過程中的容器；主要提供生產(chǎn)、消費接口供外部調(diào)用做數(shù)據(jù)的存
  儲和獲取。

1.1.3、消息隊列的分類

MQ主要分為兩類：點對點p2p、發(fā)布訂閱（Pub / Sub）

• Peer-to-Peer 一般基于Pull或者Polling接收數(shù)據(jù) 發(fā)送到隊列中的消息被一個而且僅僅一個接收者所接受，即
  使有多個接收者在同一個隊列中偵聽同一消息 即支持異步“即發(fā)即收”的消息傳遞方式，也支持同步請求/應(yīng)答
  傳送方式
  

• 發(fā)布訂閱 發(fā)布到同一個主題的消息，可被多個訂閱者所接收 發(fā)布/訂閱即可基于Push消費數(shù)據(jù)，也可基于Pull
  或者Polling消費數(shù)據(jù) 解耦能力比P2P模型更強
  

1.1.4、p2p 和 發(fā)布訂閱MQ的比較

• 共同點
  消息生產(chǎn)者生產(chǎn)消息發(fā)送到queue中，然后消費者從queue中讀取并且消費消息。
• 不同點
  p2p模型包括：消息隊列(Queue)、發(fā)送者(Sender)、接收者(Receiver)
  一個生產(chǎn)者生產(chǎn)的消息只有一個消費者(Consumer)(即一旦被消費，消息就不在消息隊列中)。比如說打電
  話。
  pub/Sub包含：消息隊列(Queue)、主題(Topic)、發(fā)布者(Publisher)、訂閱者(Subscriber)
  每個消息可以有多個消費者，彼此互不影響。比如我發(fā)布一個微博：關(guān)注我的人都能夠看到。

1.1.5、消息系統(tǒng)的使用場景

• 解耦 各系統(tǒng)之間通過消息系統(tǒng)這個統(tǒng)一的接口交換數(shù)據(jù)，無須了解彼此的存在
• 冗余 部分消息系統(tǒng)具有消息持久化能力，可規(guī)避消息處理前丟失的風(fēng)險
• 擴展 消息系統(tǒng)是統(tǒng)一的數(shù)據(jù)接口，各系統(tǒng)可獨立擴展
• 峰值處理能力 消息系統(tǒng)可頂住峰值流量，業(yè)務(wù)系統(tǒng)可根據(jù)處理能力從消息系統(tǒng)中獲取并處理對應(yīng)量的請求
• 可恢復(fù)性 系統(tǒng)中部分鍵失效并不會影響整個系統(tǒng)，它恢復(fù)會仍然可從消息系統(tǒng)中獲取并處理數(shù)據(jù)
• 異步通信 在不需要立即處理請求的場景下，可以將請求放入消息系統(tǒng)，合適的時候再處理

1.1.6、常見的消息系統(tǒng)

• RabbitMQ Erlang編寫，支持多協(xié)議AMQP，XMPP，SMTP，STOMP。支持負(fù)載均衡、數(shù)據(jù)持久化。同時支
  持Peer-to-Peer和發(fā)布/訂閱模式。
• Redis 基于Key-Value對的NoSQL數(shù)據(jù)庫，同時支持MQ功能，可做輕量級隊列服務(wù)使用。就入隊操作而言，
  Redis對短消息(小于10kb)的性能比RabbitMQ好，長消息性能比RabbitMQ差。
• ZeroMQ 輕量級，不需要單獨的消息服務(wù)器或中間件，應(yīng)用程序本身扮演該角色，Peer-to-Peer。它實質(zhì)上是
  一個庫，需要開發(fā)人員自己組合多種技術(shù)，使用復(fù)雜度高。
• ActiveMQ JMS實現(xiàn)，Peer-to-Peer，支持持久化、XA(分布式)事務(wù)
• Kafka/Jafka 高性能跨語言的分布式發(fā)布/訂閱消息系統(tǒng)，數(shù)據(jù)持久化，全分布式，同時支持在線和離線處理
• MetaQ/RocketMQ 純Java實現(xiàn)，發(fā)布/訂閱消息系統(tǒng)，支持本地事務(wù)和XA分布式事務(wù)

1.2、Kafka簡介

1.2.1、簡介

Kafka是分布式的發(fā)布—訂閱消息系統(tǒng)。它最初由LinkedIn(領(lǐng)英)公司發(fā)布，使用Scala語言編寫，與2010年12月份
開源，成為Apache的頂級項目。Kafka是一個高吞吐量的、持久性的、分布式發(fā)布訂閱消息系統(tǒng)。它主要用于處理
活躍live的數(shù)據(jù)(登錄、瀏覽、點擊、分享、喜歡等用戶行為產(chǎn)生的數(shù)據(jù))。

Kafka三大特點：

• 高吞吐量
  可以滿足每秒百萬級別消息的生產(chǎn)和消費——生產(chǎn)消費。
• 持久性
  有一套完善的消息存儲機制，確保數(shù)據(jù)的高效安全的持久化——中間存儲。
• 分布式
  基于分布式的擴展和容錯機制；Kafka的數(shù)據(jù)都會復(fù)制到幾臺服務(wù)器上。當(dāng)某一臺故障失效時，生產(chǎn)者和消
  費者轉(zhuǎn)而使用其它的機器——整體
  健壯性。

1.2.2、設(shè)計目標(biāo)

• 高吞吐率 在廉價的商用機器上單機可支持每秒100萬條消息的讀寫
• 消息持久化 所有消息均被持久化到磁盤，無消息丟失，支持消息重放
• 完全分布式 Producer，Broker，Consumer均支持水平擴展
• 同時適應(yīng)在線流處理和離線批處理

1.2.3、kafka核心的概念

一個MQ需要哪些部分？生產(chǎn)、消費、消息類別、存儲等等。 對于kafka而言，kafka服務(wù)就像是一個大的水池。不
斷的生產(chǎn)、存儲、消費著各種類別的消息。那么kafka由何組成呢？

Kafka服務(wù)：

• Topic：主題，Kafka處理的消息的不同分類。
• Broker：消息服務(wù)器代理，Kafka集群中的一個kafka服務(wù)節(jié)點稱為一個broker，主要存儲消息數(shù)據(jù)。存在硬
  盤中。每個topic都是有分區(qū)的。
• Partition：Topic物理上的分組，一個topic在broker中被分為1個或者多個partition，分區(qū)在創(chuàng)建topic的時候
  指定。
• Message：消息，是通信的基本單位，每個消息都屬于一個partition

Kafka服務(wù)相關(guān)

• Producer：消息和數(shù)據(jù)的生產(chǎn)者，向Kafka的一個topic發(fā)布消息。
• Consumer：消息和數(shù)據(jù)的消費者，定于topic并處理其發(fā)布的消息。
• Zookeeper：協(xié)調(diào)kafka的正常運行。

2、Kafka的分布式安裝

下載地址：https://kafka.apache.org/downloads
中文下載官網(wǎng)：https://kafka.apachecn.org/downloads.html
安裝包鏈接：https://pan.baidu.com/s/1G9F8TEfI88wPi_j2-hkK1A?pwd=e9tu
源碼包鏈接：https://pan.baidu.com/s/1LR7X3Is-JRsOOu3DdAp2aw?pwd=7249

2.1 jdk & zookeeper安裝

我們知道Kafka是由Zookeeper管理的，那么在安裝Kafka之前，先來安裝一下Zookeeper吧~

1、jdk 安裝配置

首先CentOS7中會默認(rèn)自帶jdk的，我的虛擬機里centos7默認(rèn)自帶的是open jdk 1.8.0_262_b10。

如果想要安裝指定版本的jdk，則先下載jdk安裝包。
Linux安裝jdk的詳細(xì)步驟

2、zookeeper安裝

我的kafka安裝包是3.4.0版本的，對應(yīng)的zookeeper版本是3.6.3，那么去官網(wǎng)下載好壓縮包（注意是 bin.tar.gz壓縮包）：
官網(wǎng)：http://archive.apache.org/dist/zookeeper/

首先，將安裝包放到Linux目錄下執(zhí)行以下命令：

$ mkdir zk
# 創(chuàng)建Zookeeper數(shù)據(jù)存儲路徑
$ mkdir zk/data
# 創(chuàng)建Zookeeper日志存放路徑
$ mkdir zk/logs
# 解壓安裝包
$ tar -zxvf apache-zookeeper-3.8.1-bin.tar.gz
# 配置環(huán)境變量,添加下述內(nèi)容
$ vi /etc/profile
export ZK_HOME=/home/install_package/apache-zookeeper-3.8.1-bin/bin
export PATH=$ZK_HOME/bin:$PATH
$ source /etc/profile
# 生成Zookeeper配置文件
$ cd apache-zookeeper-3.8.1-bin/conf
$ cp zoo_sample.cfg zoo.cfg   # 因為zookeeper默認(rèn)加載的配置文件名是zoo.cfg

然后修改一下配置(數(shù)據(jù)目錄和日志目錄)：

vim zoo.cfg

# 心跳間隔時間，時間單位為毫秒值
tickTime=2000
# leader與客戶端連接超時時間，設(shè)為5個心跳間隔
initLimit=10
# Leader與Follower之間的超時時間，設(shè)為2個心跳間隔
syncLimit=5
# 數(shù)據(jù)存放目錄
dataDir=/home/admin/Study/zk/data
# 日志存放目錄
dataLogDir=/home/admin/Study/zk/logs
# 客戶端通信端口
clientPort=2181
# 清理間隔，單位是小時，默認(rèn)是0，表示不開啟
#autopurge.purgeInterval=1
# 這個參數(shù)和上面的參數(shù)搭配使用，這個參數(shù)指定了需要保留的文件數(shù)目，默認(rèn)是保留3個
#autopurge.snapRetainCount=5
# 單機版不配下述配置
# server.NUM=IP:port1:port2 NUM表示本機為第幾號服務(wù)器；IP為本機ip地址；
# port1為leader與follower通信端口；port2為參與競選leader的通信端口
# 多個實例的端口配置不能重復(fù)，如下：
#server.0=192.168.101.136:12888:13888
#server.1=192.168.101.146:12888:13888

1、啟動zookeeper后臺服務(wù)：

zkServer.sh start

2、關(guān)閉zookeeper后臺服務(wù)：

zkServer.sh stop

3、查看zookeeper后臺服務(wù)運行狀態(tài)：

zkServer.sh status

2.2、Kafka安裝步驟

1、首先，在Linux下kafka壓縮包所在的目錄下，解壓：

$ mkdir kafka
# 創(chuàng)建kafka日志存放路徑
$ mkdir kafka/logs
# 解壓安裝包
$ tar -zxvf kafka_2.12-3.4.0.tgz
# 移動到kafka目錄下
mv kafka_2.12-3.4.0 kafka
# 配置環(huán)境變量,添加下述內(nèi)容
$ vi /etc/profile
export KAFKA_HOME=/home/admin/Study/kafka/kafka_2.12-3.4.0
export PATH=$KAFKA_HOME/bin:$PATH
$ source /etc/profile
# 修改kafka配置
$ cd kafka_2.12-3.4.0/config
$ vi server.properties

修改Kafka配置

# broker.id每個實例的值不能重復(fù)
broker.id=0
# 配置主機的ip和端口
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://192.168.57.30:9092
#advertised.listeners=PLAINTEXT://10.11.0.203:9092
# 配置日志存儲路徑
log.dirs=/home/admin/Study/kafka/logs
# 配置zookeeper集群
zookeeper.connect=localhost:2181

啟動kafka，zokeeper啟動的前提下：

bin/kafka-server-start.sh -daemon config/server.properties

判斷kafka啟動成功：

kafka關(guān)閉：

bin/kafka-server-stop.sh -daemon config/server.properties

3、Kafka集群

準(zhǔn)備三臺虛擬機。這里將上面安裝好的虛擬機直接克隆兩份。
克隆過程很簡單，這里不再贅述。接下來看一下克隆好后需要哪些配置的修改。

3.1、克隆機配置修改

① 修改主機名
關(guān)閉全部虛擬機，打開克隆好的第一臺，修改主機名為kafka02：

vim /etc/hostname

② 修改網(wǎng)絡(luò)地址

vim /etc//sysconfig/network-scripts/ifcfg-ens33

③ 重啟：

reboot

另一臺同樣方法修改：

主機名：kafka03
ip地址：192.168.255.214

（題外話：修改命令行背景色和字體色）

打開命令行——> Edit ——> Preferences ——> Colors ：取消勾選Use colors from system theme

④ 進入到kafka安裝目錄下修改kafka server.properties配置文件：

vim config/server.properties

broker.id

該屬性?于唯?標(biāo)記?個Kafka的Broker，它的值是?個任意integer值。
當(dāng)Kafka以分布式集群運?的時候，尤為重要。
最好該值跟該Broker所在的物理主機有關(guān)的，如主機名為host1.lagou.com，則broker.id=1，如果主機名為192.168.100.101，則broker.id=101等等。

listeners

?于指定當(dāng)前Broker向外發(fā)布服務(wù)的地址和端?。
與advertised.listeners配合，?于做內(nèi)外?隔離。

內(nèi)外?隔離配置：

• listener.security.protocol.map

監(jiān)聽器名稱和安全協(xié)議的映射配置。

?如，可以將內(nèi)外?隔離，即使它們都使?SSL。

• listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL

每個監(jiān)聽器的名稱只能在map中出現(xiàn)?次。

• inter.broker.listener.name

?于配置broker之間通信使?的監(jiān)聽器名稱，該名稱必須在advertised.listeners列表中。

• inter.broker.listener.name=EXTERNAL

• listeners

?于配置broker監(jiān)聽的URI以及監(jiān)聽器名稱列表，使?逗號隔開多個URI及監(jiān)聽器名稱。

如果監(jiān)聽器名稱代表的不是安全協(xié)議，必須配置 listener.security.protocol.map。

每個監(jiān)聽器必須使?不同的?絡(luò)端?。

• advertised.listeners

需要將該地址發(fā)布到zookeeper供客戶端使?，如果客戶端使?的地址與listeners配置不同。

可以在zookeeper的get /myKafka/brokers/ids/<broker.id>中找到。

在IaaS環(huán)境，該條?的?絡(luò)接?得與broker綁定的?絡(luò)接?不同。

如果不設(shè)置此條?，就使?listeners的配置。跟listeners不同，該條?不能使?0.0.0.0?絡(luò)端?。

advertised.listeners的地址必須是listeners中配置的或配置的?部分。

zookeeper.connect

該參數(shù)?于配置Kafka要連接的Zookeeper/集群的地址。

它的值是?個字符串，使?逗號分隔Zookeeper的多個地址。Zookeeper的單個地址是host:port形式的，可以在最后添加Kafka在Zookeeper中的根節(jié)點路徑。

3.2、kafka集群啟動

1、zookeeper啟動

zkServer.sh start

2、kafka啟動

命令：

bin/kafka-server-start.sh -daemon config/server.properties

以下報錯信息：

是因為kafka目錄下logs文件夾meta.properties文件中的broker.id和server.properties中的不一致了，修改一下即可。

啟動成功：

3.3、kafka操作命令

參考官網(wǎng)快速開始：Kafka中文文檔

1、查看主題

（新安裝kafka是沒有主題的）

bin/kafka-topics.sh --list --bootstrap-server 192.168.255.212:9092

2、創(chuàng)建主題：

2.1、創(chuàng)建一個名為“test”的topic，它有一個分區(qū)和一個副本

bin/kafka-topics.sh --create --bootstrap-server 192.168.255.212:9092 --replication-factor 1 --partitions 1 --topic test

此時查看主題：

在另外兩臺kafka主機上同樣可以看到：


此時，我們查看3臺kafka的logs目錄下：



可以看到創(chuàng)建的test主題。

2.2、測試再次創(chuàng)建一個主題，設(shè)置分區(qū)為3，（最好跟主機數(shù)量一致）：

bin/kafka-topics.sh --create --bootstrap-server 192.168.255.212:9092 --replication-factor 1 --partitions 3 --topic city

可以看到3臺機器logs目錄下分別都有一個主題分區(qū)。

2.3、創(chuàng)建主題cities，復(fù)制因子為2，分區(qū)為3

命令：

bin/kafka-topics.sh --create --bootstrap-server 192.168.255.212:9092 --replication-factor 2 --partitions 3 --topic cities

查看日志目錄下可以看到3個分區(qū)每個都有兩份：

3、刪除主題

bin/kafka-topics.sh --delete --bootstrap-server 192.168.255.212:9092 --topic 主題名

4、啟動生產(chǎn)者端/消費者端

以下命令：創(chuàng)建一個生產(chǎn)者客戶端，產(chǎn)生消息，主題為test。注意，生產(chǎn)者客戶端可以在任意主機上只要包含kafka，命令存在即可執(zhí)行，當(dāng)前在213這臺kafka上，既作為server又充當(dāng)客戶端。

bin/kafka-console-producer.sh --broker-list 192.168.255.213:9092 --topic test

回車，輸入消息：

>Beijing
>Shanghai

現(xiàn)在啟動一個消費者：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

后面加–from-beginning表示接受所有包括之前的消息；不加表示只接受最新的消息，過往的（消費者啟動前）不接受。


此時，生產(chǎn)一個Hello消息：

兩個消費者的接收：

注意：消費者不管消費哪個ip，都可以收到消息~

4、zokeeper查看kafka日志

上述操作命令我們都依次執(zhí)行并在kafka logs目錄下查看到產(chǎn)生的主題。那么實際上我們也可以在zokeeper中查看。
命令：
進入到zookeeper bin目錄下

zkCli.sh

ls /

ls /brokers
ls /brokers/ids
ls /brokers/topics

打開 brokerid = 0，查看數(shù)據(jù)內(nèi)容：

ls /brokers/ids/0
get /brokers/ids/0

可以查看到當(dāng)前主機的信息，以json格式存儲。

接下來查看主題內(nèi)容：

ls /brokers/topics/cities
ls /brokers/topics/cities/partitions
ls /brokers/topics/cities/partitions/0
ls /brokers/topics/cities/partitions/0/state
get /brokers/topics/cities/partitions/0/state

get /brokers/topics/cities

0號主機上有cities-1、cities-2分區(qū)。

同理，其他部分也可查看：

段segment

segment 是一個邏輯概念，其由兩類物理文件組成，分別為“.index”文件和“.log”文
件?！?log”文件中存放的是消息，而“.index”文件中存放的是“.log”文件中消息的索引。

進入test主題所在主機的logs下：

表示前面有0條消息。
00000000000000001456.log 表示前面有1456條消息。

查看segment

想要查看segment中的log文件，需要通過kafka自帶的一個工具查看。

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 
/home/admin/Study/kafka/logs/test-0/00000000000000000000.log --print-data-log

一個用戶的一個主題會被提交到一個__consumer_offsets 分區(qū)中。使用主題字符串的
hash 值與 50 取模，結(jié)果即為分區(qū)索引。一般默認(rèn)為50個分區(qū)（0 ~ 49）。
文章來源地址http://www.zghlxwxcb.cn/news/detail-645119.html

到了這里，關(guān)于消息中間件 —— 初識Kafka的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！