kafka群集设置的读取套接字超时错误

Question

我想为三个具有相同队列的类似应用程序设置kafka集群类型设置，如AppA->{TopicX，TopicY，TopicZ}，AppB->{TopicX，TopicZ}，AppC->{Topic X，Topic Y}。生产者和消费者将是特定于应用程序的。我在具有不同端口的三个不同配置文件中设置了具有分区1,2,3的三个代理的kafka集群。然后启动kafka服务器（集群）

我使用kafkaphp包装器http://github.com/nmred/kafka-php

所以我使用了类似的应用程序A的Producer代码

       $producer->setRequireAck(-1);
       $producer->setMessages("TopicX", 0, array(json_encode($this->data)));
       $producer->send();

并使用了类似的应用程序B的生产者代码

       $producer->setRequireAck(-1);
       $producer->setMessages("TopicX", 1, array(json_encode($this->data)));
       $producer->send();

等等。

然后我为等三个应用程序制作了消费者脚本

        $queues = array("TopicX", "TopicY", "TopicZ");
        while(true) {
            foreach($queues as $queue) {
                $consumer = KafkaConsumer::getInstance('localhost:2181');
                $consumer->setGroup('testgroup');
                $consumer->setPartition($queue, 0);
                $result = $consumer->fetch();
           }
        }

但当我尝试为任何应用程序执行消费者脚本时，我都会遇到类似的错误

"读取750437字节时读取套接字超时，还有750323字节"

我只是不知道如何解决这个问题我试图修改一些kafka配置参数，比如

 zookeeper.connection.timeout.ms=24000         # Initially 6000
 replica.socket.timeout.ms=15000                      # Not exists in default file

但这并没有奏效。

Answer 1

您实际上是在破坏Kafka消费者，方法是将其声明到foreach循环中，将其移出循环，使用循环设置分区，然后获取结果并区分来源：

$queues = array("TopicX", "TopicY", "TopicZ");
$consumer = KafkaConsumer::getInstance('localhost:2181');
$consumer->setGroup('testgroup');
foreach($queues as $queue) {
  $consumer->setPartition($queue, 0);
}
while(true) {
  $result = $consumer->fetch();
  foreach ($result as $topicName => $topic) {
    foreach ($topic as $partId => $partition) {
      var_dump($partition->getHighOffset());
      foreach ($partition as $message) {
        var_dump((string)$message);
      }
    }
  }
}

请参阅kafkaphp项目的github README以查看示例。