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

前提条件

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

步骤一:获取接入点

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

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

步骤二:创建Topic

创建用于存储消息的Topic。

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

步骤三:Filebeat发送消息

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

  1. 执行以下命令切换到Filebeat的安装目录。
    cd /root/filebeat/filebeat-7.7.0-linux-x86_64
  2. 执行以下命令下载CA证书文件。
    wget https://github.com/AliwareMQ/aliware-kafka-demos/blob/master/kafka-filebeat-demo/vpc-ssl/ca-cert
  3. 创建output.conf配置文件。
    1. 执行命令vim output.conf
    2. i键。
    3. 输入以下内容。
      filebeat.inputs:
      - type: stdin
      
      output.kafka:
        hosts: ["121.XX.XX.XX:9093", "120.XX.XX.XX:9093", "120.XX.XX.XX:9093"]
        username: "alikafka_pre-cn-v641e1d***"
        password: "aeN3WLRoMPRXmAP2jvJuGk84Kuuo***"
      
        topic: 'filebeat_test'
        partition.round_robin:
          reachable_only: false
        ssl.certificate_authorities: ["/root/filebeat/filebeat-7.7.0-linux-x86_64/tasks/vpc_ssl/ca-cert"]
        ssl.verification_mode: none
      
        required_acks: 1
        compression: none
        max_message_bytes: 1000000
      参数 描述 示例值
      hosts 消息队列Kafka版提供的公网接入点为SSL接入点。 121.XX.XX.XX:9093, 120.XX.XX.XX:9093, 120.XX.XX.XX:9093
      username 公网/VPC实例的用户名。 alikafka_pre-cn-v641e1d***
      password 公网/VPC实例的密码。 aeN3WLRoMPRXmAP2jvJuGk84Kuuo***
      topic Topic的名称。 filebeat_test
      reachable_only 消息是否只发送到可用的分区。取值:
      • true:如果主分区不可用,输出可能阻塞。
      • false:即使主分区不可用,输出不被阻塞。
      false
      ssl.certificate_authorities CA证书所在位置。 /root/filebeat/filebeat-7.7.0-linux-x86_64/ca-cert
      ssl.verification_mode 认证模式。 none
      required_acks ACK可靠性。取值:
      • 0:无响应
      • 1:等待本地提交
      • -1:等待所有副本提交
      默认值为1。
      1
      compression 数据压缩编译码器。取值:
      • none
      • snappy
      • lz4
      • gzip
      默认值为gzip。
      none
      max_message_bytes 最大消息大小。单位为字节。默认值为1000000。该值应小于您配置的消息队列Kafka版最大消息大小。 1000000

      更多参数说明,请参见Kafka output plugin

    4. Esc键,输入:wq,然后按回车键。
  4. 向创建的Topic发送消息。
    1. 执行./filebeat -c ./output.yml
    2. 输入test,然后按回车键。

步骤四:查看Topic分区

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

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

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

    topic_status

步骤五:按位点查询消息

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

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