Logstash接入Kafka专享版

更新时间：2024-01-03

前提条件

已创建消息服务 for kafka集群
下载并安装Logstash。具体操作，请参见Download Logstash。
下载并安装JDK 8。具体操作，请参见Download JDK 8。

Logstash接入

步骤一：获取接入点

具体请参考查看集群接入点

步骤二：创建/获取主题

具体请参考创建主题、查看主题详情

步骤三：下载证书文件

SSL/SASL方式下载证书参考：如何下载证书？。

Logstash 发送消息

准备配置文件

创建jaas.conf配置文件

输入以下内容。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  # 用户管理中创建用户的用户名
  username="XXX"
  # 用户管理中创建用户的密码
  password="XXX";
};

创建 output.conf 配置文件

输入以下内容。

input {
    stdin{}
}

output {
   kafka {
        bootstrap_servers => "接入点"
        topic_id => "topic名称"
         # 固定为 SSL
        security_protocol => "SSL"
        # jaas.conf 文件地址
        jaas_path => "/home/kafka/logstash-7.10.2/jaas.conf"
        # 配置为 client.truststore.jks 文件路径
        ssl.truststore.location =>  "/home/kafka/logstash-7.10.2/client.truststore.jks"
        #  固定为 bms@kafka
        ssl.truststore.password => "bms@kafka"
        # 配置为 kafka_client_jaas.conf 文件路径
        ssl.keystore.location => "/home/kafka/logstash-7.10.2/client.keystore.jks"
        # 设置为client_ssl.properties文件内ssl.keystore.password
        ssl.keystore.password => "qKiHmaov8GNR"
        # SSL接入点辨识算法。6.x及以上版本Logstash需要加上该参数。空值
        ssl_endpoint_identification_algorithm => ""
    }
}

Logstash发送消息

在安装目录下bin执行以下命令

./logstash -f output.conf

Logstash 消费消息

准备配置文件

创建jaas.conf配置文件

输入以下内容。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="XXX"
  password="XXX";
};

创建 input.conf 配置文件

输入以下内容。

input {
    kafka {
         bootstrap_servers => "接入点"
        # topic名称
        topics => ["logstash_test"]
        # 固定为 SSL
        security_protocol => "SSL"
        # jaas.conf 文件地址
        jaas_path => "/home/kafka/logstash-7.10.2/jaas.conf"
        # 配置为 client.truststore.jks 文件路径
        ssl.truststore.location =>  "/home/kafka/logstash-7.10.2/client.truststore.jks"
        #  固定为 bms@kafka
        ssl.truststore.password => "bms@kafka"
        # 配置为 kafka_client_jaas.conf 文件路径
        ssl.keystore.location => "/home/kafka/logstash-7.10.2/client.keystore.jks"
        # 设置为client_ssl.properties文件内ssl.keystore.password
        ssl.keystore.password => "qKiHmaov8GNR"

        group_id => "logstash_group"
        consumer_threads => 3
        auto_offset_reset => "earliest"
    }
}

output {
    stdout {
        codec => rubydebug
    }
}