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

雲訊息佇列 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。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊作為Input接入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。

步驟三：發送訊息

向建立的Topic發送訊息。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊目標執行個體名稱。
在左側導覽列，單擊Topic 管理。
在Topic 管理頁面，單擊目標Topic名称進入Topic 详情頁面，然後單擊体验发送消息。
在快速体验消息收发面板，發送測試訊息。
- 发送方式選擇控制台。
  1. 在消息 Key文字框中輸入訊息的Key值，例如demo。
  2. 在消息内容文字框輸入測試的訊息內容，例如 {"key": "test"}。
  3. 設定发送到指定分区，選擇是否指定分區。
    - 單擊是，在分区 ID文字框中輸入分區的ID，例如0。如果您需查詢分區的ID，請參見查看分區狀態。
    - 單擊否，不指定分區。
  4. 根據介面提示資訊，通過SDK訂閱訊息，或者執行Docker命令訂閱訊息。
- 发送方式選擇Docker，運行Docker容器。
  1. 執行运行 Docker 容器生产示例消息地區的Docker命令，發送訊息。
  2. 執行发送后如何消费消息？地區的Docker命令，訂閱訊息。
- 发送方式選擇SDK，根據您的業務需求，選擇需要的語言或者架構的SDK以及接入方式，通過SDK體驗訊息收發。

步驟四：建立Group

建立Logstash所屬的Group。

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊目標執行個體名稱。
在左側導覽列，單擊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
```
返回結果如下。

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