SpringBoot2 整合Kafka組件，應用案例和流程詳解

1、搭建Kafka環境一、下載解壓-- 下載wget http://mirror.bit.edu.cn/apac..._2.11-2.2.0.tgz-- 解壓tar -zxvf kafka_2.11-2.2.0.tgz-- 重命名mv kafka_2.11-2.2.0 kafka2.11二、啓動Kafka服務

kafka依賴ZooKeeper服務，須要本地安裝並啓動ZooKeeper。

參考文章： Linux系統搭建ZooKeeper3.4中間件，經常使用命令總結

-- 執行位置-- /usr/local/mysoft/kafka2.11bin/kafka-server-start.shconfig/server.properties三、查看服務ps -aux|grep kafka四、開放地址端口-- 基礎路徑-- /usr/local/mysoft/kafka2.11/configvim server.properties-- 添加下面註釋advertised.listeners=PLAINTEXT://192.168.72.130:90922、Kafka基礎概念一、基礎描述

Kafka是由Apache開源，具備分佈式、分區的、多副本的、多訂閱者，基於Zookeeper協調的分佈式處理平臺，由Scala和Java語言編寫。一般用來蒐集用戶在應用服務中產生的動做日誌數據，並高速的處理。日誌類的數據須要高吞吐量的性能要求，對於像Hadoop同樣的日誌數據和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。Kafka的目的是經過Hadoop的並行加載機制來統一線上和離線的消息處理，也是爲了經過集羣來提供實時的消息。

二、功能特色

(1)、經過磁盤數據結構提供消息的持久化，消息存儲也可以保持長時間穩定性;

(2)、高吞吐量，即便是很是普通的硬件Kafka也能夠支持每秒超高的併發量;

(3)、支持經過Kafka服務器和消費機集羣來分區消息;

(4)、支持Hadoop並行數據加載;

(5)、API包封裝的很是好，簡單易用，上手快 ;

(6)、分佈式消息隊列。Kafka對消息保存時根據Topic進行歸類，發送消息者稱爲Producer，消息接受者稱爲Consumer;

三、消息功能

• 點對點模式

點對點模型一般是一個基於拉取或者輪詢的消息傳遞模型，消費者主動拉取數據，消息收到後從隊列移除消息，這種模型不是將消息推送到客戶端，而是從隊列中請求消息。特色是發送到隊列的消息被一個且只有一個消費者接收處理，即便有多個消費者監聽隊列也是如此。

• 發佈訂閱模式

發佈訂閱模型則是一個基於推送的消息傳送模型，消息產生後，推送給全部訂閱者。發佈訂閱模型能夠有多種不一樣的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的全部消息，即便當前訂閱者不可用，處於離線狀態。

四、消息隊列做用

• 程序解耦，生產者和消費者獨立，各自異步執行;
• 消息數據進行持久化存儲，直到被所有消費，規避了數據丟失風險;
• 流量削峯，使用消息隊列承接訪問壓力，儘可能避免程序雪崩 ;
• 下降進程間的耦合度，系統部分組件崩潰時，不會影響到整個系統;
• 保證消息順序執行，解決特定場景業務需求 ;

五、專業術語簡介

• 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是物理上的概念，方便在集羣中擴展，提升併發。

3、整合SpringBoot2框架一、案例結構

• 消息生產者 : kafka-producer-server
• 消息消費方 : kafka-consumer-server

二、基礎依賴<!-- 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>三、生產者配置spring:  kafka:   bootstrap-servers: 127.0.0.1:9092四、消息生成@RestControllerpublicclass ProducerWeb {   @ResourceprivateKafkaTemplate<String, String> kafkaTemplate;   @RequestMapping("/send")    public String sendMsg () {        MsgLog msgLog =newMsgLog(1,"消息生成",                                 1,"消息日誌",newDate()) ;        String msg = JSON.toJSONString(msgLog) ;        // 這裏Topic若是不存在，會自動建立        kafkaTemplate.send("cicada-topic", msg);        returnmsg ;    }}五、消費者配置spring:  kafka:   bootstrap-servers: 127.0.0.1:9092   consumer:      group-id: test-consumer-group六、消息消費@Componentpublicclass ConsumerMsg {   privatestaticLogger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class);   @KafkaListener(topics ="cicada-topic")    public void listenMsg (ConsumerRecord<?,String> record) {        String value = record.value();        LOGGER.info("ConsumerMsg====>>"+value);    }}4、消息流程分析一、生產者分析

• 寫入方式

生產者基於推push推模式將消息發佈到broker，每條消息都被追加到分區patition中，屬於磁盤順序寫，效率比隨機寫內存要高，保障kafka高吞吐量。

• 分區概念

消息發送時都被髮送到一個topic，而topic是由Partition Logs(分區日誌)組成，其組織結構以下圖所示：

每一個Partition中的消息都是有序的，生產的消息被不斷追加到Partitionlog上，其中的每個消息都被賦予了一個惟一的offset值。每一個Partition能夠經過調整以適配它所在的機器，而一個topic又能夠有多個Partition組成，所以整個集羣就能夠適應任意大小的數據。分區的原則：指定patition，則直接使用；未指定patition但指定key，經過對key的value進行hash出一個patition；patition和key都未指定，使用輪詢選出一個patition。

二、消費者分析

• 消費圖解

消費者是以consumer group消費者組的方式工做，由一個或者多個消費者組成一個組，共同消費一個topic。每一個分區在同一時間只能由group中的一個消費者讀取，可是多個group能夠同時消費一個partition。

• 消費方式

消費者採用pull拉模式從broker中讀取數據。對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費消息的速率，同時consumer能夠本身控制消費方式——便可批量消費也可逐條消費，同時還能選擇不一樣的提交方式從而實現不一樣的數據傳輸場景。

5、源代碼地址GitHub·地址https://github.com/cicadasmil...·地址https://gitee.com/cicadasmile...