from kafka import KafkaProducer, KafkaConsumer
import time
import json
topic = 'my-topic'
def test_consumer():
consumer = KafkaConsumer(topic, group_id='test_id', enable_auto_commit=True, auto_commit_interval_ms=5000, bootstrap_servers=['xxxxx:30901', 'xxxx:30902', 'xxxx:30903'])
for message in consumer:
yield message
def get_data():
message = test_consumer()
for msg in message:
try:
msg_json = json.loads(msg.value.decode('utf-8'))
key = msg.key.decode()
print(msg_json)
except Exception as e:
print(e)
def test_producer():
print('begin')
n = time.time()
producer = KafkaProducer(bootstrap_servers=['xxxxx:30901', 'xxxx:30902', 'xxxx:30903'])
msg = {"send" : str(n)}
parmas_message = json.dumps(msg)
# producer.send(topic, parmas_message.encode('utf-8'))
producer.send(topic, key=bytes("send", encoding='utf-8'), value=bytes(parmas_message, encoding='utf-8'))
print ("send" + str(n))
time.sleep(0.5)
print('done')
if __name__ == '__main__':
# test_producer()
# test_consumer()
get_data()
