如何在公网环境通过Filebeat从消息队列Kafka版消费消息

背景信息

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

购买并部署消息队列Kafka版实例。具体操作，请参见公网和VPC接入。
下载并安装Filebeat。具体操作，请参见Download Filebeat。
下载并安装JDK 8。具体操作，请参见Download JDK 8。

步骤一：获取接入点

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

登录消息队列Kafka版控制台。
在概览页面的资源分布区域，选择地域。
在实例列表页面，单击作为Input接入Filebeat的实例名称。
在实例详情页面的接入点信息区域，获取实例的接入点。在配置信息区域，获取用户名与密码。

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

步骤二：创建Topic

创建用于存储消息的Topic。

登录消息队列Kafka版控制台。
在概览页面的资源分布区域，选择地域。

重要 Topic需要在应用程序所在的地域（即所部署的ECS的所在地域）进行创建。Topic不能跨地域使用。例如Topic创建在华北2（北京）这个地域，那么消息生产端和消费端也必须运行在华北2（北京）的ECS。
在实例列表页面，单击目标实例名称。
在左侧导航栏，单击Topic 管理。
在Topic 管理页面，单击创建 Topic。

在创建 Topic面板，设置Topic属性，然后单击确定。


参数	说明	示例
名称	Topic名称。	demo
描述	Topic的简单描述。	demo test
分区数	Topic的分区数量。	12
存储引擎说明当前仅专业版实例支持选择存储引擎类型，标准版暂不支持，默认选择为云存储类型。	Topic消息的存储引擎。消息队列Kafka版支持以下两种存储引擎。云存储：底层接入阿里云云盘，具有低时延、高性能、持久性、高可靠等特点，采用分布式3副本机制。实例的规格类型为标准版（高写版）时，存储引擎只能为云存储。 Local 存储：使用原生Kafka的ISR复制算法，采用分布式3副本机制。	云存储
消息类型	Topic消息的类型。普通消息：默认情况下，保证相同Key的消息分布在同一个分区中，且分区内消息按照发送顺序存储。集群中出现机器宕机时，可能会造成消息乱序。当存储引擎选择云存储时，默认选择普通消息。分区顺序消息：默认情况下，保证相同Key的消息分布在同一个分区中，且分区内消息按照发送顺序存储。集群中出现机器宕机时，仍然保证分区内按照发送顺序存储。但是会出现部分分区发送消息失败，等到分区恢复后即可恢复正常。当存储引擎选择Local 存储时，默认选择分区顺序消息。	普通消息
日志清理策略	Topic日志的清理策略。当存储引擎选择Local 存储（当前仅专业版实例支持选择存储引擎类型为Local存储，标准版暂不支持）时，需要配置日志清理策略。消息队列Kafka版支持以下两种日志清理策略。 Delete：默认的消息清理策略。在磁盘容量充足的情况下，保留在最长保留时间范围内的消息；在磁盘容量不足时（一般磁盘使用率超过85%视为不足），将提前删除旧消息，以保证服务可用性。 Compact：使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保证相同Key的消息，最新的value值一定会被保留。主要适用于系统宕机后恢复状态，系统重启后重新加载缓存等场景。例如，在使用Kafka Connect或Confluent Schema Registry时，需要使用Kafka Compact Topic存储系统状态信息或配置信息。重要 Compact Topic一般只用在某些生态组件中，例如Kafka Connect或Confluent Schema Registry，其他情况的消息收发请勿为Topic设置该属性。具体信息，请参见消息队列Kafka版Demo库。	Compact
标签	Topic的标签。	demo

创建完成后，在Topic 管理页面的列表中显示已创建的Topic。

步骤三：发送消息

向创建的Topic发送消息。

登录消息队列Kafka版控制台。
在概览页面的资源分布区域，选择地域。
在实例列表页面，单击目标实例名称。
在左侧导航栏，单击Topic 管理。
在Topic 管理页面，找到目标Topic，在其操作列中，选择更多 > 体验发送消息。
在快速体验消息收发面板，发送测试消息。
- 发送方式选择控制台。
  1. 在消息 Key文本框中输入消息的Key值，例如demo。
  2. 在消息内容文本框输入测试的消息内容，例如 {"key": "test"}。
  3. 设置发送到指定分区，选择是否指定分区。
    - 单击是，在分区 ID文本框中输入分区的ID，例如0。如果您需查询分区的ID，请参见查看分区状态。
    - 单击否，不指定分区。
  4. 根据界面提示信息，通过SDK订阅消息，或者执行Docker命令订阅消息。
- 发送方式选择Docker，运行Docker容器。
  1. 执行运行 Docker 容器生产示例消息区域的Docker命令，发送消息。
  2. 执行发送后如何消费消息？区域的Docker命令，订阅消息。
- 发送方式选择SDK，根据您的业务需求，选择需要的语言或者框架的SDK以及接入方式，通过SDK体验消息收发。

步骤四：创建Group

创建Filebeat所属的Group。

登录消息队列Kafka版控制台。
在概览页面的资源分布区域，选择地域。
在实例列表页面，单击目标实例名称。
在左侧导航栏，单击Group 管理。
在Group 管理页面，单击创建 Group。
在创建 Group面板的Group ID文本框输入Group的名称，在描述文本框简要描述Group，并给Group添加标签，单击确定。
创建完成后，在Group 管理页面的列表中显示已创建的Group。

步骤五：Filebeat消费消息

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

执行cd命令切换到Filebeat的安装目录。

执行以下命令下载CA证书文件。

wget https://code.aliyun.com/alikafka/aliware-kafka-demos/raw/master/kafka-filebeat-demo/vpc-ssl/ca-cert

创建input.yml配置文件。

执行命令vim input.yml创建空的配置文件。
按i键进入插入模式。

输入以下内容。

filebeat.inputs:
- type: kafka
  hosts:
    - alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093
    - alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093
    - alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com: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接入点。	- alikafka-pre-cn-zv********-1.alikafka.aliyuncs.com:9093 - alikafka-pre-cn-zv******-2.alikafka.aliyuncs.com:9093 - alikafka-pre-cn-zv********-3.alikafka.aliyuncs.com:9093
username	公网/VPC实例的用户名。	alikafka_pre-cn-v641e1d***
password	公网/VPC实例的密码。	aeN3WLRoMPRXmAP2jvJuGk84Kuuo***
topics	Topic的名称。	filebeat_test
group_id	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。

按Esc键回到命令行模式。
按：键进入底行模式，输入wq，然后按回车键保存文件并退出。

执行以下命令消费消息。
```
./filebeat -c ./input.yml
```