Code for my blog

consumer.py

from kafka import KafkaConsumer


consumer1 = KafkaConsumer('payments', group_id="group1", bootstrap_servers='localhost:9092')

for msg in consumer1:
    print(msg)

consumer2.py

from kafka import KafkaConsumer


consumer1 = KafkaConsumer('payments', group_id="group1", bootstrap_servers='localhost:9092')

for msg in consumer1:
    print(msg)

consumer3.py

from kafka import KafkaConsumer


consumer1 = KafkaConsumer('payments', group_id="group2", bootstrap_servers='localhost:9092')

for msg in consumer1:
    print(msg)

consumer4.py

from kafka import KafkaConsumer


consumer1 = KafkaConsumer('payments', group_id="group2", bootstrap_servers='localhost:9092')

for msg in consumer1:
    print(msg)

producer.py

import json
import random

from kafka import KafkaProducer
from faker import Faker

obj = Faker()

producer = KafkaProducer(bootstrap_servers='localhost:9092', key_serializer=str.encode, value_serializer=lambda v: json.dumps(v).encode('utf-8'))

usernames = [obj.user_name() for _ in range(100)]

data = [{"username": usernames[random.randint(0, 9)], "address": obj.address()} for _ in range(10000)]

for each_data in data:
    print("sending message %s" % each_data)
    producer.send('payments', key=each_data["username"], value=each_data)

requirements.txt

faker
python-kafka