步驟一：獲取接入點

Logstash通過云消息隊列 Kafka 版的接入點與云消息隊列 Kafka 版建立連接。

1. 登錄云消息隊列 Kafka 版控制臺。

2. 在概覽頁面的資源分布區(qū)域，選擇地域。

3. 在實例列表頁面，單擊作為Input接入Logstash的實例名稱。

4. 在實例詳情頁面的接入點信息區(qū)域，獲取實例的接入點。在配置信息區(qū)域，獲取用戶名與密碼。

   

   說明

   不同接入點的差異，請參見接入點對比。

步驟二：創(chuàng)建Topic

創(chuàng)建用于存儲消息的Topic。

1. 登錄云消息隊列 Kafka 版控制臺。

2. 在概覽頁面的資源分布區(qū)域，選擇地域。

   重要

   Topic需要在應用程序所在的地域（即所部署的ECS的所在地域）進行創(chuàng)建。Topic不能跨地域使用。例如Topic創(chuàng)建在華北2（北京）這個地域，那么消息生產(chǎn)端和消費端也必須運行在華北2（北京）的ECS。

3. 在實例列表頁面，單擊目標實例名稱。

4. 在左側導航欄，單擊Topic 管理。

5. 在Topic 管理頁面，單擊創(chuàng)建 Topic。

6. 在創(chuàng)建 Topic面板，設置Topic屬性，然后單擊確定。

   參數(shù)	說明	示例
   名稱	Topic名稱。	demo
   描述	Topic的簡單描述。	demo test
   分區(qū)數(shù)	Topic的分區(qū)數(shù)量。	12
   存儲引擎 說明 當前僅專業(yè)版實例支持選擇存儲引擎類型，標準版暫不支持，默認選擇為云存儲類型。	Topic消息的存儲引擎。 云消息隊列 Kafka 版支持以下兩種存儲引擎。 云存儲：底層接入阿里云云盤，具有低時延、高性能、持久性、高可靠等特點，采用分布式3副本機制。實例的規(guī)格類型為標準版（高寫版）時，存儲引擎只能為云存儲。 Local 存儲：使用原生Kafka的ISR復制算法，采用分布式3副本機制。	云存儲
   消息類型	Topic消息的類型。 普通消息：默認情況下，保證相同Key的消息分布在同一個分區(qū)中，且分區(qū)內(nèi)消息按照發(fā)送順序存儲。集群中出現(xiàn)機器宕機時，可能會造成消息亂序。當存儲引擎選擇云存儲時，默認選擇普通消息。 分區(qū)順序消息：默認情況下，保證相同Key的消息分布在同一個分區(qū)中，且分區(qū)內(nèi)消息按照發(fā)送順序存儲。集群中出現(xiàn)機器宕機時，仍然保證分區(qū)內(nèi)按照發(fā)送順序存儲。但是會出現(xiàn)部分分區(qū)發(fā)送消息失敗，等到分區(qū)恢復后即可恢復正常。當存儲引擎選擇Local 存儲時，默認選擇分區(qū)順序消息。	普通消息
   日志清理策略	Topic日志的清理策略。 當存儲引擎選擇Local 存儲（當前僅專業(yè)版實例支持選擇存儲引擎類型為Local存儲，標準版暫不支持）時，需要配置日志清理策略。 云消息隊列 Kafka 版支持以下兩種日志清理策略。 Delete：默認的消息清理策略。在磁盤容量充足的情況下，保留在最長保留時間范圍內(nèi)的消息；在磁盤容量不足時（一般磁盤使用率超過85%視為不足），將提前刪除舊消息，以保證服務可用性。 Compact：使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保證相同Key的消息，最新的value值一定會被保留。主要適用于系統(tǒng)宕機后恢復狀態(tài)，系統(tǒng)重啟后重新加載緩存等場景。例如，在使用Kafka Connect或Confluent Schema Registry時，需要使用Kafka Compact Topic存儲系統(tǒng)狀態(tài)信息或配置信息。 重要 Compact Topic一般只用在某些生態(tài)組件中，例如Kafka Connect或Confluent Schema Registry，其他情況的消息收發(fā)請勿為Topic設置該屬性。具體信息，請參見云消息隊列 Kafka 版Demo庫。	Compact
   標簽	Topic的標簽。	demo

   創(chuàng)建完成后，在Topic 管理頁面的列表中顯示已創(chuàng)建的Topic。

