public function __construct(Config $config)
 {
     $this->ownerPid = $config->get('ownerPid');
     $this->ipcKey = new IpcKey($this->ownerPid, str_replace('\\', '_', get_class($this)));
     $this->id = msg_get_queue($this->ipcKey->generate());
     $this->stat = msg_stat_queue($this->id);
 }

 /**
     Peek at a Message Queue Return the Number of Messages
 */
 static function mqPeek($name)
 {
     $mq = msg_get_queue(self::_ftok($name));
     $stat = msg_stat_queue($mq);
     return $stat['msg_qnum'];
     // Items in the Queue
 }

 private function fillQueueGeneralInformation()
 {
     $status = msg_stat_queue($this->resource);
     $this->ownerId = $status['msg_perm.uid'];
     $this->ownerGroupId = $status['msg_perm.gid'];
     $this->messageAllowedSize = $status['msg_qbytes'];
 }

 /**
  * {@inheritdoc}
  */
 public function count()
 {
     $data = msg_stat_queue($this->getQueue());
     if (!is_array($data)) {
         throw new QueueException($this, 'Failed to get the meta data for the queue.');
     }
     return $data['msg_qnum'];
 }

 public function putMessage(&$message)
 {
     $messageId = $this->storage->saveMessage($message);
     // notify queue via ipc
     $ipcStat = msg_stat_queue($this->ipcChannel);
     if (is_array($ipcStat) && $ipcStat['msg_qnum'] < 5) {
         msg_send($this->ipcChannel, 1, '*', false, false, $ipcError);
     }
     return $messageId;
 }

 /**
  * addMessage: Given a key, store a new message into our queue.
  *
  * @param $key string - Reference to the message (PK)
  * @param $data array - Some data to pass into the message
  */
 public static function addMessage($key, $data = array())
 {
     # What to send
     $message = new Message($key, $data);
     # Try to send the message
     if (msg_send(self::$queue, QUEUE_TYPE_START, $message)) {
         print_r(msg_stat_queue(self::$queue));
     } else {
         echo "Error adding to the queue";
     }
 }

 public static function initQueue($queue_file)
 {
     if (!file_exists($queue_file)) {
         touch($queue_file);
     }
     $msg_id = msg_get_queue(ftok($queue_file, 'r'), 0666);
     // Remove any messages in queue
     $queue = msg_stat_queue($msg_id);
     if ($queue['msg_qnum'] > 0) {
         msg_remove_queue($msg_id);
         $msg_id = msg_get_queue(ftok($queue_file, 'r'), 0666);
     }
     return $msg_id;
 }

 private function showAllDaemon()
 {
     $messageQueueKey = ftok(App::path('cache') . "/queue/daemon.queue", "a");
     $messageQueue = msg_get_queue($messageQueueKey, 0666);
     $messageQueueState = msg_stat_queue($messageQueue);
     $msgCount = $messageQueueState['msg_qnum'];
     if (0 === $msgCount) {
         Console::line("None service is running.");
     }
     while ($msgCount > 0) {
         /** @var MessageQueue $message */
         msg_receive($messageQueue, 0, $messageType, 1024, $message, true, MSG_IPC_NOWAIT);
         Console::line("PID:{$message->pid},NAME:{$message->name},TIME:" . date("Y-m-d H:i:s", $message->timestamp) . "Alive.");
         $messageQueueState = msg_stat_queue($messageQueue);
         $msgCount = $messageQueueState['msg_qnum'];
     }
 }

 /**
  * The class destructor.
  */
 public function __destruct()
 {
     parent::__destruct();
     if ($this->iAmParent()) {
         $desiredType = 1;
         $option_receive = MSG_IPC_NOWAIT;
         $stats = msg_stat_queue($this->queueResourceResponse);
         $queueMessageSize = $stats['msg_qbytes'];
         // the Report
         $messageReport = '';
         // receive the report
         $stats = msg_stat_queue($this->queueResourceResponse);
         while ($stats['msg_qnum']) {
             //for($i=0;count($all_servers);$i++){
             $status = msg_receive($this->queueResourceResponse, $desiredType, $type, $queueMessageSize, $mixed, true, $option_receive);
             if ($status == true) {
                 $messageReport .= $mixed['message'];
                 if ($mixed['return'] < 0) {
                     // check if server counldn't do the backups because shutdown doesn't work
                     if ($mixed['error'] && $mixed['error']['error']['error'] == '_ERR_VM_BACKUP_STILL_RUNNING_') {
                         $etva_server = $mixed['server'];
                         $this->log("[WARN] Receive error that VM '" . $etva_server->getName() . "' couldn't make backup beacuse is still running, so i will try start again...");
                         $etva_node = $etva_server->getEtvaNode();
                         if ($etva_node) {
                             $res_start = $etva_node->soapSend(EtvaServer_VA::SERVER_START, array('uuid' => $etva_server->getUuid(), 'name' => $etva_server->getName()));
                             $this->log("[WARN] Start '" . $etva_server->getName() . "' VM and receive the following message: " . print_r($res_start, true));
                         }
                     }
                 }
             } else {
                 $err_m = "[ERROR] Receive error when wait for response... " . print_r($err, true);
                 $messageReport .= $err_m . "\r\n";
                 $this->log($err_m);
             }
             $stats = msg_stat_queue($this->queueResourceResponse);
         }
         msg_remove_queue($this->queueResourceResponse);
         //$this->log($messageReport);
         $this->sendReport($messageReport);
     }
 }

 /**
  * {@inheritdoc}
  */
 public function count()
 {
     $stat = msg_stat_queue($this->queue);
     return $stat['msg_qnum'];
 }

