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

背景信息

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

步骤一：获取接入点

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

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

步骤二：创建Topic

创建用于存储消息的Topic。

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

步骤三：发送消息

向创建的Topic发送消息。

  1. 在消息队列Kafka版控制台的Topic管理页面，找到创建的Topic，在其右侧操作列，单击发送消息。
  2. 在发送消息对话框，输入消息信息，然后单击发送。
    send_msg

步骤四：创建Consumer Group

创建Filebeat所属的Consumer Group。

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

步骤五：Filebeat消费消息

在安装了Filebeat的机器上启动Filebeat，从创建的Topic中消费消息。

  1. 执行cd命令切换到Filebeat的安装目录。
  2. 获取CA证书文件。
  3. 创建input.yml配置文件。
    1. 执行命令vim input.yml创建空的配置文件。
    2. 按i键进入插入模式。
    3. 输入以下内容。
      filebeat.inputs:
- type: kafka
  hosts:
    - 121.XX.XX.XX:9093
    - 120.XX.XX.XX:9093
    - 120.XX.XX.XX:9093
  username: "alikafka_pre-cn-v641e1dt***"
  password: "aeN3WLRoMPRXmAP2jvJuGk84Kuuo***"
  topics: ["filebeat_test"]
  group_id: "filebeat_group"
  ssl.certificate_authorities: ["/root/filebeat/filebeat-7.7.0-linux-x86_64/ca-cert"]
  ssl.verification_mode: none

output.console:
  pretty: true
      参数 描述 示例值
      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***
      topics Topic的名称。 filebeat_test
      group_id Consumer Group的名称。 filebeat_group
      ssl.certificate_authorities CA证书所在位置。 /root/filebeat/filebeat-7.7.0-linux-x86_64/ca-cert
      ssl.verification_mode 认证模式。 none

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

    4. 按Esc键回到命令行模式。
    5. 按：键进入底行模式，输入wq，然后按回车键保存文件并退出。
  4. 执行以下命令消费消息。
    ./filebeat -c ./input.yml
    result