云消息隊列 Kafka 版可以作為Input接入Logstash。本文說明如何在VPC環境下通過Logstash從云消息隊列 Kafka 版消費消息。
前提條件
在開始本教程前,請確保您已完成以下操作:
購買并部署云消息隊列 Kafka 版實例。具體操作,請參見VPC接入。
下載并安裝Logstash。具體操作,請參見Download Logstash。
下載并安裝JDK 8。具體操作,請參見Download JDK 8。
步驟一:獲取接入點
Logstash通過云消息隊列 Kafka 版的接入點與云消息隊列 Kafka 版建立連接。
云消息隊列 Kafka 版支持以下VPC環境接入點:
默認接入點:端口號為9092。
SASL接入點:端口號為9094。如需使用SASL接入點,請開啟ACL,具體步驟,請參見開啟ACL。
在概覽頁面的資源分布區域,選擇地域。
在實例列表頁面,單擊作為Input接入Logstash的實例名稱。
在實例詳情頁面的接入點信息區域,獲取實例的接入點。在配置信息區域,獲取用戶名與密碼。
說明不同接入點的差異,請參見接入點對比。
步驟二:創建Topic
創建用于存儲消息的Topic。
在概覽頁面的資源分布區域,選擇地域。
重要Topic需要在應用程序所在的地域(即所部署的ECS的所在地域)進行創建。Topic不能跨地域使用。例如Topic創建在華北2(北京)這個地域,那么消息生產端和消費端也必須運行在華北2(北京)的ECS。
在實例列表頁面,單擊目標實例名稱。
在左側導航欄,單擊Topic 管理。
在Topic 管理頁面,單擊創建 Topic。
在創建 Topic面板,設置Topic屬性,然后單擊確定。
參數
說明
示例
名稱
Topic名稱。
demo
描述
Topic的簡單描述。
demo test
分區數
Topic的分區數量。
12
存儲引擎
說明當前僅專業版實例支持選擇存儲引擎類型,標準版暫不支持,默認選擇為云存儲類型。
Topic消息的存儲引擎。
云消息隊列 Kafka 版支持以下兩種存儲引擎。
云存儲:底層接入阿里云云盤,具有低時延、高性能、持久性、高可靠等特點,采用分布式3副本機制。實例的規格類型為標準版(高寫版)時,存儲引擎只能為云存儲。
Local 存儲:使用原生Kafka的ISR復制算法,采用分布式3副本機制。
云存儲
消息類型
Topic消息的類型。
普通消息:默認情況下,保證相同Key的消息分布在同一個分區中,且分區內消息按照發送順序存儲。集群中出現機器宕機時,可能會造成消息亂序。當存儲引擎選擇云存儲時,默認選擇普通消息。
分區順序消息:默認情況下,保證相同Key的消息分布在同一個分區中,且分區內消息按照發送順序存儲。集群中出現機器宕機時,仍然保證分區內按照發送順序存儲。但是會出現部分分區發送消息失敗,等到分區恢復后即可恢復正常。當存儲引擎選擇Local 存儲時,默認選擇分區順序消息。
普通消息
日志清理策略
Topic日志的清理策略。
當存儲引擎選擇Local 存儲(當前僅專業版實例支持選擇存儲引擎類型為Local存儲,標準版暫不支持)時,需要配置日志清理策略。
云消息隊列 Kafka 版支持以下兩種日志清理策略。
Delete:默認的消息清理策略。在磁盤容量充足的情況下,保留在最長保留時間范圍內的消息;在磁盤容量不足時(一般磁盤使用率超過85%視為不足),將提前刪除舊消息,以保證服務可用性。
Compact:使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保證相同Key的消息,最新的value值一定會被保留。主要適用于系統宕機后恢復狀態,系統重啟后重新加載緩存等場景。例如,在使用Kafka Connect或Confluent Schema Registry時,需要使用Kafka Compact Topic存儲系統狀態信息或配置信息。
重要Compact Topic一般只用在某些生態組件中,例如Kafka Connect或Confluent Schema Registry,其他情況的消息收發請勿為Topic設置該屬性。具體信息,請參見云消息隊列 Kafka 版Demo庫。
Compact
標簽
Topic的標簽。
demo
創建完成后,在Topic 管理頁面的列表中顯示已創建的Topic。
步驟三:發送消息
向創建的Topic發送消息。
在概覽頁面的資源分布區域,選擇地域。
在實例列表頁面,單擊目標實例名稱。
在左側導航欄,單擊Topic 管理。
在Topic 管理頁面,找到目標Topic,在其操作列中,選擇 。
在快速體驗消息收發面板,發送測試消息。
發送方式選擇控制臺。
在消息 Key文本框中輸入消息的Key值,例如demo。
在消息內容文本框輸入測試的消息內容,例如 {"key": "test"}。
設置發送到指定分區,選擇是否指定分區。
單擊是,在分區 ID文本框中輸入分區的ID,例如0。如果您需查詢分區的ID,請參見查看分區狀態。
單擊否,不指定分區。
根據界面提示信息,通過SDK訂閱消息,或者執行Docker命令訂閱消息。
發送方式選擇Docker,運行Docker容器。
執行運行 Docker 容器生產示例消息區域的Docker命令,發送消息。
執行發送后如何消費消息?區域的Docker命令,訂閱消息。
發送方式選擇SDK,根據您的業務需求,選擇需要的語言或者框架的SDK以及接入方式,通過SDK體驗消息收發。
步驟四:創建Group
創建Logstash所屬的Group。
在概覽頁面的資源分布區域,選擇地域。
在實例列表頁面,單擊目標實例名稱。
在左側導航欄,單擊Group 管理。
在Group 管理頁面,單擊創建 Group。
在創建 Group面板的Group ID文本框輸入Group的名稱,在描述文本框簡要描述Group,并給Group添加標簽,單擊確定。
創建完成后,在Group 管理頁面的列表中顯示已創建的Group。
步驟五:Logstash消費消息
在安裝了Logstash的機器上啟動Logstash,從創建的Topic中消費消息。
執行cd命令切換到logstash的bin目錄。
創建input.conf配置文件。
執行命令
vim input.conf
創建空的配置文件。按i鍵進入插入模式。
輸入以下內容。
input { 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" group_id => "logstash_group" topics => ["logstash_test"] consumer_threads => 12 auto_offset_reset => "earliest" } } output { stdout{codec=>rubydebug} }
參數
描述
示例值
bootstrap_servers
云消息隊列 Kafka 版提供以下VPC接入點:
默認接入點
SASL接入點
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
group_id
Consumer Group的名稱。
logstash_group
topics
Topic的名稱。
logstash_test
consumer_threads
消費線程數。建議與Topic的分區數保持一致。
12
auto_offset_reset
重置偏移量。取值:
earliest:讀取最早的消息。
latest:讀取最新的消息。
earliest
按Esc鍵回到命令行模式。
按:鍵進入底行模式,輸入wq,然后按回車鍵保存文件并退出。
執行以下命令消費消息。
./logstash -f input.conf
返回結果如下。
更多信息
更多參數設置,請參見Kafka input plugin。