<?php

// link: http://www.ebrueggeman.com/blog/creating-a-message-queue-in-php
// filename: add_to_queue.php
//creating a queue requires we come up with an arbitrary number
define('QUEUE', 21671);
//add message to queue
$queue = msg_get_queue(QUEUE);
// Create dummy message object
$object = new stdclass();
$object->name = 'foo';
$object->id = uniqid();
$object->data = array('name' => 'Ligia');
//try to add message to queue
if (msg_send($queue, 1, $object)) {
    echo "added to queue  \n";
    // you can use the msg_stat_queue() function to see queue status
    print_r(msg_stat_queue($queue));
} else {
    echo "could not add message to queue \n";
}

 public function stats()
 {
     return msg_stat_queue($this->queue);
 }

$json = json_decode(file_get_contents($file));
foreach ($json->queue as $queue) {
    ?>
	<tr>
		<td><?php 
    echo $queue;
    ?>
</td>
		<td>
			<?php 
    if (msg_queue_exists($queue)) {
        echo "ON";
    } else {
        echo "OFF";
    }
    ?>
		</td>
		<td>
			<?php 
    if (msg_queue_exists($queue)) {
        $openqueue = msg_get_queue($queue);
        echo msg_stat_queue($openqueue)['msg_qnum'];
    } else {
        echo "---";
    }
    ?>
			
		</td>
	</tr>
<?php 
}

 public function count($queueName)
 {
     $stats = msg_stat_queue($this->getQueue($queueName));
     return $stats['msg_qnum'];
 }

 public function non_blocking_in($msg)
 {
     if ($this->closed || !msg_queue_exists($this->key)) {
         return self::CLOSED;
     }
     $shm = new Message();
     $shm->store($msg);
     $error = 0;
     @msg_send($this->ipc, 1, $shm->key(), false, false, $error);
     if (MSG_EAGAIN === $error) {
         $shmAbortedMessage = new Message($shm->key());
         $shmAbortedMessage->destroy();
         return false;
     }
     ++$this->msg_count;
     $first_loop = true;
     do {
         $data = msg_stat_queue($this->ipc);
         if (!$first_loop && 0 == $data['msg_qnum']) {
             break;
         }
         $first_loop = false;
     } while (true);
     return true;
 }

    exit(1);
}
$queue = msg_get_queue($token);
var_dump(msg_queue_exists($token));
$pid = pcntl_fork();
if ($pid == 0) {
    $q = msg_get_queue($token);
    msg_send($q, 2, "start");
    msg_receive($q, 1, $type, 100, $msg);
    msg_send($q, 2, $msg);
    // echo
    exit(0);
}
msg_receive($queue, 2, $type, 100, $msg);
var_dump($msg);
msg_send($queue, 1, "ok");
msg_receive($queue, 2, $type, 100, $msg);
var_dump($msg);
$ret = @msg_send($queue, 0, 'msg', false, false, $s_error_code);
var_dump($ret);
var_dump(22 === $s_error_code);
// 22 - invalid argument
$ret = msg_receive($queue, 0, $type, 100, $msg, false, MSG_IPC_NOWAIT, $r_error_code);
var_dump($ret);
var_dump(MSG_ENOMSG === $r_error_code);
$ret = msg_stat_queue($queue);
var_dump($ret[$s_msg_qnum]);
msg_set_queue($queue, array("msg_perm.mode" => 0666));
msg_remove_queue($queue);
pcntl_waitpid($pid, $status);
unlink($filename);

