目錄

一、背景和描述

二、資源情況

三、技術(shù)選型

四、部署Kraft版本集群

五、配置SSL模式

六、Springboot使用SSL集成

參考資料

一、背景和描述

考慮資源安全性，需要搭建不依賴Zookeeper的kafka集群環(huán)境，并且配置SSL訪問控制

Apache Kafka Raft 是一種共識(shí)協(xié)議，它的引入是為了消除 Kafka 對(duì) ZooKeeper 的元數(shù)據(jù)管理的依賴，被社區(qū)稱之為 Kafka Raft metadata mode，簡(jiǎn)稱 KRaft 模式。

目前，Kafka在使用的過程當(dāng)中，會(huì)出現(xiàn)一些問題。由于重度依賴Zookeeper集群，當(dāng)Zookeeper集群性能發(fā)生抖動(dòng)時(shí)，Kafka的性能也會(huì)收到很大的影響。因此，在Kafka發(fā)展的過程當(dāng)中，為了解決這個(gè)問題，提供KRaft模式，來取消Kafka對(duì)Zookeeper的依賴。參考文章：百度安全驗(yàn)證

二、資源情況

以下是查看Linux機(jī)器CPU核心數(shù)、內(nèi)存、磁盤容量等信息的常用命令：

1. 查看CPU核心數(shù)：

   cat /proc/cpuinfo | grep "cpu cores" | uniq
   cat /proc/cpuinfo | grep "physical id" | sort | uniq | wc -l
   

2. 查看內(nèi)存大小以及可用空間：

   free -h
   

3. 查看磁盤容量以及可用空間：

   df -h
   

4. 查看某個(gè)目錄的占用空間：

   du -sh /path/to/directory
   

以上命令可以通過SSH連接到Linux服務(wù)器的終端或使用工具如SecureCRT、PuTTY等來執(zhí)行。

在Linux系統(tǒng)中，cpu cores和physical id都是用來表示CPU核心數(shù)量的信息。但是，它們有不同的含義和用途。

cpu cores表示每個(gè)物理CPU上的核心數(shù)量。例如，如果一臺(tái)服務(wù)器有兩個(gè)物理CPU，每個(gè)CPU包含4個(gè)核心，則cpu cores的值為8（即 2 * 4 = 8） 。此信息對(duì)于診斷單個(gè)物理CPU的性能問題非常有用，可以幫助您確定每個(gè)物理CPU上的核心數(shù)。

而physical id則表示物理處理器的ID號(hào)。在多CPU架構(gòu)中，每個(gè)CPU都有一個(gè)唯一的物理ID。如果一臺(tái)服務(wù)器有多個(gè)物理CPU，則每個(gè)物理CPU將會(huì)被分配一個(gè)不同的physical id。這些信息對(duì)于了解如何將進(jìn)程/線程分配到物理CPU上非常有用。

總之，cpu cores通常用于監(jiān)視和調(diào)整單個(gè)物理CPU的性能，而physical id則用于檢查多個(gè)物理CPU之間的區(qū)別以及了解如何在多CPU環(huán)境下分配資源。

三、技術(shù)選型

查閱Kafka官方說明文檔：Apache Kafka，3.3.x版本Kafka可用于生產(chǎn)，所以Kafka版本選擇3.3.1（3.3.0不推薦）

Apache 軟件基金會(huì)發(fā)布了包含許多新特性和改進(jìn)的 Kafka 3.3.1。這是第一個(gè)標(biāo)志著可以在生產(chǎn)環(huán)境中使用 KRaft（Kafka Raft）共識(shí)協(xié)議的版本。在幾年的開發(fā)過程中，它先是在 Kafka 2.8 早期訪問版本中發(fā)布，然后又在 Kafka 3.0 預(yù)覽版本中發(fā)布。

