python對kafka的基本操作
阿新 • • 發佈:2018-08-05
消息 server close trap cor timeout produce posit 拉取 -- coding:utf-8 --
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.structs import TopicPartition
import time
bootstrap_servers = []
class OperateKafka:
def init(self,bootstrap_servers,topic):
self.bootstrap_servers = bootstrap_servers
self.topic = topic
"""生產者""" def produce(self): producer = KafkaProducer(bootstrap_servers=self.bootstrap_servers) for i in range(4): msg = "msg%d" %i producer.send(self.topic,key=str(i),value=msg) producer.close() """一個消費者消費一個topic""" def consume(self): #consumer = KafkaConsumer(self.topic,auto_offset_reset=‘earliest‘,group_id="testgroup",bootstrap_servers=self.bootstrap_servers) consumer = KafkaConsumer(self.topic,bootstrap_servers=self.bootstrap_servers) print consumer.partitions_for_topic(self.topic) #獲取test主題的分區信息 print consumer.topics() #獲取主題列表 print consumer.subscription() #獲取當前消費者訂閱的主題 print consumer.assignment() #獲取當前消費者topic、分區信息 print consumer.beginning_offsets(consumer.assignment()) #獲取當前消費者可消費的偏移量 consumer.seek(TopicPartition(topic=self.topic, partition=0), 1) #重置偏移量,從第1個偏移量消費 for message in consumer: print ("%s:%d:%d: key=%s value=%s" % (message.topic,message.partition,message.offset, message.key,message.value)) """一個消費者訂閱多個topic """ def consume2(self): consumer = KafkaConsumer(bootstrap_servers=[‘192.168.124.201:9092‘]) consumer.subscribe(topics=(‘TEST‘,‘TEST2‘)) #訂閱要消費的主題 print consumer.topics() print consumer.position(TopicPartition(topic=‘TEST‘, partition=0)) #獲取當前主題的最新偏移量 for message in consumer: print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value)) """消費者(手動拉取消息)""" def consume3(self): consumer = KafkaConsumer(group_id="mygroup",max_poll_records=3,bootstrap_servers=[‘192.168.124.201:9092‘]) consumer.subscribe(topics=(‘TEST‘,‘TEST2‘)) while True: message = consumer.poll(timeout_ms=5) #從kafka獲取消息 if message: print message time.sleep(1)
def main():
bootstrap_servers = [‘192.168.124.201:9092‘]
topic = "TEST"
operateKafka = OperateKafka(bootstrap_servers,topic)
operateKafka.produce()
#operateKafka.consume()
#operateKafka.consume2()
operateKafka.consume3()
main()
python對kafka的基本操作