// place two messages on the queue
if (!msg_send($key_t, 1, 'This is message #1', true, true, $msg_err)) {
    echo "Msg not sent because {$msg_err}\n";
}
if (!msg_send($key_t, 1, 'This is message #2 ', true, true, $msg_err)) {
    echo "Msg not sent because {$msg_err}\n";
}
// lets look at the queue structure 'msg_qnum' is really what we want to see
// it should be '2'
print_r(msg_stat_queue($key_t));
// get rid of the queue
//msg_remove_queue ($key_t);
sleep(1000);
// recieve-msg.php
if (msg_receive($key_t, 1, $msg_type, 16384, $msg, true, 0, $msg_error)) {
    echo "{$msg}\n";
    // prints 'This is message #1'
} else {
    echo "Received {$msg_error} fetching message\n";
}
// look at the structure again, ms_qnum should be '1'
print_r(msg_stat_queue($key_t));
if (msg_receive($key_t, 2, $msg_type, 16384, $msg, true, 0, $msg_error)) {
    echo "{$msg}\n";
    // prints 'This is message #2'
} else {
    echo "Received {$msg_error} fetching message\n";
}
// look at the structure again, ms_qnum should be '0', no more messages on the queue
print_r(msg_stat_queue($key_t));

 function stat()
 {
     $this->init();
     return msg_stat_queue($this->seg);
 }

 public function getMeta()
 {
     return msg_stat_queue(msg_get_queue($this->getOption('key')));
 }

$message_queue_key = ftok(__FILE__, 'a');
$message_queue = msg_get_queue($message_queue_key, 0666);
$pids = array();
for ($i = 0; $i < 5; $i++) {
    // 创建子进程
    $pids[$i] = pcntl_fork();
    if ($pids[$i]) {
        // 父进程会得到子进程号，所以这里是父进程执行的逻辑
        echo "No.{$i} child process was created, the pid is {$pids[$i]}\r\n";
        pcntl_wait($status);
        // 等待子进程中断，防止子进程成为僵尸进程
    } else {
        if ($pids[$i] == 0) {
            // 子进程得到的$pid为0, 所以这里是子进程执行的逻辑
            $pid = posix_getpid();
            echo "process.{$pid} is writing now\r\n";
            msg_send($message_queue, 1, "this is process.{$pid}'s data\r\n");
            posix_kill($pid, SIGTERM);
        }
    }
}
do {
    msg_receive($message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT);
    echo "message = {$message} \n";
    $mq_st = msg_stat_queue($message_queue);
    if ($mq_st['msg_qnum'] == 0) {
        // 队列为空，则退出
        break;
    }
} while (true);
echo "end... \n";