云消息隊列 Kafka 版可以作為Input接入Logstash。本文說明如何在公網環境下通過Logstash從云消息隊列 Kafka 版消費消息。
前提條件
在開始本教程前,請確保您已完成以下操作:
購買并部署云消息隊列 Kafka 版實例。具體操作,請參見公網和VPC接入。
下載并安裝Logstash。具體操作,請參見Download Logstash。
下載并安裝JDK 8。具體操作,請參見Download JDK 8。
步驟一:獲取接入信息
Logstash通過云消息隊列 Kafka 版的接入點與云消息隊列 Kafka 版建立連接,并通過用戶名及密碼進行校驗。
在概覽頁面的資源分布區域,選擇地域。
在實例列表頁面,單擊作為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目錄。
執行以下命令下載kafka.client.truststore.jks證書文件。
wget -O kafka.client.truststore.jks https://github.com/AliwareMQ/aliware-kafka-demos/blob/master/kafka-log-stash-demo/vpc-ssl/mix.4096.client.truststore.jks
創建jaas.conf配置文件。
執行命令
vim jaas.conf
創建空的配置文件。按i鍵進入插入模式。
輸入以下內容。
KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="XXX" password="XXX"; };
參數
描述
示例值
username
公網/VPC實例的用戶名。
alikafka_pre-cn-v0h1***
password
公網/VPC實例的密碼。
GQiSmqbQVe3b9hdKLDcIlkrBK6***
按Esc鍵回到命令行模式。
按:鍵進入底行模式,輸入wq,然后按回車鍵保存文件并退出。
創建input.conf配置文件。
執行命令
vim input.conf
創建空的配置文件。按i鍵進入插入模式。
輸入以下內容。
input { kafka { bootstrap_servers => "alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093" topics => ["logstash_test"] security_protocol => "SASL_SSL" sasl_mechanism => "PLAIN" jaas_path => "/home/logstash-7.6.2/bin/jaas.conf" ssl_truststore_password => "KafkaOnsClient" ssl_truststore_location => "/home/logstash-7.6.2/bin/kafka.client.truststore.jks" ssl_endpoint_identification_algorithm => "" group_id => "logstash_group" consumer_threads => 3 auto_offset_reset => "earliest" } } output { stdout { codec => rubydebug } }
參數
描述
示例值
bootstrap_servers
云消息隊列 Kafka 版提供的公網接入點為SSL接入點。
alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093
topics
Topic的名稱。
logstash_test
security_protocol
安全協議。默認為SASL_SSL,無需修改。
SASL_SSL
sasl_mechanism
安全認證機制。默認為PLAIN,無需修改。
PLAIN
jaas_path
jaas.conf配置文件位置。
/home/logstash-7.6.2/bin/jaas.conf
ssl_truststore_password
kafka.client.truststore.jks證書密碼。默認值為KafkaOnsClient,無需修改。
KafkaOnsClient
ssl_truststore_location
kafka.client.truststore.jks證書位置。
/home/logstash-7.6.2/bin/kafka.client.truststore.jks
ssl_endpoint_identification_algorithm
6.x及以上版本Logstash需要加上該參數。
空值
group_id
Consumer Group的名稱。
logstash_group
consumer_threads
消費線程數。建議與Topic的分區數保持一致。
3
auto_offset_reset
重置偏移量。取值:
earliest:讀取最早的消息。
latest:讀取最新的消息。
earliest
按Esc鍵回到命令行模式。
按:鍵進入底行模式,輸入wq,然后按回車鍵保存文件并退出。
執行以下命令消費消息。
./logstash -f input.conf
返回結果如下。
更多信息
更多參數設置,請參見Kafka input plugin。