kafka-python開發kafka生產者和消費者

 

1、安裝kafka-python 

執行命令

pip install kafka-python

kafka-python        1.4.6

 

2、編寫python kafka 生產者消費者代碼

# test.py

import sys
import time
import json

from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError


KAFAKA_HOST = "127.0.0.1"
KAFAKA_PORT = 9092
KAFAKA_TOPIC = "test123"


class Kafka_producer():
    '''''
    生產模塊：根據不同的key，區分消息
    '''

    def __init__(self, kafkahost,kafkaport, kafkatopic, key):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.key = key
        print("producer:h,p,t,k",kafkahost,kafkaport,kafkatopic,key)
        bootstrap_servers = '{kafka_host}:{kafka_port}'.format(
                kafka_host=self.kafkaHost,
                kafka_port=self.kafkaPort
                )
        print("boot svr:",bootstrap_servers)
        self.producer = KafkaProducer(bootstrap_servers = bootstrap_servers
                )

    def sendjsondata(self, params):
        try:
            parmas_message = json.dumps(params,ensure_ascii=False)
            producer = self.producer
            print(parmas_message)
            v = parmas_message.encode('utf-8')
            k = key.encode('utf-8')
            print("send msg:(k,v)",k,v)
            producer.send(self.kafkatopic, key=k, value= v)
            producer.flush()
        except KafkaError as e:
            print (e)

class Kafka_consumer():
    '''''
    消費模塊: 通過不同groupid消費topic里面的消息
    '''

    def __init__(self, kafkahost, kafkaport, kafkatopic, groupid):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.groupid = groupid
        self.key = key
        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:
                yield message
                print("1")
                print(message)
        except KeyboardInterrupt as e:
            print (e)


def main(xtype, group, key):
    '''''
    測試consumer和producer
    '''
    if xtype == "p":
        # 生產模塊
        producer = Kafka_producer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, key)
        print ("===========> producer:", producer)
        for _id in range(100):
          #  params = '{"msg" : "%s"}' % str(_id)
           params=[{"msg0" :_id},{"msg1" :_id}]
           producer.sendjsondata(params)
           time.sleep(1)

    if xtype == 'c':
        # 消費模塊
        consumer = Kafka_consumer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, group)
        print ("===========> consumer:", consumer)
        message = consumer.consume_data()
        print('2')
        print(message)
        for msg in message:
            print ('msg---------------->k,v', msg.key,msg.value)
            print ('offset---------------->', msg.offset)

if __name__ == '__main__':
    xtype = sys.argv[1]
    group = sys.argv[2]
    key = sys.argv[3]
    main(xtype, group, key)

 

3、啟動kafka服務

打開終端 輸入命令：

kafka-server-start /usr/local/etc/kafka/server.properties

 

4、新開一終端創建生產者

切換到程序路徑執行如下指令

python test.py p g k

 

5、新開一終端創建消費者

切換到程序路徑執行如下指令

python test.py c g k

 

 

至此已經完成kafka 的消息收發了。