背景
写一个链上的竞猜小程序，由于交易的数据都在kafka里，并且使用了sasl进行认证。所以需要连接到kafka获取交易数据。解决两个问题: 1、连接kafka，2、sasl认证。直接上代码
代码示例
from kafka import KafkaConsumer
import time
import json
BOOTSTRAP_SERVERS='localhost:9102'
TOPIC='topic'
consumer = KafkaConsumer(TOPIC,
bootstrap_servers=BOOTSTRAP_SERVERS,
auto_offset_reset='earliest',
security_protocol='SASL_PLAINTEXT',
sasl_mechanism='PLAIN',
sasl_plain_username='username',
sasl_plain_password='password',
api_version=(0,10),
receive_buffer_bytes=1024,
enable_auto_commit='False')
# consumer_timeout_ms=1000)
# Consumption log
for msg in consumer:
print(msg)
填坑记录
- 参数必须要指定api_version，否则在连接时会提示一个check_version的异常，注意中间是逗号
- 默认是从最新的消息开始消费，如果想要获取全部消息，需要设置auto_offset_reset参数，值为earliest
- 参数security_protocol='SASL_PLAINTEXT'指定后， sasl_mechanism参数只能是PLAIN，否则连接后获取不到消息。
- consumer_timeout_ms 参数没设置，程序会一直阻塞。否则会再设置的时间到期后继续执行。