KRaft 是一種共識(shí)協(xié)議，可以直接在 Kafka 中管理元數(shù)據(jù)。元數(shù)據(jù)的管理被整合到了 Kafka 當(dāng)中，而不需要使用像 ZooKeeper 這樣的第三方工具，這大大簡(jiǎn)化了 Kafka 的架構(gòu)。這種新的 KRaft 模式提高了分區(qū)的可伸縮性和彈性，同時(shí)簡(jiǎn)化了 Kafka 的部署，現(xiàn)在可以不依賴 ZooKeeper 單獨(dú)部署 Kafka 了。

參考資料：Kafka 3.3使用KRaft共識(shí)協(xié)議替代ZooKeeper_控制器_仲裁_版本

下載地址：Kafka官方下載地址

四、部署Kraft版本集群

kafka筆記3--快速部署KRaft版本的kafka3.1.1_kafka 3.1 kraft 安裝_昕光xg的博客-CSDN博客

五、配置SSL模式

如何配置kafka的SSL鏈接訪問 - 簡(jiǎn)書

在 Kafka 3.3.1 中啟用 Kraft 模式下的 SSL 加密傳輸，您可以按照以下步驟進(jìn)行操作：

1. 生成 SSL 證書和私鑰 這可以使用 OpenSSL 工具來完成。可以根據(jù)需要自定義證書和密鑰文件名稱以及密碼等相關(guān)參數(shù)，例如：

   openssl req -new -newkey rsa:4096 -days 365 -nodes -x509 -keyout server.key -out server.crt

2. 配置 Kafka Broker 在 broker.properties 文件中，將以下參數(shù)設(shè)置為所需值：

   listeners=SSL://kafka-ssl.example.com:9093 # 定義 SSL 監(jiān)聽器 advertised.listeners=SSL://kafka-ssl.example.com:9093 # 公告 SSL 監(jiān)聽器 ssl.keystore.location=/path/to/kafka.server.keystore.jks ssl.keystore.password=<keystore-password> ssl.key.password=<key-password>

   將 SSL 監(jiān)聽器和廣告監(jiān)聽器設(shè)置為 SSL://<hostname>:9093。請(qǐng)相應(yīng)地更改 <keystore-password> 和 <key-password> 以及文件路徑等參數(shù)為實(shí)際的內(nèi)容。

3. 啟用 SSL 客戶端身份驗(yàn)證（可選） 如果需要對(duì)客戶端進(jìn)行身份驗(yàn)證，也可以在 broker.properties 中設(shè)置以下參數(shù)：

   ssl.client.auth=required ssl.truststore.location=/path/to/truststore.jks ssl.truststore.password=<truststore-password>

   這些參數(shù)指定了 SSL 客戶端身份驗(yàn)證策略，以及必須使用的可信證書頒發(fā)機(jī)構(gòu)（CA）證書的位置和密碼等信息。指定 ssl.client.auth=required 可以強(qiáng)制 Kafka Broker 對(duì)客戶端進(jìn)行身份驗(yàn)證，并防止未經(jīng)授權(quán)的訪問。

在完成上述步驟后，重新啟動(dòng) Kafka Broker ?，F(xiàn)在，您可以使用 SSL 連接到 Kafka 集群。但是，在實(shí)際生產(chǎn)環(huán)境中，請(qǐng)首先測(cè)試配置并確保其可以正常工作，然后再將其應(yīng)用于生產(chǎn)環(huán)境。

要在 Kafka 客戶端上創(chuàng)建一個(gè) client-ssl.properties 文件進(jìn)行測(cè)試 SSL 連接，您可以按照以下步驟操作：

1. 在客戶端機(jī)器上安裝 JRE (Java 運(yùn)行環(huán)境) 和 OpenSSL 工具 (如果沒有安裝的話)。

2. 生成客戶端證書和私鑰 這個(gè)證書應(yīng)該是使用與 Kafka Broker 使用的相同的 CA 簽署的。也可以為您的測(cè)試客戶端生成自簽名證書以進(jìn)行測(cè)試?？梢允褂?OpenSSL 工具來生成自簽名證書并為其設(shè)置密碼，例如：

   openssl req -new -newkey rsa:4096 -nodes -keyout client.key -out client.csr openssl x509 -req -days 365 -in client.csr -CA rootCA.crt -CAkey rootCA.key -set_serial 01 -out client.crt

