Shulou(Shulou.com)06/03 Report--

This article mainly shows you "python how to achieve the basic operation of kafka", the content is easy to understand, clear, hope to help you solve your doubts, the following let the editor lead you to study and learn "python how to achieve the basic operation of kafka" this article.

-- 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

"" producer "" 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 () "" one consumer consumes one 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) # get test topic partition information print consumer.topics () # get topic list print consumer.subscription () # get current consumer subscribed topic print consumer.assignment () # get current consumer topic, partition information print consumer.beginning_offsets (consumer.assignment ()) # get offset consumer.seek (TopicPartition (topic=self.topic, partition=0) that can be consumed by current consumer 1) # reset offset Consume for message in consumer: print ("% s:%d:%d: key=%s value=%s"% (message.topic,message.partition,message.offset, message.key,message.value)) "one consumer subscribes to multiple topic" def consume2 (self): consumer = KafkaConsumer (bootstrap_servers= ['192.168.124.201key=%s value=%s 9092']) consumer.subscribe (topics= (' TEST') "from the first offset 'TEST2')) # subscribe to the topic to consume print consumer.topics () print consumer.position (TopicPartition (topic='TEST', partition=0)) # get the latest offset of the current topic for message in consumer: print ("% s:%d:%d: key=%s value=%s"% (message.topic, message.partition, message.offset, message.key) Message.value) "" Consumer (pull messages manually) "" def consume3 (self): consumer = KafkaConsumer (group_id= "mygroup", max_poll_records=3,bootstrap_servers= ['192.168.124.201consumer 9092']) consumer.subscribe (topics= ('TEST') 'TEST2')) while True: message = consumer.poll (timeout_ms=5) # get message from kafka if message: print message time.sleep (1)

Def main ():

Bootstrap_servers = ['192.168.124.201purl 9092']

Topic = "TEST"

OperateKafka = OperateKafka (bootstrap_servers,topic)

OperateKafka.produce ()

# operateKafka.consume ()

# operateKafka.consume2 ()

OperateKafka.consume3 ()

Main ()

The above is all the contents of the article "how to achieve the basic operation of kafka by python". Thank you for reading! I believe we all have a certain understanding, hope to share the content to help you, if you want to learn more knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.