Kafka 通过python简单的生产消费实现

Posted 2020-11-16 xibuhaohao

使用CentOS6.5、python3.6、kafkaScala 2.10  - kafka_2.10-0.8.2.2.tgz (asc, md5)

一、下载kafka

下载地址

https://kafka.apache.org/downloads

里面包含zookeeper

二、安装Kafka

1、安装zookeeper

mkdir /root/kafka/

tar -vzxf kafka_2.10-0.8.2.2

 

cd /root/kafka/kafka_2.10-0.8.2.2

cat  config/zookeeper.properties | grep -v \'#\' >> config/zk.properties

mkdir -p /home/kafka/zk

vi zk.properties
dataDir=/home/kafka/zk  #因为zookeeper变更为zk,所以需要在这里修改一下

启动zookeeper(后台启动)

/root/kafka/kafka_2.10-0.8.2.2/bin/zookeeper-server-start.sh /root/kafka/kafka_2.10-0.8.2.2/config/zk.properties &

2、安装Kafka

cd /root/kafka/kafka_2.10-0.8.2.2

cat config/server.properties | grep -v \'#\'  >> config/kafka_01.properties

启动Kafka(后台启动)

/root/kafka/kafka_2.10-0.8.2.2/bin/kafka-server-start.sh /root/kafka/kafka_2.10-0.8.2.2/config/kafka_01.properties &

三、新建Kafka topic

1、新建topic

cd /root/kafka/kafka_2.10-0.8.2.2

./bin/kafka-topics.sh --create --zookeeper 192.168.50.33:2181 --replication-factor 1 --partitions 1 --topic test

2、查看topic

./bin/kafka-topics.sh --list --zookeeper 192.168.50.33:2181

四、kafka生产者脚本

1、安装python的Kafka模块

pip3 install kafka-python（之前已安装）

 

 

2、kafka生产者脚本

cat kafka_pro.py
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import time

class Kafka_producer():
    \'\'\'
    使用kafka的生产模块
    \'\'\'

    def __init__(self, kafkahost, kafkaport, kafkatopic):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.producer = KafkaProducer(bootstrap_servers=\'{kafka_host}:{kafka_port}\'.format(
            kafka_host=self.kafkaHost,
            kafka_port=self.kafkaPort
        ))

    def sendjsondata(self, params):
        try:
            parmas_message = json.dumps(params)
            producer = self.producer
            producer.send(self.kafkatopic, parmas_message.encode(\'utf-8\'))
            producer.flush()
        except KafkaError as e:
            print(e)

def main():
    \'\'\'
    测试consumer和producer
    :return:
    \'\'\'
    # 测试生产模块
    producer = Kafka_producer("127.0.0.1",9092,"test")
    for i in range(1000000000000):
        params = \'test---\' + str(i)
        print(params)
        producer.sendjsondata(params)
        time.sleep(1)

if __name__ == \'__main__\':
    main()
    import os
    print(os.uname)

 

 

五、kafka消费者脚本

cat kafka_cust.py
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import time

class Kafka_consumer():
    \'\'\'
    使用Kafka—python的消费模块
    \'\'\'

    def __init__(self, kafkahost, kafkaport, kafkatopic, groupid):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.groupid = groupid
        self.consumer = KafkaConsumer(self.kafkatopic, group_id=self.groupid,
                                      bootstrap_servers=\'{kafka_host}:{kafka_port}\'.format(
                                          kafka_host=self.kafkaHost,
                                          kafka_port=self.kafkaPort))

    def consume_data(self):
        try:
            for message in self.consumer:
                # print json.loads(message.value)
                yield message
        except KeyboardInterrupt as e:
            print(e)

def main():
    \'\'\'
    测试consumer和producer
    :return:
    \'\'\'
    # 测试消费模块
    # 消费模块的返回格式为ConsumerRecord(topic=u\'ranktest\', partition=0, offset=202, timestamp=None,
    # \\timestamp_type=None, key=None, value=\'"{abetst}:{null}---0"\', checksum=-1868164195,
    # \\serialized_key_size=-1, serialized_value_size=21)
    consumer = Kafka_consumer(\'127.0.0.1\',
                              9092,
                              "test",
                              \'test-python-test\')
    message = consumer.consume_data()
    for i in message:
        print(i.value)

if __name__ == \'__main__\':
    main()

 

 

整理自：

https://www.cnblogs.com/hunttown/p/9041036.html

https://gitee.com/jalright/scriptstodo/blob/master/kafka/producer.py

https://gitee.com/jalright/scriptstodo/blob/master/kafka/cunsumer.py