单节点多代理配置上的kafka消费者

Question

有人在 kafka python 上工作过单节点多代理设置吗？

我能够使用单节点单代理设置来生成和使用数据，但是当我将其更改为单节点时，多代理数据被生成并存储在主题中，但是当我运行消费者代码时，数据没有被使用。

任何关于上述内容的建议将不胜感激。提前致谢！

注意：生产者、消费者和服务器属性等所有配置都经过验证并且正常。

生产者代码：

from kafka.producer import KafkaProducer

def producer():
    data = {'desc' : 'testing', 'data' : 'testing single node multi broker'}
    topic = 'INTERNAL' 
    producer = KafkaProducer(value_serializer=lambda v:json.dumps(v).encode('utf-8'), bootstrap_servers=["localhost:9092","localhost:9093","localhost:9094"])

    producer.send(topic, data)

    producer.flush()

消费者代码：

from kafka.consumer import KafkaConsumer

def consumer():  
    topic = 'INTERNAL'
    consumer = KafkaConsumer(topic,bootstrap_servers=["localhost:9092","localhost:9093","localhost:9094"])


    for data in consumer:
        print data

服务器 1 配置：我为其他代理添加了两个类似这样的服务器文件，其参数相同，broker.id、log.dirs 值不同。

broker.id=1
port=9092
num.network.threads=3
log.dirs=/tmp/kafka-logs-1
num.partitions=3
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=localhost:2181
delete.topic.enable=true

生产者配置：

metadata.broker.list=localhost:9092,localhost:9093,localhost:9094

消费者配置：

zookeeper.connect=127.0.0.1:2181
zookeeper.connection.timeout.ms=6000

Answer 1

您是否通过简单的 Kafka 消费者接收消息？

bin/kafka-console-consumer.sh –bootstrap-server localhost:9092,localhost:9093,localhost:9094 –topic INTERNAL –from-beginning

或者用这个：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic INTERNAL

如果您使用第二个命令收到消息，请尝试删除您的代理的/tmp/log.dir 并将文件记录在/tmp/zookeepker/version-2/ 中。然后重新启动 zookeeper 和您的代理并再次创建您的主题。