我有一个使用kafka消息的php应用程序。问题是如何知道Kafka有新的信息?第一种解决方案是在php中创建consumer,然后在循环中运行它来检查新消息。像这样的
<?php
namespace MyAppBundle\Command;
use MyAppBundle\EventSourcing\EventSerializer\JSONEventSerializer;
use MyAppBundle\Service\EventProjectorService;
use MyAppBundle\Service\KafkaService;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\Exception\RuntimeException;
class EventCommand extends ContainerAwareCommand
{
protected function configure()
{
$this
->setName('events:fetch');
}
protected function execute(InputInterface $input, OutputInterface $output)
{
/**@var KafkaService $kafkaService */
$kafkaService = $this->getContainer()->get('store_locator.kafka_service');
/**@var EventProjectorService $eventProjector */
$eventProjector = $this->getContainer()->get('store_locator.event_projector');
while(1){
$messages = $kafkaService->fetchEvents();
foreach ($messages as $message) {
$eventProjector->aggregate($message);
}
}
$output->writeln("Finish");
}
}
但我不喜欢。。。还有别的办法吗?
如果没有更好的方法,如何让它继续运行?例如当某件事失败时。
1条答案
按热度按时间11dmarpk1#
据我所知,没有比无限循环和不断检查新消息更好的方法了。一种常见的方法是让任务在经过一定的时间或迭代次数后自杀,然后使用
supervisord
以检测死亡和复活消费者,以防止它吞噬你所有的资源。