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

步驟一：擷取接入資訊

Logstash通過雲訊息佇列 Kafka 版的存取點與雲訊息佇列 Kafka 版建立串連，並通過使用者名稱及密碼進行校正。

1. 登入雲訊息佇列 Kafka 版控制台。

2. 在概览頁面的资源分布地區，選擇地區。

3. 在实例列表頁面，單擊作為Input接入Logstash的執行個體名稱。

4. 在实例详情頁面的接入点信息地區，擷取執行個體的存取點。在配置信息地區，擷取用户名與密码。

   

   說明

   不同存取點的差異，請參見存取點對比。

步驟二：建立Topic

建立用於儲存訊息的Topic。

1. 登入雲訊息佇列 Kafka 版控制台。

2. 在概览頁面的资源分布地區，選擇地區。

   重要

   Topic需要在應用程式所在的地區（即所部署的ECS的所在地區）進行建立。Topic不能跨地區使用。例如Topic建立在華北2（北京）這個地區，那麼訊息生產端和消費端也必須運行在華北2（北京）的ECS。

3. 在实例列表頁面，單擊目標執行個體名稱。

4. 在左側導覽列，單擊Topic 管理。

5. 在Topic 管理頁面，單擊创建 Topic。

6. 在创建 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發送訊息。

1. 登入雲訊息佇列 Kafka 版控制台。

2. 在概览頁面的资源分布地區，選擇地區。

3. 在实例列表頁面，單擊目標執行個體名稱。

4. 在左側導覽列，單擊Topic 管理。

5. 在Topic 管理頁面，單擊目標Topic名称進入Topic 详情頁面，然後單擊体验发送消息。

6. 在快速体验消息收发面板，發送測試訊息。

   • 发送方式選擇控制台。

     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。

1. 登入雲訊息佇列 Kafka 版控制台。

2. 在概览頁面的资源分布地區，選擇地區。

3. 在实例列表頁面，單擊目標執行個體名稱。

4. 在左側導覽列，單擊Group 管理。

5. 在Group 管理頁面，單擊创建 Group。

6. 在创建 Group面板的Group ID文字框輸入Group的名稱，在描述文字框簡要描述Group，並給Group添加標籤，單擊確定。

   建立完成後，在Group 管理頁面的列表中顯示已建立的Group。

步驟五：Logstash消費訊息

在安裝了Logstash的機器上啟動Logstash，從建立的Topic中消費訊息。

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

2. 執行以下命令下載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

3. 建立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，然後按斷行符號鍵儲存檔案並退出。

4. 建立input.conf設定檔。

   1. 執行命令vim input.conf建立空的設定檔。

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

   3. 輸入以下內容。

      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

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

   5. 按：鍵進入底行模式，輸入wq，然後按斷行符號鍵儲存檔案並退出。

5. 執行以下命令消費訊息。

   ./logstash -f input.conf

   返回結果如下。

   

更多資訊

更多參數設定，請參見Kafka input plugin。