本文源碼：GitHub·點這里 || GitEE·點這里
一、搭建Kafka環境
1、下載解壓
-- 下載
wget http://mirror.bit.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz
-- 解壓
tar -zxvf kafka_2.11-2.2.0.tgz
-- 重命名
mv kafka_2.11-2.2.0 kafka2.11
2、啟動Kafka服務
kafka依賴ZooKeeper服務，需要本地安裝並啟動ZooKeeper。
-- 執行位置
-- /usr/local/mysoft/kafka2.11
bin/kafka-server-start.sh config/server.properties
3、查看服務
ps -aux |grep kafka
4、開放地址端口
-- 基礎路徑
-- /usr/local/mysoft/kafka2.11/config
vim server.properties
-- 添加下面注釋
advertised.listeners=PLAINTEXT://192.168.72.130:9092
二、Kafka基礎概念
1、基礎描述
Kafka是由Apache開源，具有分布式、分區的、多副本的、多訂閱者，基於Zookeeper協調的分布式處理平台，由Scala和Java語言編寫。通常用來搜集用戶在應用服務中產生的動作日志數據，並高速的處理。日志類的數據需要高吞吐量的性能要求，對於像Hadoop一樣的日志數據和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。Kafka的目的是通過Hadoop的並行加載機制來統一線上和離線的消息處理，也是為了通過集群來提供實時的消息。
2、功能特點
(1)、通過磁盤數據結構提供消息的持久化，消息存儲也能夠保持長時間穩定性;
(2)、高吞吐量，即使是非常普通的硬件Kafka也可以支持每秒超高的並發量;
(3)、支持通過Kafka服務器和消費機集群來分區消息;
(4)、支持Hadoop並行數據加載;
(5)、API包封裝的非常好，簡單易用，上手快 ;
(6)、分布式消息隊列。Kafka對消息保存時根據Topic進行歸類，發送消息者稱為Producer，消息接受者稱為Consumer;
3、消息功能
- 點對點模式
點對點模型通常是一個基於拉取或者輪詢的消息傳遞模型，消費者主動拉取數據，消息收到后從隊列移除消息，這種模型不是將消息推送到客戶端，而是從隊列中請求消息。特點是發送到隊列的消息被一個且只有一個消費者接收處理，即使有多個消費者監聽隊列也是如此。
- 發布訂閱模式
發布訂閱模型則是一個基於推送的消息傳送模型，消息產生后，推送給所有訂閱者。發布訂閱模型可以有多種不同的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的所有消息，即使當前訂閱者不可用，處於離線狀態。
4、消息隊列作用
- 程序解耦，生產者和消費者獨立，各自異步執行;
- 消息數據進行持久化存儲，直到被全部消費，規避了數據丟失風險;
- 流量削峰，使用消息隊列承接訪問壓力，盡量避免程序雪崩 ;
- 降低進程間的耦合度，系統部分組件崩潰時，不會影響到整個系統;
- 保證消息順序執行，解決特定場景業務需求 ;
5、專業術語簡介
- Broker
一台kafka服務器就是一個broker。一個集群由多個broker組成。一個broker可以容納多個topic。
- Producer
消息生產者，就是向kafka broker發消息的客戶端。
- Consumer
消息消費者，向kafka broker取消息的客戶端。
- Topic
每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic，可以理解為一個隊列。
- Consumer Group
每個Consumer屬於一個特定的Consumer Group，可為每個Consumer指定group name，若不指定group name則屬於默認的分組。
- Partition
一個龐大大的topic可以分布到多個broker上，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id。kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的整體的順序。Partition是物理上的概念，方便在集群中擴展，提高並發。
三、整合SpringBoot2框架
1、案例結構
-
消息生產者 :
kafka-producer-server -
消息消費方 :
kafka-consumer-server
2、基礎依賴
<!-- SpringBoot依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- kafka 依賴 -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.2.4.RELEASE</version>
</dependency>
3、生產者配置
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
4、消息生成
@RestController
public class ProducerWeb {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
@RequestMapping("/send")
public String sendMsg () {
MsgLog msgLog = new MsgLog(1,"消息生成",
1,"消息日志",new Date()) ;
String msg = JSON.toJSONString(msgLog) ;
// 這里Topic如果不存在，會自動創建
kafkaTemplate.send("cicada-topic", msg);
return msg ;
}
}
5、消費者配置
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
consumer:
group-id: test-consumer-group
6、消息消費
@Component
public class ConsumerMsg {
private static Logger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class);
@KafkaListener(topics = "cicada-topic")
public void listenMsg (ConsumerRecord<?,String> record) {
String value = record.value();
LOGGER.info("ConsumerMsg====>>"+value);
}
}
四、消息流程分析
1、生產者分析
- 寫入方式
生產者基於推push推模式將消息發布到broker，每條消息都被追加到分區patition中，屬於磁盤順序寫，效率比隨機寫內存要高，保障kafka高吞吐量。
- 分區概念
消息發送時都被發送到一個topic，而topic是由Partition Logs(分區日志)組成，其組織結構如下圖所示：
每個Partition中的消息都是有序的，生產的消息被不斷追加到Partitionlog上，其中的每一個消息都被賦予了一個唯一的offset值。每個Partition可以通過調整以適配它所在的機器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應任意大小的數據。分區的原則：指定patition，則直接使用；未指定patition但指定key，通過對key的value進行hash出一個patition；patition和key都未指定，使用輪詢選出一個patition。
2、消費者分析
- 消費圖解
消費者是以consumer group消費者組的方式工作，由一個或者多個消費者組成一個組，共同消費一個topic。每個分區在同一時間只能由group中的一個消費者讀取，但是多個group可以同時消費一個partition。
- 消費方式
消費者采用pull拉模式從broker中讀取數據。對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費消息的速率，同時consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的數據傳輸場景。
五、源代碼地址
GitHub·地址
https://github.com/cicadasmile/middle-ware-parent
GitEE·地址
https://gitee.com/cicadasmile/middle-ware-parent
推薦閱讀：SpringBoot2整合中間件
《整合 ElasticSearch框架,實現高性能搜索引擎》