在VPC環境中作為Output接入 - ApsaraMQ for Kafka

雲訊息佇列 Kafka 版可以作為Output接入Logstash。本文說明如何在VPC環境下通過Logstash向雲訊息佇列 Kafka 版發送訊息。

前提條件

購買並部署雲訊息佇列 Kafka 版執行個體。本文以非Serverless執行個體為例進行說明，具體操作請參見VPC接入。
下載並安裝Logstash。具體操作，請參見Download Logstash。
下載並安裝JDK 8。具體操作，請參見Download JDK 8。

步驟一：擷取存取點

Logstash通過雲訊息佇列 Kafka 版的存取點與雲訊息佇列 Kafka 版建立串連。

說明

雲訊息佇列 Kafka 版支援以下VPC環境存取點：

預設存取點：連接埠號碼為9092。
SASL存取點：連接埠號碼為9094。如需使用SASL存取點，請開啟ACL，具體步驟，請參見開啟ACL。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊作為Output接入Logstash的執行個體。
在实例详情頁面的接入点信息地區，擷取執行個體的存取點。在配置信息地區，擷取用户名與密码。
說明
不同存取點的差異，請參見存取點對比。

步驟二：建立Topic

建立用於儲存訊息的Topic。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
重要
Topic需要在應用程式所在的地區（即所部署的ECS的所在地區）進行建立。Topic不能跨地區使用。例如Topic建立在華北2（北京）這個地區，那麼訊息生產端和消費端也必須運行在華北2（北京）的ECS。
在实例列表頁面，單擊目標執行個體名稱。
在左側導覽列，單擊Topic 管理。
在Topic 管理頁面，單擊创建 Topic。

在创建 Topic面板，設定Topic屬性，然後單擊確定。

參數	說明	樣本
名称	Topic名稱。	demo
描述	Topic的簡單描述。	demo test
分区数	Topic的分區數量。	12
存储引擎說明當前僅非Serverless專業版執行個體支援選擇儲存引擎類型，其他執行個體暫不支援選擇，預設為雲端儲存類型。	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。

步驟三：Logstash發送訊息

在安裝了Logstash的機器上啟動Logstash，向建立的Topic發送訊息。

執行cd命令切換到logstash的bin目錄。

建立output.conf設定檔。

執行命令vim output.conf建立空的設定檔。
按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"
       }
}

參數

描述

樣本值

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

topic_id

Topic的名稱。

logstash_test

按Esc鍵回到命令列模式。
按：鍵進入底行模式，輸入wq，然後按斷行符號鍵儲存檔案並退出。

向建立的Topic發送訊息。
1. 執行./logstash -f output.conf。
2. 輸入test，然後按斷行符號鍵。
  返回結果如下。

步驟四：查看Topic分區

查看訊息發送到Topic的情況。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊作為Output接入Logstash的執行個體。
在左側導覽列，單擊Topic 管理。

在Topic 管理頁面，單擊目標Topic名称進入Topic 详情頁面，然後單擊分区状态頁簽。

表 1. 分區狀態資訊

參數	說明
分區ID	該Topic分區的ID號。
最小位點	該Topic在當前分區下的最小消費位點。
最大位點	該Topic在當前分區下的最大消費位點。
分區訊息量	該Topic在當前分區下的訊息總量。
最新動向時間	本分區中最近一條訊息的儲存時間。

分區狀態資訊

步驟五：按位點查詢訊息

您可以根據發送的訊息的分區ID和位點資訊查詢該訊息。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊目標執行個體名稱。
在左側導覽列，單擊消息查询。
在消息查询頁面的查询方式列表中，選擇按位点查询。

在Topic列表中，選擇訊息所屬Topic名稱；在分区列表中，選擇訊息所屬的分區；在起始位点文字框，輸入訊息所在分區的位點，然後單擊查询。

展示該查詢位點及以後連續的訊息。例如，指定的分區和位點都為“5”，那麼返回的結果從位點“5”開始。

表 2. 查詢結果參數解釋

參數	描述
分区	訊息的Topic分區。
位点	訊息的所在的位點。
Key	訊息的鍵（已強制轉化為String類型）。
Value	訊息的值（已強制轉化為String類型），即訊息的具體內容。
消息创建时间	發送訊息時，用戶端內建的或是您指定的`ProducerRecord`中的訊息建立時間。說明如果配置了該欄位，則按配置值顯示。如果未配置該欄位，則預設取訊息發送時的系統時間。如果顯示值為1970/x/x x:x:x，則說明發送時間配置為0或其他有誤的值。 0.9及以前版本的雲訊息佇列 Kafka 版用戶端不支援配置該時間。
操作	單擊下载 Key：下載訊息的索引值。單擊下载 Value：下載訊息的具體內容。重要查詢到的每條訊息在控制台上最多顯示1 KB的內容，超過1 KB的部分將自動截斷。如需查看完整的訊息內容，請下載相應的訊息。下載的訊息最大為10 MB。如果訊息超過10 MB，則只下載10 MB的內容。

ApsaraMQ for Kafka：在VPC環境中作為Output接入