作为Output接入

步骤一：获取接入点

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

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

   说明 不同接入点的差异，请参见接入点对比。

步骤二：创建Topic

创建用于存储消息的Topic。

1. 在消息队列Kafka版控制台的左侧导航栏，单击Topic管理。
2. 在Topic管理页面，单击创建Topic。
3. 在创建Topic页面，输入Topic信息，然后单击创建。
   

步骤三：Logstash发送消息

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

1. 执行以下命令切换到logstash的bin目录。

   cd /home/logstash/logstash-7.6.2/bin

2. 创建output.conf配置文件。
   1. 执行命令vim output.conf。
   2. 按i键。
   3. 输入以下内容。

      input {
          input {
            stdin{}
        }
      }
      
      output {
         kafka {
              bootstrap_servers => "192.168.XXX.XXX:9092,192.168.XXX.XXX:9092,192.168.XXX.XXX:9092"
              topic_id => "logstash_test"
             }
      }

      参数	描述	示例值
      bootstrap_servers	消息队列Kafka版提供以下VPC接入点： 默认接入点 SASL接入点	192.168.XXX.XXX:9092,192.168.XXX.XXX:9092,192.168.XXX.XXX:9092
      topic_id	Topic的名称。	logstash_test

   4. 按Esc键，输入:wq，然后回车。
3. 向创建的Topic发送消息。
   1. 执行./logstash -f output.conf。
   2. 输入test，然后按Enter键。

      返回结果如下。

      

步骤四：查看Topic分区

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

1. 在消息队列Kafka版控制台的左侧导航栏，单击Topic管理。
2. 在Topic管理页面，选择作为Output接入Logstash的实例，找到发送消息的Topic，在其右侧操作列单击分区状态。
3. 在分区状态页面，单击刷新。

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

   

步骤五：按位点查询消息

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

1. 在消息队列Kafka版控制台的左侧导航栏，单击消息查询。
2. 在消息查询页面，单击按位点查询页签。
3. 选择发送了消息的Topic，选择发送的消息的分区ID，选择发送的消息的位点，然后单击搜索。
   
4. 可选：在搜索结果右侧的操作列，单击消息详情。
   

更多信息

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