消息队列Kafka版可以作为Output接入Logstash。本文说明如何在公网环境下通过Logstash向消息队列Kafka版发送消息。

前提条件

在开始本教程前,请确保您已完成以下操作:

步骤一:获取接入点

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

  1. 登录消息队列Kafka版控制台
  2. 在左侧导航栏,单击实例详情
  3. 实例详情页面,选择要作为Output接入Logstash的实例。
  4. 基本信息区域,获取实例的接入点。
    endpoint
    说明 不同接入点的差异,请参见接入点对比
  5. 安全配置区域,获取实例的用户名和密码。
    configure

步骤二:创建Topic

创建用于存储消息的Topic。

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

步骤三:Logstash发送消息

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

  1. 执行cd命令切换到logstash的bin目录。
  2. 下载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,然后按回车键。
      output_result

步骤四:查看Topic分区

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

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

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

    topic_status

步骤五:按位点查询消息

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

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

更多信息

更多参数设置,请参见Kafka output plugin