作为Output接入

步骤一：获取接入点

Logstash通过消息队列Kafka版的接入点与消息队列Kafka版建立连接。

1. 登录消息队列Kafka版控制台。
2. 在左侧导航栏，单击实例列表。
3. 在实例列表页面，单击要作为Output接入Logstash的实例的名称。
4. 在实例详情页面的基本信息区域，获取实例的接入点。
   

   说明 不同接入点的差异，请参见接入点对比。
5. 在安全配置区域，获取实例的用户名和密码。

步骤二：创建Topic

创建用于存储消息的Topic。

1. 登录消息队列Kafka版控制台。
2. 在左侧导航栏，单击实例列表。
3. 在实例列表页面，单击目标实例名称。
4. 在左侧导航栏，单击Topic管理。
5. 在Topic管理页面，单击创建Topic。
6. 在创建Topic对话框，输入Topic信息，然后单击创建。
   

步骤三：Logstash发送消息

在安装了Logstash的机器上启动Logstash，向创建的Topic发送消息。

1. 执行cd命令切换到logstash的bin目录。
2. 执行以下命令下载kafka.client.truststore.jks证书文件。

   wget https://code.aliyun.com/alikafka/aliware-kafka-demos/raw/master/kafka-log-stash-demo/vpc-ssl/kafka.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. 创建output.conf配置文件。
   1. 执行命令vim output.conf创建空的配置文件。
   2. 按i键进入插入模式。
   3. 输入以下内容。

      input {
          stdin{}
      }
      
      output {
         kafka {
              bootstrap_servers => "121.40.XXX.XXX:9093,120.26.XXX.XXX:9093,120.26.XXX.XXX: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接入点。	121.XX.XX.XX:9093,120.XX.XX.XX:9093,120.XX.XX.XX: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需要加上该参数。	空值

   4. 按Esc键回到命令行模式。
   5. 按：键进入底行模式，输入wq，然后按回车键保存文件并退出。
5. 向创建的Topic发送消息。
   1. 执行./logstash -f output.conf。
   2. 输入test，然后按回车键。
      

步骤四：查看Topic分区

查看消息发送到Topic的情况。

1. 登录消息队列Kafka版控制台。
2. 在左侧导航栏，单击实例列表。
3. 在实例列表页面，单击目标实例名称。
4. 在左侧导航栏，单击Topic管理。
5. 在Topic管理页面，找到发送消息的Topic，在其右侧操作列，单击分区状态。
6. 在分区状态对话框，单击刷新。

   发送的消息的分区ID和位点信息如下图所示。

   

步骤五：按位点查询消息

您可以根据发送的消息的分区ID和位点信息查询该消息。

1. 登录消息队列Kafka版控制台。
2. 在左侧导航栏，单击实例列表。
3. 在实例列表页面，单击目标实例名称。
4. 在左侧导航栏，单击消息查询。
5. 在消息查询页面，单击按位点查询页签。
6. 从请输入Topic列表，选择发送了消息的Topic，从请选择分区列表，选择发送的消息的分区ID，从请输入位点列表，选择发送的消息的位点，然后单击查询。
7. 可选：在查询结果右侧，可以查看消息详情。

更多信息

更多参数设置，请参见Kafka output plugin。