概括

linux既支持POSIX标准的消息队列，也支持System V的消息队列。
先看几个基本的概念，这些概念在Syste V通信方式中的信号量和共享内存当中同样也是适用的。

标识符（id）

每个System V的进程通信机制中的对象都和唯一的一个引用标识符相联系，如果进程要访问此IPC对象，则需要在系统中传递这个唯一的引用标识符。具有一定权限的任何进程都可以往给定的队列当中放置一个消息，同样，具有一定权限的进程都可以从消息队列当中读取一个消息。

关键字（key）

是用来定位System V中IPC机制的对象的引用标识符的。关键字的类型为key_t，是系统中预先定义好的，它在头文件当中定义，是一个32位的整数。函数ftok()就是用来产生关键字的，它把一个已存在的路径名和一个整数标识符转换成一个ket_t值，称为IPC键。其实，这个关键字的作用就是不同进程根据它来创建IPC的标识符的。

System V查看

方法一

[root@VM-0-13-centos ~]# ipcs

------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status

------ Semaphore Arrays --------
key        semid      owner      perms      nsems

方法二

[root@VM-0-13-centos sysvipc]# pwd
/proc/sysvipc
[root@VM-0-13-centos sysvipc]# ls
msg  sem  shm
[root@VM-0-13-centos sysvipc]# cat msg
       key      msqid perms      cbytes       qnum lspid lrpid   uid   gid  cuid  cgid      stime      rtime      ctime

上PHP代码

消息队列的进程通信方式PHP只封装了System V的方式

<?php
$key = ftok("system_queue.php", 'x');
//创建队列
$msg_id = msg_get_queue($key);
fprintf(STDOUT, "队列 key = %s ID = %d \n", $key, $msg_id);
//发送消息
msg_send($msg_id, 1, "hello-" . time());
//去除消息
msg_receive($msg_id, 1, $msg_type, 1024, $msg);
//删除队列
msg_remove_queue($msg_id);

不熟悉函数的小伙伴可以查看这篇文章《进程通信函数》。

底层函数解析

msg_get_queue 函数底层调用 msgget 实现，先获取队列没有获取到传入参数进行创建。

PHP_FUNCTION(msg_get_queue)
{
    zend_long key;
    zend_long perms = 0666;
    sysvmsg_queue_t *mq;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "l|l", &key, &perms) == FAILURE)    {
        RETURN_THROWS();
    }

    object_init_ex(return_value, sysvmsg_queue_ce);
    mq = Z_SYSVMSG_QUEUE_P(return_value);

    mq->key = key;
    mq->id = msgget(key, 0);
    if (mq->id < 0)    {
        /* doesn't already exist; create it */
        mq->id = msgget(key, IPC_CREAT | IPC_EXCL | perms);
        if (mq->id < 0)    {
            php_error_docref(NULL, E_WARNING, "Failed for key 0x" ZEND_XLONG_FMT ": %s", key, strerror(errno));
            zval_ptr_dtor(return_value);
            RETURN_FALSE;
        }
    }
}

msg_send 函数底层调用 msgsnd 实现，涉及到对数据的序列化和长度计算比较复杂。

PHP_FUNCTION(msg_send)
{
    zval *message, *queue, *zerror=NULL;
    zend_long msgtype;
    bool do_serialize = 1, blocking = 1;
    sysvmsg_queue_t * mq = NULL;
    struct php_msgbuf * messagebuffer = NULL; /* buffer to transmit */
    int result;
    size_t message_len = 0;

    RETVAL_FALSE;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "Olz|bbz",
                &queue, sysvmsg_queue_ce, &msgtype, &message, &do_serialize, &blocking, &zerror) == FAILURE) {
        RETURN_THROWS();
    }

    mq = Z_SYSVMSG_QUEUE_P(queue);

    if (do_serialize) {
        smart_str msg_var = {0};
        php_serialize_data_t var_hash;

        PHP_VAR_SERIALIZE_INIT(var_hash);
        php_var_serialize(&msg_var, message, &var_hash);
        PHP_VAR_SERIALIZE_DESTROY(var_hash);

        /* NB: php_msgbuf is 1 char bigger than a long, so there is no need to
         * allocate the extra byte. */
        messagebuffer = safe_emalloc(ZSTR_LEN(msg_var.s), 1, sizeof(struct php_msgbuf));
        memcpy(messagebuffer->mtext, ZSTR_VAL(msg_var.s), ZSTR_LEN(msg_var.s) + 1);
        message_len = ZSTR_LEN(msg_var.s);
        smart_str_free(&msg_var);
    } else {
        char *p;
        switch (Z_TYPE_P(message)) {
            case IS_STRING:
                p = Z_STRVAL_P(message);
                message_len = Z_STRLEN_P(message);
                break;
            case IS_LONG:
                message_len = spprintf(&p, 0, ZEND_LONG_FMT, Z_LVAL_P(message));
                break;
            case IS_FALSE:
                message_len = spprintf(&p, 0, "0");
                break;
            case IS_TRUE:
                message_len = spprintf(&p, 0, "1");
                break;
            case IS_DOUBLE:
                message_len = spprintf(&p, 0, "%F", Z_DVAL_P(message));
                break;

            default:
                zend_argument_type_error(3, "must be of type string|int|float|bool, %s given", zend_zval_type_name(message));
                RETURN_THROWS();
        }

        messagebuffer = safe_emalloc(message_len, 1, sizeof(struct php_msgbuf));
        memcpy(messagebuffer->mtext, p, message_len + 1);

        if (Z_TYPE_P(message) != IS_STRING) {
            efree(p);
        }
    }

    /* set the message type */
    messagebuffer->mtype = msgtype;

    result = msgsnd(mq->id, messagebuffer, message_len, blocking ? 0 : IPC_NOWAIT);

    efree(messagebuffer);

    if (result == -1) {
        php_error_docref(NULL, E_WARNING, "msgsnd failed: %s", strerror(errno));
        if (zerror) {
            ZEND_TRY_ASSIGN_REF_LONG(zerror, errno);
        }
    } else {
        RETVAL_TRUE;
    }
}

