カフカ - 新しいメッセージを見る方法

Question

私はカフカメッセージを消費するPHPアプリケーションを持っています。 カフカに新しいメッセージがあることを知る方法は問題ですか？ 最初の解決策は、コンシューマーをPHPで作成し、それをループで実行して新しいメッセージをチェックすることです。このようなもの

<?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"); 
    } 
} 

しかし、私はそれが好きではありません...他の方法はありますか？

良い方法がない場合は、それをどのように実行するのですか？たとえば、何かが失敗したとき。

Answer 1

私が知る限り、無限にループして新しいメッセージをチェックするよりも良い方法はありません。一般的なアプローチは、一定の時間または反復回数の後にタスクを終了させて​​から、すべてのリソースを奪うのを防ぐために、死を検出して消費者を復活させるためにsupervisordのようなものを使用することです。