环境安装
1. 安装Python。（Python版本为2.7或3.X。）
2. 安装依赖库。（使用公网连接需要安装confluent-kafka 1.9.2及以下版本的依赖库）
pip install confluent-kafka==1.9.2 |
3. 下载Demo包kafka-confluent-python-demo.zip。
配置修改
1. 如果是ssl连接，需要在控制台下载证书。并且解压压缩包得到ssl.client.truststore.jks，执行以下命令生成caRoot.pem文件。
keytool -importkeystore -srckeystore ssl.client.truststore.jks -destkeystore caRoot.p12 -deststoretype pkcs12 openssl pkcs12 -in caRoot.p12 -out caRoot.pem |
2. 修改setting.py文件。（ca_location仅在ssl连接时需要配置）
kafka_setting = { 'bootstrap_servers': 'XXX', 'topic_name': 'XXX', 'group_name': 'XXX' } |
生产消息
发送以下命令发送消息。
python kafka_producer.py |
生产消息示例代码如下：
from confluent_kafka import Producer import setting conf = setting.kafka_setting """初始化一个 Producer 对象""" p = Producer({'bootstrap.servers': conf['bootstrap_servers']}) def delivery_report(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush(). """ if err is not None: print('Message delivery failed: {}'.format(err)) else: print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition())) """异步发送消息""" p.produce(conf['topic_name'], "Hello".encode('utf-8'), callback=delivery_report) p.poll(0) """在程序结束时，调用flush""" p.flush() |
消费消息
发送以下命令消费消息。
python kafka_consumer.py |
消费消息示例代码如下：
from confluent_kafka import Consumer, KafkaError import setting conf = setting.kafka_setting c = Consumer({ 'bootstrap.servers': conf['bootstrap_servers'], 'group.id': conf['group_name'], 'auto.offset.reset': 'latest' }) c.subscribe([conf['topic_name']]) while True: msg = c.poll(1.0) if msg is None: continue if msg.error(): if msg.error().code() == KafkaError._PARTITION_EOF: continue else: print("Consumer error: {}".format(msg.error())) continue print('Received message: {}'.format(msg.value().decode('utf-8'))) c.close() |