对Kafka和阿夫罗来说非常陌生。我被一个问题困住了,似乎不知道这里出了什么问题。我写了一个Kafka的生产者和消费者,它使用avro作为序列化格式。生产者代码工作正常。当我运行 kafka-avro-console-consumer
它给了我如下信息-
bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic test --property schema.registry.url=http://127.0.0.1:8081 --from-beginning
{"name":{"string":"Hello World!"}}
{"name":{"string":"Hello World!"}}
{"name":{"string":"Hello World!"}}
然而,当我尝试使用python做同样的事情时(下面是最基本的示例),我编写了以下代码-
from confluent_kafka import KafkaError
from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError
class AvroConsumerAdapter(object):
def __init__(self, topic='test'):
self.topic = topic
self.consumer = AvroConsumer({'bootstrap.servers': 'localhost:9092',
'schema.registry.url': 'http://127.0.0.1:8081',
'group.id': 'mygroup'})
self.consumer.subscribe([topic])
def start_consuming(self):
running = True
while running:
try:
msg = self.consumer.poll(10)
if msg:
print(msg.value())
if not msg.error():
print("Here - 1")
print(msg.value())
elif msg.error().code() != KafkaError._PARTITION_EOF:
print("here-2")
print(msg.error())
running = False
else:
print('Here-3')
print(msg.error())
except SerializerError as e:
print("Message deserialization failed for %s: %s" % (msg, e))
running = False
except Exception as ex:
print(ex)
running = False
self.consumer.close()
这个客户永远呆在那里从不打印任何东西。我不确定这里出了什么问题。谁能帮我一下吗。
1条答案
按热度按时间mlmc2os51#
查看主题配置选项--您需要设置
auto.offset.reset': 'smallest'
如果要处理主题中当前的所有数据。默认情况下是largest
这意味着它将只显示生成的新行数据。您可以通过让当前的python代码保持运行并生成新的主题消息来验证这一点—您应该看到python代码将它们收集起来。