[转帖]一、Kafka Tool使用

kafka,tool,使用 · 浏览次数 : 0

小编点评

**1. 配置Kafka Tool** ``` # 设置集群名 cluster_name="cluster2" # 开启SASL认证 security_protocol="SASL_PLAINTEXT" # 设置SASL认证用户名和密码 sasl_plain_username="kafka" sasl_plain_password="123456" # 设置SASL机制为PLAIN sasl.mechanism="PLAIN" # 设置高级设置 highlighter_nixorg.apache.kafka.common.security.plain.PlainLoginModule required_username="kafka" password="123456" ``` **2. 修改生产者脚本** ```python # 创建Kafka生产者 highlighter_iniproducer = KafkaProducer( sasl_mechanism="PLAIN", security_protocol="SASL_PLAINTEXT", sasl_plain_username=self.username, sasl_plain_password=self.password, bootstrap_servers=self.bootstrap_servers, value_serializer=lambda m: json.dumps(m).encode() ) ``` **3. 修改消费者脚本** ```python # 创建Kafka消费者 highlighter_routerosconsumer = KafkaConsumer( self.topic, sasl_mechanism="PLAIN", security_protocol="SASL_PLAINTEXT", sasl_plain_username=self.username, sasl_plain_password=self.password, bootstrap_servers=self.bootstrap_servers, consumer_timeout_ms=5000, group_id=group_id, auto_offset_reset=auto_offset_reset, enable_auto_commit=enable_auto_commit ) ``` **4. 添加排版代码** ```python # 打印数据读取完成的信息 log.warning("数据读取完成,数据读取超时,自动断开连接") # 返回数据列表 return lk_kafka ```

正文

一、Kafka Tool使用

1、添加cluster

2、开启SASL_PLAINTEXT

如果kafka 开启SASL_PLAINTEXT认证（用户名和密码认证）

3、高级设置

如果设置的是SASL Plaintext，则必须将sasl.mechanism客户端属性更改为PLAIN。可以在“高级”部分下的“ SASL机制”文本字段中输入此属性。

4、JAAS配置

highlighter- nix

org.apache.kafka.common.security.plain.PlainLoginModule required username = "kafka" password = "123456";

二、脚本修改

相应的脚本也需要对应的进行修改

1、生产者

highlighter- ini

producer = KafkaProducer(
            sasl_mechanism="PLAIN",
            security_protocol='SASL_PLAINTEXT',
            sasl_plain_username=self.username,
            sasl_plain_password=self.password,
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda m: json.dumps(m).encode())

2、消费者

highlighter- routeros

consumer = KafkaConsumer(self.topic,
                                 sasl_mechanism="PLAIN",
                                 security_protocol='SASL_PLAINTEXT',
                                 sasl_plain_username=self.username,
                                 sasl_plain_password=self.password,
                                 bootstrap_servers=self.bootstrap_servers,
                                 consumer_timeout_ms=5000,
                                 group_id=group_id,
                                 auto_offset_reset=auto_offset_reset,
                                 enable_auto_commit=enable_auto_commit
                                 )
        if consumer.bootstrap_connected():
            lk_kafka = []
            for message in consumer:
                msg = message.value.decode()
                lk_kafka.append(msg)
            log.warning("数据读取完成,数据读取超时,自动断开连接")
            return lk_kafka
        else:
            log.error("连接kafka失败,请确认连接信息是否正确")