kafka_python_producer_consumer_quickstart_example.py

1from kafka import KafkaProducer, KafkaConsumer
2
3# Produce messages
4producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
5for _ in range(100):
6    producer.send('my-topic', b'some_message_bytes')
7
8# Block until all pending messages are at least sent to the network
9producer.flush()
10
11# Consume messages
12consumer = KafkaConsumer('my-topic',
13                         group_id='my-group',
14                         bootstrap_servers=['localhost:9092'])
15for message in consumer:
16    # message value and key are raw bytes -- decode if necessary!
17    # e.g., setting value_deserializer=lambda m: json.loads(m.decode('utf-8'))
18    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
19                                          message.offset, message.key,
20                                          message.value))