multiprocess.py

import logging as log
import time
import multiprocessing

from pykafka import KafkaClient
from pykafka.common import OffsetType

log.basicConfig(level=log.DEBUG)

client = KafkaClient(hosts="127.0.0.1:9092")


def run_consumer():
    topic = client.topics['test2']
    log.info("Got the topic")
    consumer = topic.get_simple_consumer("test")
    log.info("initialized consumer")
    consumer.start()
    time.sleep(2)

    while True:
        try:
            message = consumer.consume()
            if message:
                print message.value
        except:
            break


def run_consumers():
    t = multiprocessing.Process(target=run_consumer, name="consumer")
    t.daemon = False # True
    t.start()
    t.join()

if __name__ == "__main__":
    run_consumers()