SpringBoot2中怎么整合Kafka組件，相信很多沒(méi)有經(jīng)驗(yàn)的人對(duì)此束手無(wú)策，為此本文總結(jié)了問(wèn)題出現(xiàn)的原因和解決方法，通過(guò)這篇文章希望你能解決這個(gè)問(wèn)題。

創(chuàng)新互聯(lián)公司專注于企業(yè)營(yíng)銷型網(wǎng)站建設(shè)、網(wǎng)站重做改版、大名網(wǎng)站定制設(shè)計(jì)、自適應(yīng)品牌網(wǎng)站建設(shè)、H5技術(shù)、商城開(kāi)發(fā)、集團(tuán)公司官網(wǎng)建設(shè)、外貿(mào)網(wǎng)站建設(shè)、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁(yè)設(shè)計(jì)等建站業(yè)務(wù)，價(jià)格優(yōu)惠性價(jià)比高，為大名等各大城市提供網(wǎng)站開(kāi)發(fā)制作服務(wù)。

一、搭建Kafka環(huán)境

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、啟動(dòng)Kafka服務(wù)

kafka依賴ZooKeeper服務(wù)，需要本地安裝并啟動(dòng)ZooKeeper。

參考文章： Linux系統(tǒng)搭建ZooKeeper3.4中間件，常用命令總結(jié)

-- 執(zhí)行位置
-- /usr/local/mysoft/kafka2.11
bin/kafka-server-start.sh config/server.properties

3、查看服務(wù)

ps -aux |grep kafka

4、開(kāi)放地址端口

-- 基礎(chǔ)路徑
-- /usr/local/mysoft/kafka2.11/config
vim server.properties
-- 添加下面注釋
advertised.listeners=PLAINTEXT://192.168.72.130:9092

二、Kafka基礎(chǔ)概念

1、基礎(chǔ)描述

Kafka是由Apache開(kāi)源，具有分布式、分區(qū)的、多副本的、多訂閱者，基于Zookeeper協(xié)調(diào)的分布式處理平臺(tái)，由Scala和Java語(yǔ)言編寫(xiě)。通常用來(lái)搜集用戶在應(yīng)用服務(wù)中產(chǎn)生的動(dòng)作日志數(shù)據(jù)，并高速的處理。日志類的數(shù)據(jù)需要高吞吐量的性能要求，對(duì)于像Hadoop一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka的目的是通過(guò)Hadoop的并行加載機(jī)制來(lái)統(tǒng)一線上和離線的消息處理，也是為了通過(guò)集群來(lái)提供實(shí)時(shí)的消息。

2、功能特點(diǎn)

(1)、通過(guò)磁盤數(shù)據(jù)結(jié)構(gòu)提供消息的持久化，消息存儲(chǔ)也能夠保持長(zhǎng)時(shí)間穩(wěn)定性;

(2)、高吞吐量，即使是非常普通的硬件Kafka也可以支持每秒超高的并發(fā)量;

(3)、支持通過(guò)Kafka服務(wù)器和消費(fèi)機(jī)集群來(lái)分區(qū)消息;

(4)、支持Hadoop并行數(shù)據(jù)加載;

(5)、API包封裝的非常好，簡(jiǎn)單易用，上手快 ;

(6)、分布式消息隊(duì)列。Kafka對(duì)消息保存時(shí)根據(jù)Topic進(jìn)行歸類，發(fā)送消息者稱為Producer，消息接受者稱為Consumer;

3、消息功能

• 點(diǎn)對(duì)點(diǎn)模式

點(diǎn)對(duì)點(diǎn)模型通常是一個(gè)基于拉取或者輪詢的消息傳遞模型，消費(fèi)者主動(dòng)拉取數(shù)據(jù)，消息收到后從隊(duì)列移除消息，這種模型不是將消息推送到客戶端，而是從隊(duì)列中請(qǐng)求消息。特點(diǎn)是發(fā)送到隊(duì)列的消息被一個(gè)且只有一個(gè)消費(fèi)者接收處理，即使有多個(gè)消費(fèi)者監(jiān)聽(tīng)隊(duì)列也是如此。

• 發(fā)布訂閱模式

發(fā)布訂閱模型則是一個(gè)基于推送的消息傳送模型，消息產(chǎn)生后，推送給所有訂閱者。發(fā)布訂閱模型可以有多種不同的訂閱者，臨時(shí)訂閱者只在主動(dòng)監(jiān)聽(tīng)主題時(shí)才接收消息，而持久訂閱者則監(jiān)聽(tīng)主題的所有消息，即使當(dāng)前訂閱者不可用，處于離線狀態(tài)。

4、消息隊(duì)列作用

• 程序解耦，生產(chǎn)者和消費(fèi)者獨(dú)立，各自異步執(zhí)行;

• 消息數(shù)據(jù)進(jìn)行持久化存儲(chǔ)，直到被全部消費(fèi)，規(guī)避了數(shù)據(jù)丟失風(fēng)險(xiǎn);

• 流量削峰，使用消息隊(duì)列承接訪問(wèn)壓力，盡量避免程序雪崩 ;

