首页 文章

Rabbitmq队列与PHP中的进程分叉

提问于
浏览 次
0

我有基于PHP的标准AMQP类的简单队列工作者 . 它与RabbitMQ一起用作服务器 . 我有Queue Class用于初始化AMQP连接与RabbitMQ . 以下代码一切正常：

$queue = new Queue('myQueue');

 while($envelope = $queue->getEnvelope()) {
   $command = unserialize($envelope->getBody());

   if ($command instanceof QueueCommand) {
     try {
       if ($command->execute()) {
         $queue->ack($envelope->getDeliveryTag());
       }
     } catch (Exception $exc) {
       // an error occurred so do some processing to deal with it
     }
   }
 }

但是我想分叉队列命令执行，但在这种情况下，队列一遍又一遍地使用第一个命令 . 我不能承认RabbitMQ消息是通过$ queue-> ack（）收到的;我的分叉版本（为了测试而简化只有一个孩子）看起来像这样：

$queue = new Queue('myQueue');

while($envelope = $queue->getEnvelope()) {
  $command = unserialize($envelope->getBody());

  if ($command instanceof QueueCommand) {
    $pid = pcntl_fork();

    if ($pid) {
      //parent proces
      //wait for child
      pcntl_waitpid($pid, $status, WUNTRACED);

      if($status > 0) {
        // an error occurred so do some processing to deal with it
      } else {
        //remove Command from queue
        $queue->ack($envelope->getDeliveryTag());
      }
    } else {
      //child process
      try {
        if ($command->execute()) {
          exit(0);
        }
      } catch (Exception $exc) {
        exit(1);
      }
    }
  }
}

任何帮助将不胜感激...

php rabbitmq amqp

1 回答

  • 2

    我终于解决了这个问题！我不得不从子进程运行ack命令，它以这种方式工作！这是正确的代码：

    $queue = new Queue('myQueue');

while($envelope = $queue->getEnvelope()) {
  $command = unserialize($envelope->getBody());

  if ($command instanceof QueueCommand) {
    $pid = pcntl_fork();

    if ($pid) {
      //parent proces
      //wit for child
      pcntl_waitpid($pid, $status, WUNTRACED);

      if($status > 0) {
        // an error occurred so do some processing to deal with it
      } else {
        // sucess
      }
    } else {
      //child process
      try {
        if ($command->execute()) {
          $queue->ack($envelope->getDeliveryTag());
          exit(0);
        }
      } catch (Exception $exc) {
        exit(1);
      }
    }
  }
}
    回复于

相关问题