python2.7简单实现kafka消费和生产信息

1.生产者

from pykafka import KafkaClient
#from kafka import KafkaProducer

import simplejson as json
import logging
import time
import sys




#logging.basicConfig(level = logging.INFO)
client = KafkaClient(hosts = "47.104.108.21:9092" )

topic = client.topics['test']


producer = topic.get_producer()
producer.start()
   
  # 生产消息
msg_dict = {
    "sleep_time": 10,
    "db_config" : {
      "database"  : "test",
      "host"      : "192.168.137.12",
      "user"      : "root",
      "password"  : "root"
    },
    "table"     : "msg",
    "msg"       : "Hello World"
  }
msg = json.dumps(msg_dict)
producer.produce(msg)
producer.stop()

2.消费者

from pykafka import KafkaClient  
host = '47.104.108.21'  
client = KafkaClient(hosts="%s:9092" % host)  
  
print client.topics  
  
  
# 消费者  
topic = client.topics['test']  
consumer = topic.get_simple_consumer(consumer_group='test', auto_commit_enable=True, consumer_id='test')  
for message in consumer:  
    if message is not None:  
        print message.offset, message.value 

 

 

                                                                                                                           欢迎关注