如何在ConsumerGroup中设置Zookeeper IP地址而不是Kafka-Host IP地址？

Question

我必须在ConsumerGroup中设置zookeeper IP地址，而不是Kafka-Host IP地址。因为我将复制因子设置为3,并且创建了3个代理。因此，如果一个主机出现故障，那么另一个主机可以接管。

当我尝试将zookeeper的IP地址而不是Kafka-Host的IP地址放在ConsumerGroup中时，它不会接收到任何来自生产者API的消息。

var kafka = require('kafka-node')
var ConsumerGroup = kafka.ConsumerGroup

function createConsumerGroup () {
  var options = {
    kafkaHost: '127.0.0.1:9092',
    batch: undefined,
    ssl: true,
    groupId: 'demoExample',
    protocol: ['roundrobin'],
    encoding: 'utf8',
    fromOffset: 'latest',
    commitOffsetsOnFirstJoin: true,
    outOfRangeOffset: 'earliest',
    onRebalance: (isAlreadyMember, callback) => { callback(); }
  }

  var consumerGroup = new ConsumerGroup(Object.assign({ id: 'demo-' + process.pid }, options), 'example')

  consumerGroup.on('message', function (message) {
    message.value = JSON.parse(message.value)
    console.log('Message Received')
  })
}

我希望这样，如果我在ConsumerGroup中传递zookeeper的IP地址，而不是Kafka-Host的IP地址，它应该会接收到来自Producer API在“示例”主题上发送的消息。并且如果一个代理失败,那么它应该从另一个代理接收消息。因为复制因子设置为3并且创建了3个代理。

Answer 1

在Kafka0.9中引入的新消费者API不需要与Zookeeper连接。集团平衡现在由Kafaka自己处理。因此，您必须提供Kafka主机，而不是Zookeeper主机。

Confluent的博客文章应该能提供更多的信息：

在Apache Kafka的0.8.2版本发布重新设计的生产者客户机时，我们promise也会重新设计消费者客户机。我们信守诺言：0.9版本引入了对新重新设计的消费者客户端的beta支持。在高层次上，新的消费者的主要区别在于它消除了基于Zookeeper的“高级”消费者和“低级”SimpleConsumer API之间的区别，而是提供了一个统一的消费者API。

这个新的消费者是使用一个强大的新的服务器端设施来实现的，它使组管理成为Kafka协议的一个一流部分。这有几个好处。首先，它允许一个更可伸缩的组工具，它允许消费者客户机更简单、更薄，并允许更大的组以更快的速度进行重新平衡。所有客户均可使用这一设施；在C客户机LibrdKafka中使用它的工作已经接近完成。同样的工具在Kafka中对于管理数据的分布式、生产和消费是非常有用的；它是Kafaka连接以及几个即将到来的项目的基础。最后，这完成了在过去几年中完成的将Kafka客户端与Zookeeper完全脱钩的一系列项目，从而完全消除了消费者客户端对Zookeeper的依赖。Zookeeper仍然被Kafka使用，但它是代理的一个实现细节--使用这个新工具的客户机根本不需要连接到Zookeeper。这有许多操作上的好处，因为客户现在总是通过代理提供的安全和配额机制工作。这极大地简化了使用者，并为使用者API的一流非Java实现随着时间的推移而出现打开了大门。