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

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

前提條件

在開始本教程前，請確保您已完成以下操作：

購買並部署雲訊息佇列 Kafka 版執行個體。具體操作，請參見公網和VPC接入。
下載並安裝Logstash。具體操作，請參見Download Logstash。
下載並安裝JDK 8。具體操作，請參見Download JDK 8。

步驟一：擷取存取點

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

登入雲訊息佇列 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目錄。

執行以下命令下載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設定檔。
1. 執行命令vim jaas.conf建立空的設定檔。
2. 按i鍵進入插入模式。
3. 輸入以下內容。
```
KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="XXX"
  password="XXX";
};
```
  參數
  描述
  樣本值
  username
  公網/VPC執行個體的使用者名稱。
  alikafka_pre-cn-v0h1***
  password
  公網/VPC執行個體的密碼。
  GQiSmqbQVe3b9hdKLDcIlkrBK6***
4. 按Esc鍵回到命令列模式。
5. 按：鍵進入底行模式，輸入wq，然後按斷行符號鍵儲存檔案並退出。

建立output.conf設定檔。

執行命令vim output.conf建立空的設定檔。
按i鍵進入插入模式。

輸入以下內容。

input {
    stdin{}
}

output {
   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"
        topic_id => "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 => ""
    }
}

參數	描述	樣本值
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
topic_id	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	SSL存取點辨識演算法。6.x及以上版本Logstash需要加上該參數。	空值

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

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

步驟四：查看Topic分區

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

登入雲訊息佇列 Kafka 版控制台。
在概览頁面的资源分布地區，選擇地區。
在实例列表頁面，單擊目標執行個體名稱。
在左側導覽列，單擊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：在公網環境中作為Output接入