3. 創(chuàng)建 client-ssl.properties 文件 在與 Kafka 客戶端可執(zhí)行文件相同的目錄中，根據(jù)需要?jiǎng)?chuàng)建一個(gè)新的 client-ssl.properties 文件，并將以下參數(shù)設(shè)置為所需值：

   security.protocol=SSL ssl.truststore.location=<path-to-client-truststore.jks> ssl.truststore.password=<truststore-password> ssl.keystore.location=<path-to-client-keystore.jks> ssl.keystore.password=<keystore-password> ssl.key.password=<key-password>

   將 <path-to-client-truststore.jks> 和 <path-to-client-keystore.jks> 替換為實(shí)際的證書和密鑰存儲(chǔ)路徑，以及相應(yīng)的密碼。

4. 向 Kafka 發(fā)送測(cè)試消息 在與 Kafka 客戶端可執(zhí)行文件相同的目錄中，使用以下命令向 Kafka Broker 發(fā)送測(cè)試消息：

   bin/kafka-console-producer.sh --broker-list <kafka-broker>:9093 --topic test --producer.config client-ssl.properties

   將 <kafka-broker> 替換為您的 Kafka Broker 主機(jī)名或 IP 地址。

5. 檢查發(fā)送是否成功 您可以在另一個(gè)終端中使用以下命令從 Kafka Broker 消費(fèi)消息：

   bin/kafka-console-consumer.sh --bootstrap-server <kafka-broker>:9093 --topic test --from-beginning --consumer.config client-ssl.properties

   如果一切正常，您將看到來自生產(chǎn)者的消息已成功傳輸并已由消費(fèi)者接收到。

通過按照上述步驟創(chuàng)建 client-ssl.properties 文件并進(jìn)行測(cè)試 SSL 連接，您可以確保 Kafka 集群上的 SSL 連接設(shè)置正確且正常工作。

為 IP 生成 CA 證書的流程如下：

1. 生成 CA 根證書。

   openssl req -x509 -days 3650 -newkey rsa:2048 -nodes -keyout ca.key -out ca.crt

   這個(gè)命令將生成新的 CA 根證書，其中 -x509 表示生成自簽名證書，-days 3650 表示證書的有效期為 10 年。在這個(gè)過程中，您需要輸入一些信息，如國家代碼、城市名、組織名稱等。

2. 創(chuàng)建一個(gè)用于 IP 的證書簽名請(qǐng)求（CSR）文件。

   openssl req -new -nodes -newkey rsa:2048 -keyout server.key -out server.csr

   同樣要求輸入一些信息，例如國家代碼、城市名、公司名等。其中，-new 表示創(chuàng)建一個(gè)新的證書請(qǐng)求，-keyout 指定私鑰輸出的文件路徑，-out 指定新證書請(qǐng)求輸出的文件路徑。

3. 使用 CA 根證書簽署 CSR 文件并生成證書。

   openssl x509 -req -CA ca.crt -CAkey ca.key -in server.csr -out server.crt -days 3650 -CAcreateserial -extfile server.ext

   這個(gè)命令將使用 CA 根證書簽署剛才創(chuàng)建的證書請(qǐng)求，并生成新的服務(wù)器證書。其中，-CA 和 -CAkey 分別指定 CA 根證書和私鑰的位置，-in 指定服務(wù)器證書簽名請(qǐng)求的文件路徑，-out 指定生成的證書的文件路徑，-days 設(shè)置證書的有效期，-CAcreateserial 表示自動(dòng)創(chuàng)建一個(gè)序列號(hào)，并將其保存到 .srl 文件中。-extfile server.ext 是一個(gè)配置文件，在其中可以指定 SAN 擴(kuò)展等附加選項(xiàng)。

