Kafka Topic을 발행하고 producer와 consumer 프로그램 만들기 - Python

새로운 Topic을 생성하자. blog-post 라는 Topic을 생성 해서 예제로 사용 해 보고자 한다.

bin/kafka-topics.sh --create --topic blog-post --bootstrap-server localhost:9092

개발환경 구축하기

• 가상환경 설정하기

virtualenv venv
soruce venv/bin/activate

• kafka module 설치하기

(venv) $ pip3 install kafka-python

producer.py

10000개의 데이터를 Queue에 보내는데 얼마나 걸리는지 측정 해 보기 위해 time을 사용 해 보았다.

import time
from kafka import KafkaProducer

# producer 객체 생성
# acks 0 -> 빠른 전송우선, acks 1 -> 데이터 정확성 우선
producer = KafkaProducer(acks=0, compression_type='gzip',bootstrap_servers=['localhost:9092'])

start = time.time()

for i in range(10000):
 producer.send('blog-post',b'Kafka Blog Post Event Message')
 producer.flush() #queue에 있는 데이터를 보냄

end = time.time() - start
print(end)

consumer.py

from kafka import KafkaConsumer, consumer

# consumer 객체 생성
consumer = KafkaConsumer(
    'blog-post',
    bootstrap_servers=['127.0.0.1:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    consumer_timeout_ms=1000
)

while True:
    for message in consumer:
        print(message.topic, message.partition, message.offset, message.key, message.value)

실행

• 시작하기 전, Topic을 생성했기 때문에 producer.py와 consumer.py만 실행 해 주면 된다.
• producer.py는 데이터를 발생시키는 친구이고, 10000개의 데이터를 ‘blog-post’ Topic으로 전송한다.
• consumer.py는 데이터를 읽어오는 친구이고, 무한루프를 만들었기 때문에 데이터를 받는대로 출력 해 준다.

> python producer.py
3.8476390838623047

> python consumer.py
...
blog-post 0 100 None b'Kafka Blog Post Event Message'
blog-post 0 101 None b'Kafka Blog Post Event Message'
blog-post 0 102 None b'Kafka Blog Post Event Message'
blog-post 0 103 None b'Kafka Blog Post Event Message'
...