目錄

一、背景和描述

二、資源情況

三、技術選型

四、部署Kraft版本集群

五、配置SSL模式

六、Springboot使用SSL集成

參考資料

一、背景和描述

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

Apache Kafka Raft 是一種共識協議，它的引入是為了消除 Kafka 對 ZooKeeper 的元數據管理的依賴，被社區稱之為 Kafka Raft metadata mode，簡稱 KRaft 模式。

目前，Kafka在使用的過程當中，會出現一些問題。由于重度依賴Zookeeper集群，當Zookeeper集群性能發生抖動時，Kafka的性能也會收到很大的影響。因此，在Kafka發展的過程當中，為了解決這個問題，提供KRaft模式，來取消Kafka對Zookeeper的依賴。參考文章：百度安全驗證

二、資源情況

服務器IPCPU內存磁盤容量

127.0.0.1	8核	16G	100G
127.0.0.1	8核	16G	100G
127.0.0.1	8核	16G	100G

以下是查看Linux機器CPU核心數、內存、磁盤容量等信息的常用命令：

查看CPU核心數：

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

查看內存大小以及可用空間：

free -h

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

df -h

查看某個目錄的占用空間：

du -sh /path/to/directory

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

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

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

而physical id則表示物理處理器的ID號。在多CPU架構中，每個CPU都有一個唯一的物理ID。如果一臺服務器有多個物理CPU，則每個物理CPU將會被分配一個不同的physical id。這些信息對于了解如何將進程/線程分配到物理CPU上非常有用。

總之，cpu cores通常用于監視和調整單個物理CPU的性能，而physical id則用于檢查多個物理CPU之間的區別以及了解如何在多CPU環境下分配資源。

三、技術選型

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

Apache 軟件基金會發布了包含許多新特性和改進的 Kafka 3.3.1。這是第一個標志著可以在生產環境中使用 KRaft（Kafka Raft）共識協議的版本。在幾年的開發過程中，它先是在 Kafka 2.8 早期訪問版本中發布，然后又在 Kafka 3.0 預覽版本中發布。

KRaft 是一種共識協議，可以直接在 Kafka 中管理元數據。元數據的管理被整合到了 Kafka 當中，而不需要使用像 ZooKeeper 這樣的第三方工具，這大大簡化了 Kafka 的架構。這種新的 KRaft 模式提高了分區的可伸縮性和彈性，同時簡化了 Kafka 的部署，現在可以不依賴 ZooKeeper 單獨部署 Kafka 了。

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

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

四、部署Kraft版本集群

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

五、配置SSL模式

如何配置kafka的SSL鏈接訪問 - 簡書

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

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

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

配置 Kafka Broker 在 broker.properties 文件中，將以下參數設置為所需值：

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

將 SSL 監聽器和廣告監聽器設置為 SSL://<hostname>:9093。請相應地更改 <keystore-password> 和 <key-password> 以及文件路徑等參數為實際的內容。

啟用 SSL 客戶端身份驗證（可選） 如果需要對客戶端進行身份驗證，也可以在 broker.properties 中設置以下參數：

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

這些參數指定了 SSL 客戶端身份驗證策略，以及必須使用的可信證書頒發機構（CA）證書的位置和密碼等信息。指定 ssl.client.auth=required 可以強制 Kafka Broker 對客戶端進行身份驗證，并防止未經授權的訪問。

在完成上述步驟后，重新啟動 Kafka Broker 。現在，您可以使用 SSL 連接到 Kafka 集群。但是，在實際生產環境中，請首先測試配置并確保其可以正常工作，然后再將其應用于生產環境。

要在 Kafka 客戶端上創建一個 client-ssl.properties 文件進行測試 SSL 連接，您可以按照以下步驟操作：

在客戶端機器上安裝 JRE (Java 運行環境) 和 OpenSSL 工具 (如果沒有安裝的話)。

生成客戶端證書和私鑰 這個證書應該是使用與 Kafka Broker 使用的相同的 CA 簽署的。也可以為您的測試客戶端生成自簽名證書以進行測試。可以使用 OpenSSL 工具來生成自簽名證書并為其設置密碼，例如：

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

創建 client-ssl.properties 文件 在與 Kafka 客戶端可執行文件相同的目錄中，根據需要創建一個新的 client-ssl.properties 文件，并將以下參數設置為所需值：

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> 替換為實際的證書和密鑰存儲路徑，以及相應的密碼。

向 Kafka 發送測試消息 在與 Kafka 客戶端可執行文件相同的目錄中，使用以下命令向 Kafka Broker 發送測試消息：

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

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

檢查發送是否成功 您可以在另一個終端中使用以下命令從 Kafka Broker 消費消息：

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

如果一切正常，您將看到來自生產者的消息已成功傳輸并已由消費者接收到。

通過按照上述步驟創建 client-ssl.properties 文件并進行測試 SSL 連接，您可以確保 Kafka 集群上的 SSL 連接設置正確且正常工作。

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

生成 CA 根證書。

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

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

創建一個用于 IP 的證書簽名請求（CSR）文件。

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

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

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

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

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

使用新的證書和密鑰保護 IP

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

請注意，在使用 openssl 命令生成證書時，要特別注意您輸入的各種數據信息因素，因為這些信息會在后續內容以及過程中得到反復使用。

六、Springboot使用SSL集成

SpringBoot整合并簡單使用SSL認證的Kafka_springboot ssl kafka_justry_deng的博客-CSDN博客

在 Spring Kafka 中，可以通過配置 KafkaTemplate 和 ProducerFactory 來實現 SSL 驗證。下面是一個簡單的配置示例：

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;@Beanpublic 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);}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());} }

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

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

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

kafka.bootstrap.servers=kafka.example.com:9093 kafka.ssl.truststore.location=/path/to/truststore.jks kafka.ssl.truststore.password=password 需要注意的是，在 SSL 模式下，Kafka 的安全驗證非常關鍵。為了保護數據安全，在實際場景中可能還需要采用其他的安全措施，例如身份驗證、消息加密等。具體措施可以根據企業的實際情況進行確定。

參考資料

Kafka實戰：集群SSL加密認證和配置（最新版kafka-2.7.0）

78.kafka高級-kraft集群搭建_嗶哩嗶哩_bilibili

[教程]【海牛大數據】Kafka 教程（Kafka3.3.2 原理、安裝、應用場景、源碼解析、高級優化、實戰案例應有盡有）[云平臺課程] | 海牛部落 高品質的 大數據技術社區

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認證_ssl.ca.location kafka_justry_deng的博客-CSDN博客

總結

以上是生活随笔為你收集整理的【Kafka】Kafka3.3.1集群搭建指南KRaft版本的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。