Python3 安装模块

pip3 install kafka-python

启动kafka

消费者

from kafka import KafkaConsumer

#consumer=KafkaConsumer('result',group_id='consumer-20171017',bootstrap_servers=['192.168.126.132:9092'])
consumer = KafkaConsumer('result')
for msg in consumer:
    print((msg.value).decode('utf8')) #以utf-8解码消息

生产者

from kafka import KafkaProducer
import csv
import time

producer = KafkaProducer(bootstrap_servers='localhost:9092') #定义生产者
csvfile = open("../data/user_log.csv","r")
reader = csv.reader(csvfile)  #处理csv格式数据文件

for line in reader:
    gender = line[9]
    if gender == 'gender':
        continue
    print(line[9])
    time.sleep(0.1)
    producer.send('sex',line[9].encode('utf8'))#向topic sex中发消息