linux中编写自己的并发队列类(Queue 并发阻塞队列)

所属分类: 脚本专栏 / linux shell 阅读数: 727
收藏 0 赞 0 分享

设计并发队列

复制代码 代码如下:

#include <pthread.h>
#include <list>
using namespace std;

template <typename T>
class Queue
{
public:
    Queue( )
    {
        pthread_mutex_init(&_lock, NULL);
    }
    ~Queue( )
    {
        pthread_mutex_destroy(&_lock);
    }
    void push(const T& data);
    T pop( );
private:
    list<T> _list;
    pthread_mutex_t _lock;
};

template <typename T>
void Queue<T>::push(const T& value )
{
    pthread_mutex_lock(&_lock);
    _list.push_back(value);
    pthread_mutex_unlock(&_lock);
}

template <typename T>
T Queue<T>::pop( )
{
    if (_list.empty( ))
    {
        throw "element not found";
    }
    pthread_mutex_lock(&_lock);
    T _temp = _list.front( );
    _list.pop_front( );
    pthread_mutex_unlock(&_lock);
    return _temp;
}

上述代码是有效的。但是,请考虑这样的情况:您有一个很长的队列(可能包含超过 100,000 个元素),而且在代码执行期间的某个时候,从队列中读取数据的线程远远多于添加数据的线程。因为添加和取出数据操作使用相同的互斥锁,所以读取数据的速度会影响写数据的线程访问锁。那么,使用两个锁怎么样?一个锁用于读取操作,另一个用于写操作。给出修改后的 Queue 类。

复制代码 代码如下:

template <typename T>
class Queue
{
public:
    Queue( )
    {
        pthread_mutex_init(&_rlock, NULL);
        pthread_mutex_init(&_wlock, NULL);
    }
    ~Queue( )
    {
        pthread_mutex_destroy(&_rlock);
        pthread_mutex_destroy(&_wlock);
    }
    void push(const T& data);
    T pop( );
private:
    list<T> _list;
    pthread_mutex_t _rlock, _wlock;
};


template <typename T>
void Queue<T>::push(const T& value )
{
    pthread_mutex_lock(&_wlock);
    _list.push_back(value);
    pthread_mutex_unlock(&_wlock);
}

template <typename T>
T Queue<T>::pop( )
{
    if (_list.empty( ))
    {
        throw "element not found";
    }
    pthread_mutex_lock(&_rlock);
    T _temp = _list.front( );
    _list.pop_front( );
    pthread_mutex_unlock(&_rlock);
    return _temp;
}

设计并发阻塞队列

目前,如果读线程试图从没有数据的队列读取数据,仅仅会抛出异常并继续执行。但是,这种做法不总是我们想要的,读线程很可能希望等待(即阻塞自身),直到有数据可用时为止。这种队列称为阻塞的队列。如何让读线程在发现队列是空的之后等待?一种做法是定期轮询队列。但是,因为这种做法不保证队列中有数据可用,它可能会导致浪费大量 CPU 周期。推荐的方法是使用条件变量,即 pthread_cond_t 类型的变量。

复制代码 代码如下:

template <typename T>
class BlockingQueue
{
public:
    BlockingQueue ( )
    {
        pthread_mutexattr_init(&_attr);
        // set lock recursive
        pthread_mutexattr_settype(&_attr,PTHREAD_MUTEX_RECURSIVE_NP);
        pthread_mutex_init(&_lock,&_attr);
        pthread_cond_init(&_cond, NULL);
    }
    ~BlockingQueue ( )
    {
        pthread_mutex_destroy(&_lock);
        pthread_cond_destroy(&_cond);
    }
    void push(const T& data);
    bool push(const T& data, const int seconds); //time-out push
    T pop( );
    T pop(const int seconds); // time-out pop

private:
    list<T> _list;
    pthread_mutex_t _lock;
    pthread_mutexattr_t _attr;
    pthread_cond_t _cond;
};

template <typename T>
T BlockingQueue<T>::pop( )
{
    pthread_mutex_lock(&_lock);
    while (_list.empty( ))
    {
        pthread_cond_wait(&_cond, &_lock) ;
    }
    T _temp = _list.front( );
    _list.pop_front( );
    pthread_mutex_unlock(&_lock);
    return _temp;
}

