kafka_python_ng_producer_consumer_json_messages_quickstart.py

1from kafka import KafkaProducer, KafkaConsumer
2import json
3
4# --- Producer Example ---
5# Initialize the producer
6producer = KafkaProducer(
7    bootstrap_servers=['localhost:9092'],
8    value_serializer=lambda v: json.dumps(v).encode('utf-8')
9)
10
11# Send a message
12producer.send('test-topic', {'key': 'value'})
13producer.flush()
14print("Message sent successfully")
15
16# --- Consumer Example ---
17# Initialize the consumer
18consumer = KafkaConsumer(
19    'test-topic',
20    bootstrap_servers=['localhost:9092'],
21    auto_offset_reset='earliest',
22    group_id='my-group',
23    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
24)
25
26# Read messages
27print("Waiting for messages...")
28for message in consumer:
29    print(f"Received message: {message.value}")
30    # Break after first message for demonstration purposes
31    break