[python] kafka연동

 

1. other app / views.py 

from django.http import HttpResponse
from .kafka_producer_consumer import send_to_kafka

# 다른 app의 views.py에서 kafka에 message를 보내는 함수 호출

def other_view(request):
  send_to_kafka('test','hello')
  return HttpResponse("Message sent to Kafka")

 

2. send_to_kafka

from kafka import KafkaProducer, KafkaConsumer

def send_to_kafka(topic, message):
    producer = KafkaProducer(
       bootstrap_servers=['kafka1:9092', 'kafka2:9092', 'kafka3:9092']
    )
    producer.send(topic, value=message.encode("utf-8"))
    producer.flush()
    producer.close()

 

3. message_from_kafka

def consume_from_kafka(topic):
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=['kafka1:9092', 'kafka2:9092', 'kafka3:9092']
    )

    print('[start]')

    for message in consumer:
        print("Topic: %s, Value: %s" % (message.topic, message.value.decode('utf-8')))

    print('[end]')