template <typename T>
void BlockingQueue <T>::push(const T& value )
{
    pthread_mutex_lock(&_lock);
    const bool was_empty = _list.empty( );
    _list.push_back(value);
    pthread_mutex_unlock(&_lock);
    if (was_empty)
        pthread_cond_broadcast(&_cond);
}

并发阻塞队列设计有两个要注意的方面:

1.可以不使用 pthread_cond_broadcast,而是使用 pthread_cond_signal。但是,pthread_cond_signal 会释放至少一个等待条件变量的线程,这个线程不一定是等待时间最长的读线程。尽管使用 pthread_cond_signal 不会损害阻塞队列的功能,但是这可能会导致某些读线程的等待时间过长。

2.可能会出现虚假的线程唤醒。因此,在唤醒读线程之后,要确认列表非空,然后再继续处理。强烈建议使用基于 while 循环的 pop()。

设计有超时限制的并发阻塞队列

在许多系统中,如果无法在特定的时间段内处理新数据,就根本不处理数据了。例如,新闻频道的自动收报机显示来自金融交易所的实时股票行情,它每 n 秒收到一次新数据。如果在 n 秒内无法处理以前的一些数据,就应该丢弃这些数据并显示最新的信息。根据这个概念,我们来看看如何给并发队列的添加和取出操作增加超时限制。这意味着,如果系统无法在指定的时间限制内执行添加和取出操作,就应该根本不执行操作。

复制代码 代码如下:

template <typename T>
bool BlockingQueue <T>::push(const T& data, const int seconds)
{
    struct timespec ts1, ts2;
    const bool was_empty = _list.empty( );
    clock_gettime(CLOCK_REALTIME, &ts1);
    pthread_mutex_lock(&_lock);
    clock_gettime(CLOCK_REALTIME, &ts2);
    if ((ts2.tv_sec – ts1.tv_sec) <seconds)
    {
        was_empty = _list.empty( );
        _list.push_back(value);
    }
    pthread_mutex_unlock(&_lock);
    if (was_empty)
        pthread_cond_broadcast(&_cond);
}

template <typename T>
T BlockingQueue <T>::pop(const int seconds)
{
    struct timespec ts1, ts2;
    clock_gettime(CLOCK_REALTIME, &ts1);
    pthread_mutex_lock(&_lock);
    clock_gettime(CLOCK_REALTIME, &ts2);

    // First Check: if time out when get the _lock
    if ((ts1.tv_sec – ts2.tv_sec) < seconds)
    {
        ts2.tv_sec += seconds; // specify wake up time
        while(_list.empty( ) && (result == 0))
        {
            result = pthread_cond_timedwait(&_cond, &_lock, &ts2) ;
        }
        if (result == 0) // Second Check: if time out when timedwait 
        {
            T _temp = _list.front( );
            _list.pop_front( );
            pthread_mutex_unlock(&_lock);
            return _temp;
        }
    }
    pthread_mutex_unlock(&lock);
    throw "timeout happened";
}

设计有大小限制的并发阻塞队列

最后,讨论有大小限制的并发阻塞队列。这种队列与并发阻塞队列相似,但是对队列的大小有限制。在许多内存有限的嵌入式系统中,确实需要有大小限制的队列。
对于阻塞队列,只有读线程需要在队列中没有数据时等待。对于有大小限制的阻塞队列,如果队列满了,写线程也需要等待。

复制代码 代码如下:

template <typename T>
class BoundedBlockingQueue
{
public:
    BoundedBlockingQueue (int size) : maxSize(size)
    {
        pthread_mutex_init(&_lock, NULL);
        pthread_cond_init(&_rcond, NULL);
        pthread_cond_init(&_wcond, NULL);
        _array.reserve(maxSize);
    }
    ~BoundedBlockingQueue ( )
    {
        pthread_mutex_destroy(&_lock);
        pthread_cond_destroy(&_rcond);
        pthread_cond_destroy(&_wcond);
    }
    void push(const T& data);
    T pop( );
private:
    vector<T> _array; // or T* _array if you so prefer
    int maxSize;
    pthread_mutex_t _lock;
    pthread_cond_t _rcond, _wcond;
};