msg_receive 函数底层调用 msgrcv 实现。

PHP_FUNCTION(msg_receive)
{
    zval *out_message, *queue, *out_msgtype, *zerrcode = NULL;
    zend_long desiredmsgtype, maxsize, flags = 0;
    zend_long realflags = 0;
    bool do_unserialize = 1;
    sysvmsg_queue_t *mq = NULL;
    struct php_msgbuf *messagebuffer = NULL; /* buffer to transmit */
    int result;

    RETVAL_FALSE;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "Olzlz|blz",
                &queue, sysvmsg_queue_ce, &desiredmsgtype, &out_msgtype, &maxsize,
                &out_message, &do_unserialize, &flags, &zerrcode) == FAILURE) {
        RETURN_THROWS();
    }

    if (maxsize <= 0) {
        zend_argument_value_error(4, "must be greater than 0");
        RETURN_THROWS();
    }

    if (flags != 0) {
        if (flags & PHP_MSG_EXCEPT) {
#ifndef MSG_EXCEPT
            php_error_docref(NULL, E_WARNING, "MSG_EXCEPT is not supported on your system");
            RETURN_FALSE;
#else
            realflags |= MSG_EXCEPT;
#endif
        }
        if (flags & PHP_MSG_NOERROR) {
            realflags |= MSG_NOERROR;
        }
        if (flags & PHP_MSG_IPC_NOWAIT) {
            realflags |= IPC_NOWAIT;
        }
    }

    mq = Z_SYSVMSG_QUEUE_P(queue);

    messagebuffer = (struct php_msgbuf *) safe_emalloc(maxsize, 1, sizeof(struct php_msgbuf));

    result = msgrcv(mq->id, messagebuffer, maxsize, desiredmsgtype, realflags);

    if (result >= 0) {
        /* got it! */
        ZEND_TRY_ASSIGN_REF_LONG(out_msgtype, messagebuffer->mtype);
        if (zerrcode) {
            ZEND_TRY_ASSIGN_REF_LONG(zerrcode, 0);
        }

        RETVAL_TRUE;
        if (do_unserialize)    {
            php_unserialize_data_t var_hash;
            zval tmp;
            const unsigned char *p = (const unsigned char *) messagebuffer->mtext;

            PHP_VAR_UNSERIALIZE_INIT(var_hash);
            if (!php_var_unserialize(&tmp, &p, p + result, &var_hash)) {
                php_error_docref(NULL, E_WARNING, "Message corrupted");
                ZEND_TRY_ASSIGN_REF_FALSE(out_message);
                RETVAL_FALSE;
            } else {
                ZEND_TRY_ASSIGN_REF_TMP(out_message, &tmp);
            }
            PHP_VAR_UNSERIALIZE_DESTROY(var_hash);
        } else {
            ZEND_TRY_ASSIGN_REF_STRINGL(out_message, messagebuffer->mtext, result);
        }
    } else {
        ZEND_TRY_ASSIGN_REF_LONG(out_msgtype, 0);
        ZEND_TRY_ASSIGN_REF_FALSE(out_message);
        if (zerrcode) {
            ZEND_TRY_ASSIGN_REF_LONG(zerrcode, errno);
        }
    }
    efree(messagebuffer);
}

msg_remove_queue 函数底层调用 msgctl 实现。

PHP_FUNCTION(msg_remove_queue)
{
    zval *queue;
    sysvmsg_queue_t *mq = NULL;

    if (zend_parse_parameters(ZEND_NUM_ARGS(), "O", &queue, sysvmsg_queue_ce) == FAILURE) {
        RETURN_THROWS();
    }

    mq = Z_SYSVMSG_QUEUE_P(queue);

    if (msgctl(mq->id, IPC_RMID, NULL) == 0) {
        RETVAL_TRUE;
    } else {
        RETVAL_FALSE;
    }
}