kafka_python_basic_producer_consumer_quickstart.py

1from kafka import KafkaConsumer, KafkaProducer
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 put on the network
9# to be sent to the broker.
10producer.flush()
11
12# Consume messages
13consumer = KafkaConsumer('my-topic', bootstrap_servers='localhost:9092')
14for message in consumer:
15    # message value and key are raw bytes -- decode if necessary!
16    # e.g., for unicode: `message.value.decode('utf-8')`
17    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
18                                          message.offset, message.key,
19                                          message.value))