在VPC環(huán)境中作為Output接入
云消息隊(duì)列 Kafka 版可以作為Output接入Logstash。本文說(shuō)明如何在VPC環(huán)境下通過(guò)Logstash向云消息隊(duì)列 Kafka 版發(fā)送消息。
前提條件
購(gòu)買(mǎi)并部署云消息隊(duì)列 Kafka 版實(shí)例。具體操作,請(qǐng)參見(jiàn)VPC接入。
下載并安裝Logstash。具體操作,請(qǐng)參見(jiàn)Download Logstash。
下載并安裝JDK 8。具體操作,請(qǐng)參見(jiàn)Download JDK 8。
步驟一:獲取接入點(diǎn)
Logstash通過(guò)云消息隊(duì)列 Kafka 版的接入點(diǎn)與云消息隊(duì)列 Kafka 版建立連接。
云消息隊(duì)列 Kafka 版支持以下VPC環(huán)境接入點(diǎn):
默認(rèn)接入點(diǎn):端口號(hào)為9092。
SASL接入點(diǎn):端口號(hào)為9094。如需使用SASL接入點(diǎn),請(qǐng)開(kāi)啟ACL,具體步驟,請(qǐng)參見(jiàn)開(kāi)啟ACL。
在概覽頁(yè)面的資源分布區(qū)域,選擇地域。
在實(shí)例列表頁(yè)面,單擊作為Output接入Logstash的實(shí)例。
在實(shí)例詳情頁(yè)面的接入點(diǎn)信息區(qū)域,獲取實(shí)例的接入點(diǎn)。在配置信息區(qū)域,獲取用戶名與密碼。
說(shuō)明不同接入點(diǎn)的差異,請(qǐng)參見(jiàn)接入點(diǎn)對(duì)比。
步驟二:創(chuàng)建Topic
創(chuàng)建用于存儲(chǔ)消息的Topic。
在概覽頁(yè)面的資源分布區(qū)域,選擇地域。
重要Topic需要在應(yīng)用程序所在的地域(即所部署的ECS的所在地域)進(jìn)行創(chuàng)建。Topic不能跨地域使用。例如Topic創(chuàng)建在華北2(北京)這個(gè)地域,那么消息生產(chǎn)端和消費(fèi)端也必須運(yùn)行在華北2(北京)的ECS。
在實(shí)例列表頁(yè)面,單擊目標(biāo)實(shí)例名稱。
在左側(cè)導(dǎo)航欄,單擊Topic 管理。
在Topic 管理頁(yè)面,單擊創(chuàng)建 Topic。
在創(chuàng)建 Topic面板,設(shè)置Topic屬性,然后單擊確定。
參數(shù)
說(shuō)明
示例
名稱
Topic名稱。
demo
描述
Topic的簡(jiǎn)單描述。
demo test
分區(qū)數(shù)
Topic的分區(qū)數(shù)量。
12
存儲(chǔ)引擎
說(shuō)明當(dāng)前僅專業(yè)版實(shí)例支持選擇存儲(chǔ)引擎類型,標(biāo)準(zhǔn)版暫不支持,默認(rèn)選擇為云存儲(chǔ)類型。
Topic消息的存儲(chǔ)引擎。
云消息隊(duì)列 Kafka 版支持以下兩種存儲(chǔ)引擎。
云存儲(chǔ):底層接入阿里云云盤(pán),具有低時(shí)延、高性能、持久性、高可靠等特點(diǎn),采用分布式3副本機(jī)制。實(shí)例的規(guī)格類型為標(biāo)準(zhǔn)版(高寫(xiě)版)時(shí),存儲(chǔ)引擎只能為云存儲(chǔ)。
Local 存儲(chǔ):使用原生Kafka的ISR復(fù)制算法,采用分布式3副本機(jī)制。
云存儲(chǔ)
消息類型
Topic消息的類型。
普通消息:默認(rèn)情況下,保證相同Key的消息分布在同一個(gè)分區(qū)中,且分區(qū)內(nèi)消息按照發(fā)送順序存儲(chǔ)。集群中出現(xiàn)機(jī)器宕機(jī)時(shí),可能會(huì)造成消息亂序。當(dāng)存儲(chǔ)引擎選擇云存儲(chǔ)時(shí),默認(rèn)選擇普通消息。
分區(qū)順序消息:默認(rèn)情況下,保證相同Key的消息分布在同一個(gè)分區(qū)中,且分區(qū)內(nèi)消息按照發(fā)送順序存儲(chǔ)。集群中出現(xiàn)機(jī)器宕機(jī)時(shí),仍然保證分區(qū)內(nèi)按照發(fā)送順序存儲(chǔ)。但是會(huì)出現(xiàn)部分分區(qū)發(fā)送消息失敗,等到分區(qū)恢復(fù)后即可恢復(fù)正常。當(dāng)存儲(chǔ)引擎選擇Local 存儲(chǔ)時(shí),默認(rèn)選擇分區(qū)順序消息。
普通消息
日志清理策略
Topic日志的清理策略。
當(dāng)存儲(chǔ)引擎選擇Local 存儲(chǔ)(當(dāng)前僅專業(yè)版實(shí)例支持選擇存儲(chǔ)引擎類型為L(zhǎng)ocal存儲(chǔ),標(biāo)準(zhǔn)版暫不支持)時(shí),需要配置日志清理策略。
云消息隊(duì)列 Kafka 版支持以下兩種日志清理策略。
Delete:默認(rèn)的消息清理策略。在磁盤(pán)容量充足的情況下,保留在最長(zhǎng)保留時(shí)間范圍內(nèi)的消息;在磁盤(pán)容量不足時(shí)(一般磁盤(pán)使用率超過(guò)85%視為不足),將提前刪除舊消息,以保證服務(wù)可用性。
Compact:使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保證相同Key的消息,最新的value值一定會(huì)被保留。主要適用于系統(tǒng)宕機(jī)后恢復(fù)狀態(tài),系統(tǒng)重啟后重新加載緩存等場(chǎng)景。例如,在使用Kafka Connect或Confluent Schema Registry時(shí),需要使用Kafka Compact Topic存儲(chǔ)系統(tǒng)狀態(tài)信息或配置信息。
重要Compact Topic一般只用在某些生態(tài)組件中,例如Kafka Connect或Confluent Schema Registry,其他情況的消息收發(fā)請(qǐng)勿為T(mén)opic設(shè)置該屬性。具體信息,請(qǐng)參見(jiàn)云消息隊(duì)列 Kafka 版Demo庫(kù)。
Compact
標(biāo)簽
Topic的標(biāo)簽。
demo
創(chuàng)建完成后,在Topic 管理頁(yè)面的列表中顯示已創(chuàng)建的Topic。
步驟三:Logstash發(fā)送消息
在安裝了Logstash的機(jī)器上啟動(dòng)Logstash,向創(chuàng)建的Topic發(fā)送消息。
執(zhí)行cd命令切換到logstash的bin目錄。
創(chuàng)建output.conf配置文件。
執(zhí)行命令
vim output.conf
創(chuàng)建空的配置文件。按i鍵進(jìn)入插入模式。
輸入以下內(nèi)容。
input { input { stdin{} } } output { kafka { bootstrap_servers => "alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092" topic_id => "logstash_test" } }
參數(shù)
描述
示例值
bootstrap_servers
云消息隊(duì)列 Kafka 版提供以下VPC接入點(diǎn):
默認(rèn)接入點(diǎn)
SASL接入點(diǎn)
alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092
topic_id
Topic的名稱。
logstash_test
按Esc鍵回到命令行模式。
按:鍵進(jìn)入底行模式,輸入wq,然后按回車(chē)鍵保存文件并退出。
向創(chuàng)建的Topic發(fā)送消息。
執(zhí)行
./logstash -f output.conf
。輸入test,然后按回車(chē)鍵。
返回結(jié)果如下。
步驟四:查看Topic分區(qū)
查看消息發(fā)送到Topic的情況。
在概覽頁(yè)面的資源分布區(qū)域,選擇地域。
在實(shí)例列表頁(yè)面,單擊作為Output接入Logstash的實(shí)例。
在左側(cè)導(dǎo)航欄,單擊Topic 管理。
在Topic 管理頁(yè)面,單擊目標(biāo)Topic名稱進(jìn)入Topic 詳情頁(yè)面,然后單擊分區(qū)狀態(tài)頁(yè)簽。
表 1. 分區(qū)狀態(tài)信息
參數(shù)
說(shuō)明
分區(qū)ID
該Topic分區(qū)的ID號(hào)。
最小位點(diǎn)
該Topic在當(dāng)前分區(qū)下的最小消費(fèi)位點(diǎn)。
最大位點(diǎn)
該Topic在當(dāng)前分區(qū)下的最大消費(fèi)位點(diǎn)。
分區(qū)消息量
該Topic在當(dāng)前分區(qū)下的消息總量。
最近更新時(shí)間
本分區(qū)中最近一條消息的存儲(chǔ)時(shí)間。
步驟五:按位點(diǎn)查詢消息
您可以根據(jù)發(fā)送的消息的分區(qū)ID和位點(diǎn)信息查詢?cè)撓ⅰ?/p>
在概覽頁(yè)面的資源分布區(qū)域,選擇地域。
在實(shí)例列表頁(yè)面,單擊目標(biāo)實(shí)例名稱。
在左側(cè)導(dǎo)航欄,單擊消息查詢。
在消息查詢頁(yè)面的查詢方式列表中,選擇按位點(diǎn)查詢。
在Topic列表中,選擇消息所屬Topic名稱;在分區(qū)列表中,選擇消息所屬的分區(qū);在起始位點(diǎn)文本框,輸入消息所在分區(qū)的位點(diǎn),然后單擊查詢。
展示該查詢位點(diǎn)及以后連續(xù)的消息。例如,指定的分區(qū)和位點(diǎn)都為“5”,那么返回的結(jié)果從位點(diǎn)“5”開(kāi)始。
表 2. 查詢結(jié)果參數(shù)解釋
參數(shù)
描述
分區(qū)
消息的Topic分區(qū)。
位點(diǎn)
消息的所在的位點(diǎn)。
Key
消息的鍵(已強(qiáng)制轉(zhuǎn)化為String類型)。
Value
消息的值(已強(qiáng)制轉(zhuǎn)化為String類型),即消息的具體內(nèi)容。
消息創(chuàng)建時(shí)間
發(fā)送消息時(shí),客戶端自帶的或是您指定的
ProducerRecord
中的消息創(chuàng)建時(shí)間。說(shuō)明如果配置了該字段,則按配置值顯示。
如果未配置該字段,則默認(rèn)取消息發(fā)送時(shí)的系統(tǒng)時(shí)間。
如果顯示值為1970/x/x x:x:x,則說(shuō)明發(fā)送時(shí)間配置為0或其他有誤的值。
0.9及以前版本的云消息隊(duì)列 Kafka 版客戶端不支持配置該時(shí)間。
操作
單擊下載 Key:下載消息的鍵值。
單擊下載 Value:下載消息的具體內(nèi)容。
重要查詢到的每條消息在控制臺(tái)上最多顯示1 KB的內(nèi)容,超過(guò)1 KB的部分將自動(dòng)截?cái)唷H缧璨榭赐暾南?nèi)容,請(qǐng)下載相應(yīng)的消息。
下載的消息最大為10 MB。如果消息超過(guò)10 MB,則只下載10 MB的內(nèi)容。
更多信息
更多參數(shù)設(shè)置,請(qǐng)參見(jiàn)Kafka output plugin。