我想为三个具有相同队列的类似应用程序设置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 = 'Kafka'Consumer::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
但这并没有奏效。
您实际上是在破坏Kafka消费者，方法是将其声明到foreach循环中，将其移出循环，使用循环设置分区，然后获取结果并区分来源：
$queues = array("TopicX", "TopicY", "TopicZ");
$consumer = 'Kafka'Consumer::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以查看示例。