一、Kafka相關(guān)概念

1、關(guān)于Kafka的描述

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、關(guān)于Kafka的功能特點(diǎn)

1. 通過(guò)磁盤(pán)數(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（生產(chǎn)者），消息接受者稱為Consumer（消費(fèi)者）;

3、Kafka消息功能

如下圖所示，Kafka作為一個(gè)中間服務(wù)，代表一個(gè)broker（經(jīng)紀(jì)人）角色，負(fù)責(zé)接收APP的消費(fèi)與推送消息給其他相關(guān)APP。這里APP可分為Producer，Consumer。

消息的消費(fèi)模式

點(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ā)布訂閱模式：訂閱模式是一個(gè)基于推送的消費(fèi)傳送模型，消息產(chǎn)生后，Kafka會(huì)推送給所有訂閱相關(guān)Topic的訂閱者。發(fā)布訂閱模型可以有多種不同的訂閱者，臨時(shí)訂閱者只在主動(dòng)監(jiān)聽(tīng)主題時(shí)才接收消息，而持久訂閱者則監(jiān)聽(tīng)主題的所有消息，即使當(dāng)前訂閱者不可用，處于離線狀態(tài)。

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

• 應(yīng)用程序之間解耦，生產(chǎn)者與消費(fèi)者相互獨(dú)立，各自異步執(zhí)行。
• 消息數(shù)據(jù)持久化存儲(chǔ)，直到所有消息都被消費(fèi)，規(guī)避消息數(shù)據(jù)丟失的風(fēng)險(xiǎn)。
• 流量削峰，使用Kafka消息隊(duì)列可以幫助server承接訪問(wèn)壓力，盡可能避免應(yīng)用程序崩潰。
• 降低進(jìn)程間的耦合度，系統(tǒng)部分應(yīng)用組件發(fā)生崩潰時(shí)，不會(huì)影響到整體系統(tǒng)的運(yùn)行。
• 保證消息順序執(zhí)行，解決特定場(chǎng)景業(yè)務(wù)需求。

5、Kafka相關(guān)術(shù)語(yǔ)介紹

• Broker

? ?一臺(tái)kafka服務(wù)器就是一個(gè)broker（經(jīng)紀(jì)人）。一個(gè)集群由多個(gè)broker組成。一個(gè)broker可以容納多個(gè)topic（消息主題）。

• Producer

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

• Consumer

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

• Topic

? ? 每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為T(mén)opic，可以理解為一個(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ā)。

二、liunx系統(tǒng)下搭建Kafka環(huán)境

? ? ? ?

--新建kafka應(yīng)用目錄。并下載到當(dāng)前目錄下
cd  /usr/localmkdir kafkacd kafka 
--下載wget https://downloads.apache.org/kafka/3.7.0/kafka-3.7.0-src.tgz--解壓tar -zxvf  kafka-3.7.0-src.tgz--啟動(dòng)服務(wù)cd kafka-3.7.0./bin/kafka-server-start.sh    config/server.properties--查看服務(wù)ps -aux |grep kafka--開(kāi)放kafka地址端口vim server.properties--添加下面注釋advertised.listeners=PLAINTEXT://10.98.3.22:9092

三、Springboot2整合Kafka 服務(wù)

1、導(dǎo)入基礎(chǔ)依賴

<!-- 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>

2、項(xiàng)目目錄結(jié)構(gòu)

3、生產(chǎn)者與消費(fèi)者yml文件配置

#消費(fèi)者配置
spring:kafka:bootstrap-servers: 127.0.0.1:9092consumer:group-id: test-consumer-group#生產(chǎn)者配置spring:kafka:bootstrap-servers: 127.0.0.1:9092

4、生成消息

@RestController
@RequestMapping("/kafka")
public class ProducerController {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;@RequestMapping("/send")public HttpResult 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 HttpResult.create(HttpStatus.SUCCESS,msg);}
}

@Component
public class ConsumerMsg {private static Logger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class);//此注解是監(jiān)聽(tīng)主題為cicada-topic的消息隊(duì)列@KafkaListener(topics = "cicada-topic")public void listenMsg (ConsumerRecord<?,String> record) {String value = record.value();LOGGER.info("ConsumerMsg====>>"+value);}
}