Kafak简单使用

本文主要是介绍Kafak简单使用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

• Concept
  待后续填坑…
• Push Data

from kafka import KafkaProducer
import jsondef push_kafka(sqlstring, valuelist):# logging.info("kafka string ----- [%s]" % (sqlstring % valuelist))producer = KafkaProducer(bootstrap_servers=["ip1:9092", "ip2:9092", "ip3:9092"])try:msg_dict = {"data": {'sqlstring': sqlstring,'paramslist': valuelist}}msg = json.dumps(msg_dict)producer.send("xx3_ccccc", value=msg.encode("utf-8"))except Exception as e:logging.exception('push_kafka Exception[%s]', str(e))

• Consumer

def consumer_session():consumer = KafkaConsumer("xx3_ccccc",bootstrap_servers=["ip1:9092", "ip2:9092", "ip3:9092"])while True:c_msgs = consumer.poll(max_records=300).values()for m_bo in c_msgs:for msg in m_bo:c_msg = msg.value.decode("utf-8")d_msg = json.loads(c_msg)try:cur.execute(d_msg['data']["sqlstring"], d_msg['data']["paramslist"])conn.commit()except Exception as e:logging.error("insert error：%s", e)# 关闭游标和连接cur.close()conn.close()

未完待续…

这篇关于Kafak简单使用的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---