4. 使用新的證書和密鑰保護(hù) IP

   將生成的 server.crt 和 server.key 證書和私鑰文件分別應(yīng)用于您需要保護(hù)的 IP、域名或其他服務(wù)上，在 Internet 上建立安全連接。

請(qǐng)注意，在使用 openssl 命令生成證書時(shí)，要特別注意您輸入的各種數(shù)據(jù)信息因素，因?yàn)檫@些信息會(huì)在后續(xù)內(nèi)容以及過程中得到反復(fù)使用。

六、Springboot使用SSL集成

SpringBoot整合并簡(jiǎn)單使用SSL認(rèn)證的Kafka_springboot ssl kafka_justry_deng的博客-CSDN博客

在 Spring Kafka 中，可以通過配置 KafkaTemplate 和 ProducerFactory 來實(shí)現(xiàn) SSL 驗(yàn)證。下面是一個(gè)簡(jiǎn)單的配置示例：

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaProducerConfiguration {

    @Value("${kafka.bootstrap.servers}")
    private String bootstrapServers;

    @Value("${kafka.ssl.truststore.location}")
    private String truststoreLocation;

    @Value("${kafka.ssl.truststore.password}")
    private String truststorePassword;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststoreLocation);
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword);
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

在上面的代碼中，定義了以下參數(shù)：

• bootstrap.servers：指定 Kafka Broker 的地址和端口號(hào)。
• ssl.truststore.location：指定 SSL 驗(yàn)證所需的證書文件路徑。
• ssl.truststore.password：指定證書文件的密碼。
• security.protocol：指定安全協(xié)議類型為 SSL。
• key.serializer：指定鍵值序列化器類。
• value.serializer：指定值序列化器類。

這個(gè)示例中，使用 @Value 注解讀取了配置文件中的屬性值，以便在代碼中動(dòng)態(tài)使用。因此，在使用之前，需要在配置文件中添加如下屬性：

kafka.bootstrap.servers=kafka.example.com:9093
kafka.ssl.truststore.location=/path/to/truststore.jks
kafka.ssl.truststore.password=password

需要注意的是，在 SSL 模式下，Kafka 的安全驗(yàn)證非常關(guān)鍵。為了保護(hù)數(shù)據(jù)安全，在實(shí)際場(chǎng)景中可能還需要采用其他的安全措施，例如身份驗(yàn)證、消息加密等。具體措施可以根據(jù)企業(yè)的實(shí)際情況進(jìn)行確定。

參考資料

Kafka實(shí)戰(zhàn)：集群SSL加密認(rèn)證和配置（最新版kafka-2.7.0）

78.kafka高級(jí)-kraft集群搭建_嗶哩嗶哩_bilibili

[教程]【海牛大數(shù)據(jù)】Kafka 教程（Kafka3.3.2 原理、安裝、應(yīng)用場(chǎng)景、源碼解析、高級(jí)優(yōu)化、實(shí)戰(zhàn)案例應(yīng)有盡有）[云平臺(tái)課程] | 海牛部落 高品質(zhì)的 大數(shù)據(jù)技術(shù)社區(qū)

Apache Kafka 官方文檔

kafka筆記3--快速部署KRaft版本的kafka3.1.1_kafka 3.1 kraft 安裝_昕光xg的博客-CSDN博客

再見！Kafka決定棄用Zookeeper...

Apache Kafka 下載地址

kafka3.4.0集群搭建（無zookeeper）_最新版kafka不需要zookeeper_小趴菜醉了的博客-CSDN博客

kafka3.1集群搭建（kraft模式）_controller.quorum.voters_阿豪咿呀的博客-CSDN博客

Kafka配置SSL認(rèn)證_ssl.ca.location kafka_justry_deng的博客-CSDN博客文章來源地址http://www.zghlxwxcb.cn/news/detail-439020.html

到了這里，關(guān)于【Kafka】Kafka3.1.1集群搭建指南KRaft版本的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！