• 降低進(jìn)程間的耦合度，系統(tǒng)部分組件崩潰時(shí)，不會(huì)影響到整個(gè)系統(tǒng);

• 保證消息順序執(zhí)行，解決特定場(chǎng)景業(yè)務(wù)需求 ;

5、專業(yè)術(shù)語(yǔ)簡(jiǎn)介

• Broker

一臺(tái)kafka服務(wù)器就是一個(gè)broker。一個(gè)集群由多個(gè)broker組成。一個(gè)broker可以容納多個(gè)topic。

• Producer

消息生產(chǎn)者，就是向kafka broker發(fā)消息的客戶端。

• Consumer

消息消費(fèi)者，向kafka broker取消息的客戶端。

• Topic

每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為Topic，可以理解為一個(gè)隊(duì)列。

• Consumer Group

每個(gè)Consumer屬于一個(gè)特定的Consumer Group，可為每個(gè)Consumer指定group name，若不指定group name則屬于默認(rèn)的分組。

• Partition

一個(gè)龐大大的topic可以分布到多個(gè)broker上，一個(gè)topic可以分為多個(gè)partition，每個(gè)partition是一個(gè)有序的隊(duì)列。partition中的每條消息都會(huì)被分配一個(gè)有序的id。kafka只保證按一個(gè)partition中的順序?qū)⑾l(fā)給consumer，不保證一個(gè)topic的整體的順序。Partition是物理上的概念，方便在集群中擴(kuò)展，提高并發(fā)。

三、整合SpringBoot2框架

1、案例結(jié)構(gòu)

• 消息生產(chǎn)者 : kafka-producer-server

• 消息消費(fèi)方 : kafka-consumer-server

2、基礎(chǔ)依賴

    org.springframework.boot
    spring-boot-starter-web



    org.springframework.kafka
    spring-kafka
    2.2.4.RELEASE

3、生產(chǎn)者配置

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092

4、消息生成

@RestController
public class ProducerWeb {
    @Resource
    private KafkaTemplate kafkaTemplate;
    @RequestMapping("/send")
    public String sendMsg () {
        MsgLog msgLog = new MsgLog(1,"消息生成",
                                 1,"消息日志",new Date()) ;
        String msg = JSON.toJSONString(msgLog) ;
        // 這里Topic如果不存在，會(huì)自動(dòng)創(chuàng)建
        kafkaTemplate.send("cicada-topic", msg);
        return msg ;
    }
}

5、消費(fèi)者配置

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: test-consumer-group

6、消息消費(fèi)

@Component
public class ConsumerMsg {
    private static Logger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class);
    @KafkaListener(topics = "cicada-topic")
    public void listenMsg (ConsumerRecord record) {
        String value = record.value();
        LOGGER.info("ConsumerMsg====>>"+value);
    }
}

四、消息流程分析

1、生產(chǎn)者分析

• 寫(xiě)入方式

生產(chǎn)者基于推push推模式將消息發(fā)布到broker，每條消息都被追加到分區(qū)patition中，屬于磁盤順序?qū)?，效率比隨機(jī)寫(xiě)內(nèi)存要高，保障kafka高吞吐量。

• 分區(qū)概念

消息發(fā)送時(shí)都被發(fā)送到一個(gè)topic，而topic是由Partition Logs(分區(qū)日志)組成，其組織結(jié)構(gòu)如下圖所示：

每個(gè)Partition中的消息都是有序的，生產(chǎn)的消息被不斷追加到Partitionlog上，其中的每一個(gè)消息都被賦予了一個(gè)唯一的offset值。每個(gè)Partition可以通過(guò)調(diào)整以適配它所在的機(jī)器，而一個(gè)topic又可以有多個(gè)Partition組成，因此整個(gè)集群就可以適應(yīng)任意大小的數(shù)據(jù)。分區(qū)的原則：指定patition，則直接使用；未指定patition但指定key，通過(guò)對(duì)key的value進(jìn)行hash出一個(gè)patition；patition和key都未指定，使用輪詢選出一個(gè)patition。

2、消費(fèi)者分析

• 消費(fèi)圖解

消費(fèi)者是以consumer group消費(fèi)者組的方式工作，由一個(gè)或者多個(gè)消費(fèi)者組成一個(gè)組，共同消費(fèi)一個(gè)topic。每個(gè)分區(qū)在同一時(shí)間只能由group中的一個(gè)消費(fèi)者讀取，但是多個(gè)group可以同時(shí)消費(fèi)一個(gè)partition。

• 消費(fèi)方式

消費(fèi)者采用pull拉模式從broker中讀取數(shù)據(jù)。對(duì)于Kafka而言，pull模式更合適，它可簡(jiǎn)化broker的設(shè)計(jì)，consumer可自主控制消費(fèi)消息的速率，同時(shí)consumer可以自己控制消費(fèi)方式——即可批量消費(fèi)也可逐條消費(fèi)，同時(shí)還能選擇不同的提交方式從而實(shí)現(xiàn)不同的數(shù)據(jù)傳輸場(chǎng)景。

看完上述內(nèi)容，你們掌握SpringBoot2中怎么整合Kafka組件的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝各位的閱讀！