步驟三：發(fā)送消息

向創(chuàng)建的Topic發(fā)送消息。

1. 登錄云消息隊列 Kafka 版控制臺。

2. 在概覽頁面的資源分布區(qū)域，選擇地域。

3. 在實例列表頁面，單擊目標實例名稱。

4. 在左側導航欄，單擊Topic 管理。

5. 在Topic 管理頁面，單擊目標Topic名稱進入Topic 詳情頁面，然后單擊體驗發(fā)送消息。

6. 在快速體驗消息收發(fā)面板，發(fā)送測試消息。

   • 發(fā)送方式選擇控制臺。

     1. 在消息 Key文本框中輸入消息的Key值，例如demo。

     2. 在消息內(nèi)容文本框輸入測試的消息內(nèi)容，例如 {"key": "test"}。

     3. 設置發(fā)送到指定分區(qū)，選擇是否指定分區(qū)。

        • 單擊是，在分區(qū) ID文本框中輸入分區(qū)的ID，例如0。如果您需查詢分區(qū)的ID，請參見查看分區(qū)狀態(tài)。

        • 單擊否，不指定分區(qū)。

     4. 根據(jù)界面提示信息，通過SDK訂閱消息，或者執(zhí)行Docker命令訂閱消息。

   • 發(fā)送方式選擇Docker，運行Docker容器。

     1. 執(zhí)行運行 Docker 容器生產(chǎn)示例消息區(qū)域的Docker命令，發(fā)送消息。

     2. 執(zhí)行發(fā)送后如何消費消息？區(qū)域的Docker命令，訂閱消息。

   • 發(fā)送方式選擇SDK，根據(jù)您的業(yè)務需求，選擇需要的語言或者框架的SDK以及接入方式，通過SDK體驗消息收發(fā)。

步驟四：創(chuàng)建Group

創(chuàng)建Logstash所屬的Group。

1. 登錄云消息隊列 Kafka 版控制臺。

2. 在概覽頁面的資源分布區(qū)域，選擇地域。

3. 在實例列表頁面，單擊目標實例名稱。

4. 在左側導航欄，單擊Group 管理。

5. 在Group 管理頁面，單擊創(chuàng)建 Group。

6. 在創(chuàng)建 Group面板的Group ID文本框輸入Group的名稱，在描述文本框簡要描述Group，并給Group添加標簽，單擊確定。

   創(chuàng)建完成后，在Group 管理頁面的列表中顯示已創(chuàng)建的Group。

步驟五：Logstash消費消息

在安裝了Logstash的機器上啟動Logstash，從創(chuàng)建的Topic中消費消息。

1. 執(zhí)行cd命令切換到logstash的bin目錄。

2. 創(chuàng)建input.conf配置文件。

   1. 執(zhí)行命令vim input.conf創(chuàng)建空的配置文件。

   2. 按i鍵進入插入模式。

   3. 輸入以下內(nè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}
      }

      參數(shù)	描述	示例值
      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	消費線程數(shù)。建議與Topic的分區(qū)數(shù)保持一致。	12
      auto_offset_reset	重置偏移量。取值： earliest：讀取最早的消息。 latest：讀取最新的消息。	earliest

   4. 按Esc鍵回到命令行模式。

   5. 按：鍵進入底行模式，輸入wq，然后按回車鍵保存文件并退出。

3. 執(zhí)行以下命令消費消息。

   ./logstash -f input.conf

   返回結果如下。

   

更多信息

更多參數(shù)設置，請參見Kafka input plugin。