PHP CLIスクリプトの非アクティブ期間後の実行機能

Question

RabbitMqBundleでSymfony2を使用して、ElasticSearchにドキュメントを送信するワーカーを作成しました。 ElasticSearch一括APIを使用するよりも、1つずつの割合でドキュメントのインデックスを作成する方がはるかに遅くなります。したがって、私は1000個のグループで文書をESにフラッシュするバッファを作成しました。コードルックス（ビット簡体字）としては、以下：

class SearchIndexator 
{ 
    protected $elasticaService; 
    protected $buffer = []; 
    protected $bufferSize = 0; 

    // The maximum number of documents to keep in the buffer. 
    // If the buffer reaches this amount of documents, then the buffers content 
    // is send to elasticsearch for indexation. 
    const MAX_BUFFER_SIZE = 1000; 

    public function __construct(ElasticaService $elasticaService) 
    { 
     $this->elasticaService = $elasticaService; 
    } 

    /** 
    * Destructor 
    * 
    * Flush any documents that remain in the buffer. 
    */ 
    public function __destruct() 
    { 
     $this->flush(); 
    } 

    /** 
    * Add a document to the indexation buffer. 
    */ 
    public function onMessage(array $document) 
    { 
     // Prepare the document for indexation. 
     $this->doHeavyWeightStuff($document); 

     // Create an Elastica document 
     $document = new \Elastica\Document(
      $document['key'], 
      $document 
     ); 

     // Add the document to the buffer. 
     $this->buffer[] = $document; 

     // Flush the buffer when max buffersize has been reached. 
     if (self::MAX_BUFFER_SIZE <= ++$this->bufferSize) { 
      $this->flush(); 
     } 
    } 

    /** 
    * Send the current buffer to ElasticSearch for indexation. 
    */ 
    public function flush() 
    { 
     // Send documents to ElasticSearch for indexation. 
     if (1 <= $this->bufferSize) { 
      $this->elasticaService->addDocuments($this->buffer); 
     } 

     // Clear buffer 
     $this->buffer = []; 
     $this->bufferSize = 0; 
    } 
} 

このすべてはかなり良い動作しますが、若干の問題があります。キューは予測不能な速度でメッセージで満たされます。 5分で100000になることもありますが、時には1時間に1回であることもあります。たとえば、82671のドキュメントがキューに入れられた場合、最後の671ドキュメントは、別の329ドキュメントを受信する前に索引付けされず、時間がかかることがあります。次のことができます：

警告：Sci-Fiコード！これは明らかに動作しません：今

class SearchIndexator 
{ 
    protected $elasticaService; 
    protected $buffer = []; 
    protected $bufferSize = 0; 
    protected $flushTimer; 

    // The maximum number of documents to keep in the buffer. 
    // If the buffer reaches this amount of documents, then the buffers content 
    // is send to elasticsearch for indexation. 
    const MAX_BUFFER_SIZE = 1000; 

    public function __construct(ElasticaService $elasticaService) 
    { 
     $this->elasticaService = $elasticaService; 

     // Highly Sci-fi code 
     $this->flushTimer = new Timer(); 
     // Flush buffer after 5 minutes of inactivity. 
     $this->flushTimer->setTimeout(5 * 60); 
     $this->flushTimer->setCallback([$this, 'flush']); 
    } 

    /** 
    * Destructor 
    * 
    * Flush any documents that remain in the buffer. 
    */ 
    public function __destruct() 
    { 
     $this->flush(); 
    } 

    /** 
    * Add a document to the indexation buffer. 
    */ 
    public function onMessage(array $document) 
    { 
     // Prepare the document for indexation. 
     $this->doHeavyWeightStuff($document); 

     // Create an Elastica document 
     $document = new \Elastica\Document(
      $document['key'], 
      $document 
     ); 

     // Add the document to the buffer. 
     $this->buffer[] = $document; 

     // Flush the buffer when max buffersize has been reached. 
     if (self::MAX_BUFFER_SIZE <= ++$this->bufferSize) { 
      $this->flush(); 
     } else { 
      // Start a timer that will flush the buffer after a timeout. 
      $this->initTimer(); 
     } 
    } 

    /** 
    * Send the current buffer to ElasticSearch for indexation. 
    */ 
    public function flush() 
    { 
     // Send documents to ElasticSearch for indexation. 
     if (1 <= $this->bufferSize) { 
      $this->elasticaService->addDocuments($this->buffer); 
     } 

     // Clear buffer 
     $this->buffer = []; 
     $this->bufferSize = 0; 

     // There are no longer messages to be send, stop the timer. 
     $this->flushTimer->stop(); 
    } 

    protected function initTimer() 
    { 
     // Start or restart timer 
     $this->flushTimer->isRunning() 
      ? $this->flushTimer->reset() 
      : $this->flushTimer->start(); 
    } 
} 

を、私は、イベントが駆動されていないPHPの限界を知っています。しかし、これは2015年であり、ReactPHPのような解決策があるので、これは可能でしょうか？ ØMQにはthis functionがあります。 RabbitMQやメッセージキューエクステンションとは独立して動作するソリューションは何でしょうか？私は懐疑的だ

ソリューション：は

1. crysalead/codeがあります。それはdeclare(ticks = 1);を使用してタイマーをシミュレートします。私はこれが実践的かつ堅実なアプローチであるかどうかはわかりません。何か案は？
2. 5分ごとに 'FLUSH'メッセージを同じキューに発行し、このメッセージを受け取ったときに明示的にバッファをフラッシュするが、それは不正行為になるcronジョブを実行できます。

Answer 1

私のコメントで述べたように、信号を使うことができます。 PHPではシグナルハンドラをスクリプトシグナル（SIGINT、SIGKILLなど）に登録することができます。

SIGALRMシグナルを使用することができます。この信号は、（設定可能な）一定時間が経過した後にあなたのスクリプトを警告します。これらの信号の正の側は、それらが非ブロッキングであることである。言い換えれば、あなたのスクリプトの通常の操作は妨げられません。

調整液（ダニはPHP 5.3以降推奨されません）：

function signal_handler($signal) { 
    // You would flush here 
    print "Caught SIGALRM\n"; 
    // Set the SIGALRM timer again or it won't trigger again 
    pcntl_alarm(300); 
} 

// register your handler with the SIGALRM signal 
pcntl_signal(SIGALRM, "signal_handler", true); 
// set the timeout for the SIGALRM signal to 300 seconds 
pcntl_alarm(300); 

// start loop and check for pending signals 
while(pcntl_signal_dispatch() && your_loop_condition) { 
    //Execute your code here 
} 

注：pcntl_alarmとあなたのためのタイマーをあなたの信号の時間を設定する場合にのみ、スクリプトで1つのSIGALRM信号を使用することができますアラームは、（信号を発射することなく）新たに設定された値にリセットされる。