我在 python 中运行 confluent_kafka 客户端。目前我在尝试生产然后消费消息时没有收到任何错误，但问题是生产者说它成功了，但消费者找不到任何消息。
我创建了一个主题，这是我正在使用的构建的类：
from confluent_kafka import Producer, Consumer
from config import config
import json
class Kafka:
"""
Kafka Handler.
"""
def __init__(self, kafka_brokers_sasl, api_key):
"""
Arguments:
kafka_brokers_sasl {str} -- String containing kafka brokers separated by comma (no spaces)
api_key {str} -- Kafka Api Key
"""
self.driver_options = {
'bootstrap.servers': kafka_brokers_sasl,
'sasl.mechanisms': 'PLAIN',
'security.protocol': 'SASL_SSL',
'sasl.username': 'token',
'sasl.password': api_key,
'log.connection.close' : False,
#'debug': 'all'
}
self.producer_options = {
'client.id': 'kafka-python-console-sample-producer'
}
self.producer_options.update(self.driver_options)
self.consumer_options = {
'client.id': 'kafka-python-console-sample-consumer',
'group.id': 'kafka-python-console-sample-group'
}
self.consumer_options.update(self.driver_options)
self.running = None
def stop(self):
self.running = False
def delivery_report(self, 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()))
def produce(self, topic, data): # Function for producing/uploading data to a Kafka topic
p = Producer(self.producer_options)
print("Running?")
# Asynchronously produce a message, the delivery report callback will be triggered from poll() above, or flush() below, when the message has been successfully delivered or failed permanently.
p.produce(topic, data, callback=self.delivery_report)
# Wait for any outstanding messages to be delivered and delivery report callbacks to be triggered.
p.flush()
print("Done?")
def consume(self, topic, method_class=None): # Function for consuming/reading data from a Kafka topic. Works as a listener and triggers the run() function on a method_class
print("raaa")
kafka_consumer = Consumer(self.consumer_options)
kafka_consumer.subscribe([topic])
# Now loop on the consumer to read messages
print("Running?")
self.running = True
while self.running:
msg = kafka_consumer.poll()
print(msg)
if msg is not None and msg.error() is None:
print('Message consumed: topic={0}, partition={1}, offset={2}, key={3}, value={4}'.format(
msg.topic(),
msg.partition(),
msg.offset(),
msg.key().decode('utf-8'),
msg.value().decode('utf-8')))
else:
print('No messages consumed')
print("Here?")
kafka_consumer.unsubscribe()
kafka_consumer.close()
print("Ending?")
mock = {'yas': 'yas', 'yas2': 'yas2'}
kafka = Kafka(config['kafka']['kafka_brokers_sasl'], config['kafka']['api_key'])
kafka.produce(config['kafka']['topic'], json.dumps(mock))
kafka.consume(config['kafka']['topic'])
运行这个我得到打印：
Running?
Message delivered to DANIEL_TEST [0]
Done?
raaa
Running?
<cimpl.Message object at 0x104e4c390>
No messages consumed