freeswitch APR-UTIL库消息队列实现

概述

freeswitch的核心源代码是基于apr库开发的，在不同的系统上有很好的移植性。

APR库在之前的文章中已经介绍过了，APR-UTIL库是和APR并列的工具库，它们都是由APACHE开源出来的跨平台可移植库，不同点在于库中实现的功能接口有区别。

在应用的开发过程中，模块内部的消息传递是经常会碰到的需求，而消息队列就是很好的解决方案。

APR-UTIL库中就提供了一套线程安全的消息队列接口。

我对几个问题比较好奇，消息队列的数据结构是什么样的？消息队列如何插入元素？消息队列如何弹出元素？如何获取最优的效率？

下面我们对apr-util库的消息队列实现做一个介绍。

环境

centos：CentOS  release 7.0 (Final)或以上版本

freeswitch：v1.8.7

GCC：4.8.5

数据结构

APR-UTIL库消息队列源代码文件：

libs\apr-util\include\apr_queue.h

libs\apr-util\misc\apr_queue.c

在apr_queue.c文件中，我们可以找到消息队列的数据结构定义

struct apr_queue_t {

    void              **data; //消息指针数组

    unsigned int        nelts;  //消息个数

    unsigned int        in;      //消息入队的数组下标

    unsigned int        out;    //消息出队的数组下标

    unsigned int        bounds;    //队列容量

    unsigned int        full_waiters;      //队列满的等待者个数

    unsigned int        empty_waiters; //队列空的等待者个数

    apr_thread_mutex_t *one_big_mutex; //多线程互斥锁

    apr_thread_cond_t  *not_empty;       //队列空转非空的条件变量

    apr_thread_cond_t  *not_full;     //队列满转非满的条件变量

    int                 terminated;      //队列终止标识

};

常用函数

apr_queue.h文件的接口函数：

apr_queue_create  //create a FIFO queue

apr_queue_push   //push/add a object to the queue, blocking if the queue is already full

apr_queue_pop     //pop/get an object from the queue, blocking if the queue is already empty

apr_queue_pop_timeout      //pop/get an object from the queue, blocking if the queue is already empty

apr_queue_trypush      //push/add a object to the queue, returning immediatly if the queue is full

apr_queue_trypop //pop/get an object to the queue, returning immediatly if the queue is empty

apr_queue_size     //returns the size of the queue.

apr_queue_interrupt_all       //interrupt all the threads blocking on this queue.

apr_queue_term    //terminate all queue, sending a interupt to all the blocking threads

apr_queue_create创建

APU_DECLARE(apr_status_t) apr_queue_create(apr_queue_t **queue,

                                           unsigned int queue_capacity,

                                           apr_pool_t *a);

接口逻辑：

1. 在内存池a中分配一块大小为apr_queue_t的内存queue。
2. 在内存池a中创建一个新的互斥量queue->one_big_mutex。
3. 在内存池a中创建两个条件变量queue->not_empty和queue->not_full。
4. 在内存池a中分配一块大小为queue_capacity * sizeof(void*)的内存queue->data。
5. 变量初始化。
6. 在内存池a注册一个清理回调函数。

消息队列创建后的内存模型如图

apr_queue_push入队

APU_DECLARE(apr_status_t) apr_queue_push(apr_queue_t *queue, void *data);

接口逻辑：

1. 判断终止标识queue->terminated。
2. 加锁queue->one_big_mutex。
3. 判断queue是满的，否则继续下一步。

a)      判断终止标识queue->terminated。

b)      队列满的等待者queue->full_waiters递增。

c)      等待条件变量queue->not_full的信号，唤醒之后继续下一步。

d)      队列满的等待者queue->full_waiters递减。

e)      再次判断queue仍然是满的，则解锁退出，返回APR_EINTR。

1. 消息入队，在queue->data[queue->in]保存消息指针。
2. 入队下标queue->in递增，超过数组长度则从0重新开始。
3. 消息个数queue->nelts递增。
4. 判断有队列空的等待者queue->empty_waiters，则给条件变量queue->not_empty发送信号。
5. 解锁queue->one_big_mutex，退出。

消息入队后的内存模型如图

apr_queue_pop出队

APU_DECLARE(apr_status_t) apr_queue_pop(apr_queue_t *queue, void **data);

接口逻辑：

1. 判断终止标识queue->terminated。
2. 加锁queue->one_big_mutex。
3. 判断queue是空的，否则继续下一步。

a)      判断终止标识queue->terminated。

b)      队列空的等待者queue->empty_waiters递增。

c)      等待条件变量queue->not_empty的信号，唤醒之后继续下一步。

d)      队列空的等待者queue->empty_waiters递减。

e)      再次判断queue仍然是空的，则解锁退出，返回APR_EINTR。

1. 消息出队，获取消息指针queue->data[queue->out]。
2. 出队下标queue->out递增，超过数组长度则从0重新开始。
3. 消息个数queue->nelts递减。
4. 判断有队列满的等待者queue->full_waiters，则给条件变量queue->not_full发送信号。
5. 解锁queue->one_big_mutex，退出。

消息出队后的内存模型如图，从图中我们可以看出消息出队仅仅是对out下标做了改动，消息内容都没有变动。

apr_queue_term终止

APU_DECLARE(apr_status_t) apr_queue_term(apr_queue_t *queue);

接口逻辑：

1. 加锁queue->one_big_mutex。
2. 设置队列终止标识queue->terminated。
3. 解锁queue->one_big_mutex。
4. 加锁queue->one_big_mutex。
5. 给条件变量queue->not_empty发送广播，唤醒所有等待的线程。
6. 给条件变量queue->not_full发送广播，唤醒所有等待的线程。
7. 解锁queue->one_big_mutex。
8. 退出。

总结

apr-util的消息队列在使用时，需要注意的地方。

消息入队，再次判断队列满时，是会丢消息的，需要调用者判断返回值为APR_EINTR并新入队。

消息出队，再次判断队列空时，也需要调用者判断返回值为APR_EINTR并重新获取消息，而不能简单的报错终止业务逻辑。

消息队列的接口中，没有销毁队列的接口，因为apr_queue_t是在内存池a中创建的，并且有注册内存池清理回调函数，所以消息队列会随着内存池a的清理自动销毁。

空空如常

求真得真