template <typename T>
void BoundedBlockingQueue <T>::push(const T& value )
{
    pthread_mutex_lock(&_lock);
    const bool was_empty = _array.empty( );
    while (_array.size( ) == maxSize)
    {
        pthread_cond_wait(&_wcond, &_lock);
    }
    _array.push_back(value);
    pthread_mutex_unlock(&_lock);
    if (was_empty)
        pthread_cond_broadcast(&_rcond);
}

template <typename T>
T BoundedBlockingQueue<T>::pop( )
{
    pthread_mutex_lock(&_lock);
    const bool was_full = (_array.size( ) == maxSize);
    while(_array.empty( ))
    {
        pthread_cond_wait(&_rcond, &_lock) ;
    }
    T _temp = _array.front( );
    _array.erase( _array.begin( ));
    pthread_mutex_unlock(&_lock);
    if (was_full)
        pthread_cond_broadcast(&_wcond);
    return _temp;
}

要注意的第一点是,这个阻塞队列有两个条件变量而不是一个。如果队列满了,写线程等待 _wcond 条件变量;读线程在从队列中取出数据之后需要通知所有线程。同样,如果队列是空的,读线程等待 _rcond 变量,写线程在把数据插入队列中之后向所有线程发送广播消息。如果在发送广播通知时没有线程在等待 _wcond 或 _rcond,会发生什么?什么也不会发生;系统会忽略这些消息。还要注意,两个条件变量使用相同的互斥锁。

 

更多精彩内容其他人还在看

Linux shell中的printf的详细用法

这篇文章主要介绍了Linux shell中的printf的详细用法的相关资料,希望能通过本文能帮助到大家,需要的朋友可以参考下
收藏 0 赞 0 分享

shell字符截取命令之cut命令的实例详解

这篇文章主要介绍了shell字符截取命令之cut命令的实例详解的相关资料,希望通过本文大家能够掌握这部分内容,需要的朋友可以参考下
收藏 0 赞 0 分享

Linux 在Bash脚本中怎么关闭文件描述符的实例

这篇文章主要介绍了Linux 在Bash脚本中怎么关闭文件描述符的实例的相关资料,希望通过本文能帮助到大家实现这样的功能,需要的朋友可以参考下
收藏 0 赞 0 分享

Linux使用文本编辑器vi常用命令

vi就是一种功能强大的文本编辑器,而vim则是高级版的vi,不但可以用不同颜色显示文字内容,还能进行诸如shell脚本、C语言程序编辑等功能,可以作为程序编辑器。下面通过本文给大家介绍linux 文本编辑器vi常用命令,一起看看吧
收藏 0 赞 0 分享

Linux 中unzip解压时中文乱码的解决办法

这篇文章主要介绍了Linux 中unzip解压时中文乱码的解决办法的相关资料,这里提供两种解决办法,希望能通过本文帮助到大家,需要的朋友可以参考下
收藏 0 赞 0 分享

Linux shell ftp命令根据文件日期下载文件的方法

最近做项目遇到这样的需求要求ftp获取远程数据的文件,根据文件的创建时间点下载文件。下面小编给大家分享知识点小结,感兴趣的朋友要求看看吧
收藏 0 赞 0 分享

Linux下服务器重启的脚本命令

Linux关闭和重启系统一般使用相同的命令可以实现。下面脚本之家小编给大家带来了Linux下服务器重启的脚本命令,感兴趣的朋友一起看看吧
收藏 0 赞 0 分享

Linux中的特殊符号与正则表达式

这篇文章主要介绍了Linux中的特殊符号与正则表达式,需要的朋友可以参考下
收藏 0 赞 0 分享

利用shell命令统计日志的方法详解

这篇文章主要给大家介绍了关于利用shell命令统计日志的方法,通过这个命令将会对大家的学习或者工作具有一定的参考学习价值,文中给出了详细的示例代码,需要的朋友可以参考借鉴,下面随着小编来一起学习学习吧。
收藏 0 赞 0 分享

Linux中的Syslog命令

syslog是Linux系统默认的日志守护进程,默认的syslog配置文件是/etc/syslog.conf文件。接下来通过本文给大家分享Linux中的Syslog命令,感兴趣的朋友一起看看吧
收藏 0 赞 0 分享
查看更多