使用場景
- 異步處理： 如用戶注冊后，發送注冊郵件，再發送注冊短信。
- 應用解耦： 如用戶下單后，訂單系統需要通知庫存系統。
- 流量削峰：如秒殺活動，一般會因為流量過大，導致流量暴增。
- 日志處理：解決大量日志采集后的傳輸問題。消息隊列負責日志后續處理轉發。
- 消息通訊：
優點
-
高吞吐量，非常普通的應將kafka也可以支持每秒數百萬的消息。
-
支持通過kafka服務器和消費機集群來區分消息。
-
支持hadoop並行數據加載。
關鍵概念：
-
broker: kafka集群中的一台或多台服務器統稱broker。
-
Topic： kafka處理的消息源（feeds of messages）的不同分類。
-
partition：Topic物理機上的分組，一個topic可分為多個partition，每個partition是一個有序的隊列。其中的每條消息都會被分配一個有序的id（offset）。
-
Message ：消息，是通信的基本單位。每個producer可以向一個topic（主題）發布一些消息。
-
Producers：消息和數據的生產者，向kafka的一個topic發布消息的過程叫producers。
-
consumers：消息和數據消費者，訂閱topic並處理其發布的消息的過程叫做consumers。
安裝運行
官網地址：http://kafka.apache.org/downloads
1. 軟件環境
- CenOS7/Ubuntu 虛擬機三台。
- 已經搭建好的zookeeper集群。
- 軟件版本：kafka_2.12-2.2.0
2. 安裝步驟
- 下載官方二進制包
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz
- 解壓
tar -zxvf kafka_2.12-2.2.0.tgz -C /usr/local/
- 修改配置文件
vim /usr/local/kafka_2.12-2.2.0/config/server.properties
# 不可重復
# 這是這台虛擬機上的值，在另外兩台虛擬機上應該是2或者3，
# 這個值是唯一的，每台虛擬機或者叫服務器不能相同
broker.id=1
# 填寫本機ip
listeners=PLAINTEXT://192.168.137.200:9092
log.retention.hours=24
# 新增下面三項
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
# 設置zookeeper的連接端口,新版本的kafka不再使用zookeeper
# 而是通過brokerlist的配置讓producer直接連接broker
# 這個brokerlist可以配置多個，只要有一個能連接上，就可以讓producer獲取道集群中的其他broker的信息,繞過了zookeeper
zookeeper.connect=192.168.137.1:2181,192.168.137.1:2182,192.168.137.1:2183
3. 啟動運行
# 進入kafka的bin目錄
cd /usr/local/kafka_2.12-2.2.0/bin/
# 啟動kafka
./kafka-server-start.sh -daemon ../config/server.properties
-
檢測是否啟動成功
jps
4. 驗證是否創建成功
- 創建 topic
./kafka-topics.sh --create --zookeeper 192.168.137.1:2181 --replication-factor 2 --partitions 1 --topic my-topic
參數解釋：
--replication-factor 2// 復制兩份
--partitions 1// 創建1個分區
--topic// 主題為my-topic
-- --zookeeper// 此處為為zookeeper監聽的地址
- 創建生產者 producer
# IP地址可以寫brokerlist中的任意一個
./kafka-console-producer.sh --broker-list 192.168.137.200:9092 --topic my-topic
此時，console處於阻塞狀態，可以直接輸入數據。
- 創建消費者 consumer
# 要切換到另一台虛擬機的shell界面輸入以下命令
./kafka-console-consumer.sh --bootstrap-server 192.168.137.100:9092 --topic my-topic --from-beginning
- 顯示所有topic
./kafka-topics.sh --list --zookeeper 192.168.137.1:2181
5. Kafka-manager 搭建
下載地址:https://github.com/yahoo/kafka-manager/releases
# 在線下載
wget https://github.com/yahoo/kafka-manager/archive/2.0.0.2.tar.gz
# 解壓到 /usr/local
tar -zxvf 2.0.0.2.tar.gz -C /usr/local
# 修改配置
vim /usr/local/kafka-manager-2.0.0.2/conf/application.conf
- 編輯配置文件 application.conf
## 注釋這一行，下面添加一行
# kafka-manager.zkhosts="kafka-manager-zookeeper:2181"
## 根據自己的zookeeper重新配置
kafka-manager.zkhosts="192.168.137.1:2181,192.168.137.1:2182,192.168.137.1:2183"
- 啟動 Kafka-manager
# 切換到bin目錄
cd /usr/local/kafka-manager-2.0.0.2/
# 執行生成zip部署文件
./sbt clean dist
# 打包結束后,提取zip文件並解壓到/usr/local
unzip /usr/local/kafka-manager-2.0.0.2/target/universal/kafka-manager-2.0.0.2.zip
# 移動到/usr/local/
mv kafka-manager-2.0.0.2 /usr/local/kafka-manager
# 啟動項目
# -Dhttp.port，指定端口,默認端口 9000
# -Dconfig.file=conf/application.conf 指定配置文件:
# 切換到目錄,
cd /usr/local/kafka-manager/bin
# 方式1 簡單啟動
./bin/kafka-manager
# 方式2 指定配置,端口,后台啟動
nohup ./kafka-manager -Dconfig.file=../conf/application.conf -Dhttp.port=9000 &