參數詳解listenersadvertised.listenerslistener.security.protocol.mapinter.broker.listener.namesecurity.inter.broker.protocolcontrol.plane.listener.name示例說明
幾種場景的配置方式1. 一臺機器部署一套集群2. 內網環境多機器部署集群3. 內網和外網分流3. 內網和外網和Controller分流

《Kafka運維管控平臺LogiKM》 ??更強大的管控能力?? 更高效的問題定位能力更便捷的集群運維能力更專業的資源治理更友好的運維生態

1參數詳解

listeners

偵聽器列表，這里配置的監聽器底層調用的是


ServerSocketAdaptor.bind(SocketAddress local)

那么這個說明什么意思呢？說明你配置的監聽器將被用于監聽網絡請求。簡單理解就是你建立監聽一個通道,別人能夠通過這個通道跟你溝通。所以我們需要設置 IP:Port.

這個屬性的格式為:

 listeners = listener_name://host_name:port,listener_name2://host_nam2e:port2

可以同時配置多個, 并且用逗號隔開
監聽器的名稱和端口必須是唯一的,端口相同,就沖突了
host_name如果為空,例如(listeners = ://host_name:port),則會綁定到默認的接口(網卡),一般情況下是localhost，底層調用的是JAVA.net.InetAddress.getCanonicalHostName()
將host_name設置為0.0.0.0 則會綁定所有的網卡, 也就是說不管從哪個網卡進入的請求都會被接受處理。但是請注意,假如你設置的是0.0.0.0,那么advertised.listeners 必須要設置,因為advertised.listeners默認請看下使用的是listeners的配置發布到zk中,發布到zk中是給其他Brokers/Clients 來跟你通信的,你設置0.0.0.0,誰知道要請求哪個IP呢, 所以它必須要指定并明確 IP:PORT。具體詳情請看下面 示例3
listener_name 是監聽名,唯一值, 他并不是安全協議(大部分人都會搞錯),因為默認的4個安全協議已經做好了映射, 例如：PLAINTEXT ==> PLAINTEXT . 所以你經常看到的配置## 這個PLAINTEXT是監聽名稱,剛好他對應的安全協議就是 PLAINTEXT
## 當然這個是可以自定義的, 詳細情況后面的配置listener.security.protocol.map
listeners = PLAINTEXT://your.host.name:9092
可動態配置該屬性

advertised.listeners

發布公開的監聽器, 啥叫發布公開的監聽器？就是,讓Brokers和Clients們都能夠知道的監聽器,你想想看,listeners是Broker用來監聽網絡請求的

那么,其他Broker或者客戶端想要與它通信,則需要知道具體的IP:PORT吧？所以,為了讓別人知道自己的監聽器,那么就需要公開出去,當然這個公開的形式,是通過zk來共享數據。

看看broker到zk節點/brokers/{brokerid}/ 下面的信息示例

{
 "features": {},
 "listener_security_protocol_map": {
  "PLAINTEXT": "PLAINTEXT"
 },
 "endpoints": ["PLAINTEXT://localhost:9092"],
 "jmx_port": -1,
 "port": 9092,
 "host": "localhost",
 "version": 5,
 "timestamp": "1647337490945"
}

其中endpoints就是我們發布出去的監聽器。這個屬性的格式為:


advertised.listeners = listener_name://host_name:port,listener_name2://host_nam2e:port2

默認情況下,advertised.listeners不設置會自動使用listeners屬性
advertised.listeners不支持0.0.0.0這種形式, 所以如果listeners屬性設置成0.0.0.0，則必須設置advertised.listeners屬性。具體請看 示例3 因為0.0.0.0是表示的是監聽Broker上任意的網卡的, 你將這個發布出去,那么別的Broker和客戶端怎么知道你具體的ip和端口呢？
可以同時配置多個, 并且用逗號隔開
可動態配置該屬性

listener.security.protocol.map

監聽器名稱和安全協議之間的映射關系集合。

listeners=PLAINTEXT://localhost:9092

看看上面的配置, PLANINTEXT是監聽器名稱,那么它對應的安全協議是什么呢？它對應的安全協議是 PLANINTEXT, 為什么呢？那是因為默認情況下,已經有了他們的映射關系。

默認集合：
PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

屬性格式:

監聽名稱1:安全協議1,監聽名稱2:安全協議2

現有的安全協議,有4種, 分別如下, 下為默認監聽器名稱映射對應的安全協議情況。

PLAINTEXT => PLAINTEXT 不需要授權,非加密通道
SSL => SSL 使用SSL加密通道
SASL_PLAINTEXT => SASL_PLAINTEXT 使用SASL認證非加密通道
SASL_SSL => SASL_SSL 使用SASL認證并且SSL加密通道

當然你也可以自己重新映射監聽器名稱和安全協議, 比如: 示例4

inter.broker.listener.name

用于Broker之間通信的listener的名稱。如果未設置，則listener名稱由 security.inter.broker.protocol定義(security.inter.broker.protocol默認值是PLAINTEXT)。

同時設置這個和 security.inter.broker.protocol 屬性是錯誤的。

默認值：空

不可動態配置。

特別注意：這個屬性表示是Broker之間的網絡通信使用的監聽器, 比如 Broker2Broker 但是還有一種就是Controller2Broker、如果沒有配置control.plane.listener.name ，那么走的也是inter.broker.listener.name 這個監聽器。

根據本地配置的監聽器名稱, 去查找其他Broker的監聽器的EndPoint。所以一般所有Broker的監聽器名稱都必須一致，否則的話就找不到具體的EndPoint,無法正確的發起請求。

security.inter.broker.protocol

用于在代理之間進行通信的安全協議。

有效值為：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。

同時設置該屬性和 inter.broker.listener.name 屬性是錯誤的。

默認值：PLAINTEXT (純文本)

注意這個跟inter.broker.listener.name是有區別的, 這個配置只有四個選項,是安全協議而inter.broker.listener.name是監聽名稱, 是需要通過這個監聽名稱去找到它映射的 安全協議 還有 IP:PORT

如果inter.broker.listener.name沒有配置,則默認使用security.inter.broker.protocol的配置. 對inter.broker.listener.name而言,最終還是要去找到對應的 IP:PORT。

一般自定義了監聽器名稱, inter.broker.listener.name就是必須要設置的, 不能使用security.inter.broker.protocol 來代替。

control.plane.listener.name

用于Controller和Broker之間通信的監聽器名稱, Broker將會使用control.plane.listener.name 來定位監聽器列表中的EndPoint

如果未設置,則默認使用inter.broker.listener.name來通信,沒有專門的鏈接。

詳情請看：Kafka的客戶端NetworkClient如何發起的請求

示例說明

1 . 綁定一個IP, 客戶端使用另外的IP訪問

讓broker 監聽localhost:9092. 然后客戶端訪問broker的具體IP.

listeners=PLAINTEXT://localhost:9092

啟動之后查看一下監聽情況

linux命令: netstat -anp |grep 9092 mac環境命令：netstat -AaLlnW

在這里插入圖片描述

當然,如果你這臺機器剛好還是Controller的話,除了了LISTEN, 還能看到ESTABLISHED狀態的連接,因為Controller也會給這臺Broker建立連接發起請求的，比如通知Broker更新元信息之類的。

在這里插入圖片描述

我們使用生產者客戶端來生產幾條消息

sh bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092  --topic Topic4
## 或者
sh bin/kafka-console-producer.sh --bootstrap-server localhost:9092  --topic Topic4

可以發現正常發送消息。

那么接下來,使用使用具體IP發起請求


sh bin/kafka-console-producer.sh --bootstrap-server 10.xxx.xx.128:9092  --topic Topic4

在這里插入圖片描述

[2022-03-16 12:59:07,024] WARN [Controller id=1000, targetBrokerId=1000] Connection to node 1000 

(/10.xxx.xxx.xx:9092) could not be established. Broker may not be available. (org.Apache.kafka.clients.NetworkClient)

可以看到,客戶端提示說不能跟這個ip:port建立連接。

2. listeners 和 advertised.listeners 配置的IP不一樣


listeners=PLAINTEXT://xx.xx.xxx.01:9092

advertised.listeners=PLAINTEXT:/xx.xx.xxx.02:9092

假設你本地監聽和發布的監聽不一樣, 那么就會造成其他broker和客戶端跟這臺broker不能正確的建立鏈接。

如果你這臺Broker剛好還是Controller，那么他也會對自己建立連接, 都是根據advertised.listeners的配置來建立的,同樣會失敗。其他broker也一樣。


[2022-03-16 12:59:07,024] WARN [Controller id=1000, targetBrokerId=1000] Connection to node 1000 

(/10.xxx.xxx.xx:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

3 . listeners監聽任意可用IP, advertised.listeners發布指定IP

在示例2中,我們指定 listeners 監聽器和advertised.listeners發布的監聽器不一致會導致異常。那么,我們只需要將監聽器的和發布的監聽器一致就行了

當然,我們還可直接設置監聽器監聽任意可用IP(該Broker上的可用IP)

listeners=PLAINTEXT://0.0.0.0:9092

當然,如果只是將host設置為 0.0.0.0. 那么會報錯

java.lang.IllegalArgumentException: requirement failed: advertised.listeners cannot use the nonroutable meta-address 0.0.0.0. Use a routable IP address.
 at kafka.server.KafkaConfig.validateValues(KafkaConfig.scala:1789)

因為默認情況下,advertised.listeners不設置的話,則默認使用listeners的屬性,然而advertised.listeners是不支持0.0.0.0的，所以需要指定暴露的監聽器,如下

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://xx.xx.xx.128:9092

這樣子配置就不會報錯了,其他Broker和客戶端會通過advertised.listeners發布的監聽器來跟該Broker建立鏈接。

注意: 這個時候你還可以在這臺Broker的機器上使用 localhost 來進行訪問。比如：


 sh bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092  --topic Topic4

在這里插入圖片描述

可以看到也是可以正常發送消息的。

4 . listeners配置多個監聽器,內外網分流


listeners = INSIDE://內網IP:9091,OUTSIDE://外網IP:9092

#把OUTSIDE 的安全協議映射成PLAINTEXT INSIDE也映射成PLAINTEXT
listener.security.protocol.map=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT

# Broker之間的連接用 INSIDE 監聽器
inter.broker.listener.name=INSIDE

設置了2個監聽器 ①. INSIDE 監聽內網IP ②. OUTSIDE 監聽外網IP

因為這個是我們自己定義的監聽名稱,listener.security.protocol.map默認映射中并沒有對應的映射關系所以我們就需要主動設置這個映射關系 listener.security.protocol.map=INSIDE:PLAINTEXT,OUTSIDE:SSL

不然會拋異常

Caused by: java.lang.IllegalArgumentException: No security protocol defined for listener INSIDE

 at kafka.cluster.EndPoint$.$anonfun$createEndPoint$2(EndPoint.scala:48)

注意:自定義了監聽器,則必須要配置inter.broker.listener.name

確定好內部的broker之間通信的監聽器. 并確保能夠正常訪問。

這樣Broker直接就會通過內網互相連接, 客戶端除了可以通過內網連接(如果在內網環境的話),也可以通過外網連接。

2幾種場景的配置方式

1. 一臺機器部署一套集群

這種場景一般是自己開發測試的時候, 比如自己搭建一個集群,學習學習,但是又沒有那么多機器,那么就可以在一臺電腦上部署多個Broker。

只配置listeners屬性


listeners = 監聽名稱://your.host.name:port

關于監聽名稱,默認的映射關系有4種。

PLAINTEXT => PLAINTEXT 不需要授權,非加密通道
SSL => SSL 使用SSL加密通道
SASL_PLAINTEXT => SASL_PLAINTEXT 使用SASL認證非加密通道
SASL_SSL => SASL_SSL 使用SASL認證并且SSL加密通道

簡單一點,用PLAINTEXT就夠了, 這里我們可以把host給去掉, 或者使用localhost


listeners = PLAINTEXT://:port
或者
listeners = PLAINTEXT://localhost:port

如果沒有配置host，會調用java.net.InetAddress.getCanonicalHostName()獲取本機host. 默認情況下就是localhost.

這里之所以建議你不填寫具體的host,是因為一般自己搭建玩玩的時候可能網絡IP會經常變動(例如家里的和公司), 如果綁定了具體的IP的話,每次重啟都要更換配置就很麻煩。

可以看看Broker啟動后注冊到zk中的配置如下

{
 "features": {},
 "listener_security_protocol_map": {
  "PLAINTEXT": "PLAINTEXT"
 },
 "endpoints": ["PLAINTEXT://localhost:9092"],
 "jmx_port": -1,
 "port": 9092,
 "host": "localhost",
 "version": 5,
 "timestamp": "1647337490945"
}

這個endpoints 就是broker注冊到zk的訪問地址, 如果其他Broker或者客戶端要跟這臺Broker發生網絡請求話, 就是拿的這里面的值。

所以,你想想看,如果是不同機器上,你配置的host是 localhost, 是不是就訪問不了？

當然,listeners屬性的host,我們也可以自己去hosts文件里面配置別的域名。配置域名指向的具體IP, 這樣的話那還能奏效。就是每個Broker和客戶端都要配置host,這就比較麻煩,所以還不如直接配置IP呢。

2. 內網環境多機器部署集群

這種是絕大部分的場景, 一般公司部署集群都是在公司內網環境下, Broker之間和Broker與客戶端之間都在同一個網絡環境。并且安全協議都是直接PLAINTEXT(明文)或者其他


listeners=PLAINTEXT://ip:port

3. 內網和外網分流


listeners=INTERNAL://內網ip:port1,EXTERNAL://外網ip:port2

#把OUTSIDE 的安全協議映射成PLAINTEXT INSIDE也映射成PLAINTEXT
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT

# Broker之間的連接用 INSIDE 監聽器
inter.broker.listener.name=INTERNAL

配置了兩個監聽器,每個Brokerinter.broker.listener.name=INTERNAL 使用內網交流。

其他的客戶端例如Producer和Consumer 請求的時候直接訪問外網IP.

3. 內網和外網和Controller分流


listeners=INTERNAL://內網ip:port1,EXTERNAL://外網ip:port2,CONTROLLER://內網ip:port3,

#把OUTSIDE 的安全協議映射成PLAINTEXT INSIDE也映射成PLAINTEXT
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT

# Broker之間的連接用 INSIDE 監聽器
inter.broker.listener.name=INTERNAL
control.plane.listener.name=CONTROLLER

這樣配置