C++11 —— 单生产者/单消费者 的 FIFO 无锁队列

??发现 zeromq 的 yqueue_t 模板类，其数据存储理念设计得非常妙。借这一理念，按照 STL 的泛型类 queue 的接口标准，我设计了一个线程安全的 单生产者/单消费者（单线程push/单线程pop） FIFO 队列，以此满足更为广泛的应用。

1. 数据存储理念的结构图

  • 队列的整体结构上，使用链表的方式，将多个固定长度的 chunk 串联起来；
  • 每个 chunk 则可用于存储队列所需要的元素；
  • 增加一个可交换的 chunk 单元，利于内存复用；
  • 队列使用时，支持 单个线程的 push（生产） 和 单个线程 pop（消费）的并发操作（内部并未加锁）。

2. 源码 （xspsc_queue.h）

/**
 * The MIT License (MIT)
 * Copyright (c) 2019, Gaaagaa All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
 * of the Software, and to permit persons to whom the Software is furnished to do
 * so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

/**
 * @file xspsc_queue.h
 * Copyright (c) 2019, Gaaagaa All rights reserved.
 *
 * @author  ：Gaaagaa
 * @date    : 2019-11-29
 * @version : 1.0.0.0
 * @brief   : 实现双线程安全的 单生产者/单消费者 FIFO 队列。
 */

#ifndef __XSPSC_QUEUE_H__
#define __XSPSC_QUEUE_H__

#include <memory>
#include <atomic>
#include <cassert>

////////////////////////////////////////////////////////////////////////////////
// x_spsc_queue_t : single producer/single consumer FIFO queue

/**
 * @class x_spsc_queue_t
 * @brief 双线程安全的 单生产者/单消费者 FIFO队列。
 *
 * @param [in ] __object_t    : 队列存储的对象类型。
 * @param [in ] __chunk_size  : 队列中的存储块可容纳对象的数量。
 * @param [in ] __allocator_t : 对象分配器。
 */
template< typename __object_t,
          size_t   __chunk_size,
          typename __allocator_t = std::allocator< __object_t > >
class x_spsc_queue_t : protected __allocator_t
{
    static_assert(__chunk_size >= 4,
                  "__chunk_size size value must be greater than or equal to 4!");

    // common data types
public:
    typedef __object_t value_type;
    using x_object_t = __object_t;

private:
    /**
     * @struct x_chunk_t
     * @brief  存储对象节点的连续内存块结构体。
     */
    typedef struct x_chunk_t
    {
        x_chunk_t  * xnext_ptr;   ///< 指向后一内存块节点
        x_object_t * xot_array;   ///< 当前内存块中的对象节点数组
    } x_chunk_t;

#ifdef _MSC_VER
    using ssize_t = std::intptr_t;
#endif // _MSC_VER

    using x_chunk_ptr_t   = x_chunk_t *;
    using x_atomic_ptr_t  = std::atomic< x_chunk_ptr_t >;
    using x_atomic_size_t = std::atomic< size_t >;
    using x_allocator_t   = __allocator_t;
    using x_chunk_alloc_t = typename std::allocator_traits<
                                        x_allocator_t >::template
                                            rebind_alloc< x_chunk_t >;

    // constructor/destructor
public:
    explicit x_spsc_queue_t(void)
        : m_chk_front(nullptr)
        , m_pos_front(0)
        , m_chk_back(nullptr)
        , m_pos_back(0)
        , m_xst_size(0)
        , m_chk_stored(nullptr)
    {
        m_chk_front = m_chk_back = alloc_chunk();
    }

    ~x_spsc_queue_t(void)
    {
        while (size() > 0)
            pop();

        assert(m_chk_front == m_chk_back);
        free_chunk(m_chk_front);

        free_chunk(m_chk_stored.exchange(nullptr));

        m_chk_front = nullptr;
        m_pos_front = 0;
        m_chk_back  = nullptr;
        m_pos_back  = 0;
    }

    x_spsc_queue_t(x_spsc_queue_t && xobject) = delete;
    x_spsc_queue_t & operator = (x_spsc_queue_t && xobject) = delete;
    x_spsc_queue_t(const x_spsc_queue_t & xobject) = delete;
    x_spsc_queue_t & operator = (const x_spsc_queue_t & xobject) = delete;

    // public interfaces
public:
    /**********************************************************/
    /**
     * @brief 当前队列中的对象数量。
     */
    inline size_t size(void) const
    {
        return m_xst_size;
    }

    /**********************************************************/
    /**
     * @brief 判断队列是否为空。
     */
    inline bool empty(void) const
    {
        return (0 == size());
    }

    /**********************************************************/
    /**
     * @brief 向队列后端压入一个对象。
     */
    void push(const x_object_t & xobject)
    {
        move_back_pos();
        x_allocator_t::construct(
            &m_chk_back->xot_array[m_pos_back], xobject);
        m_xst_size.fetch_add(1);
    }

    /**********************************************************/
    /**
     * @brief 向队列后端压入一个对象。
     */
    void push(x_object_t && xobject)
    {
        move_back_pos();
        x_allocator_t::construct(
            &m_chk_back->xot_array[m_pos_back],
            std::forward< x_object_t >(xobject));
        m_xst_size.fetch_add(1);
    }

    /**********************************************************/
    /**
     * @brief 从队列前端弹出一个对象。
     */
    void pop(void)
    {
        assert(!empty());
        m_xst_size.fetch_sub(1);
        x_allocator_t::destroy(&m_chk_front->xot_array[m_pos_front]);
        move_front_pos();
    }

    /**********************************************************/
    /**
     * @brief 返回队列前端对象。
     */
    inline x_object_t & front(void)
    {
        assert(!empty());
        return m_chk_front->xot_array[m_pos_front];
    }

    /**********************************************************/
    /**
     * @brief 返回队列前端对象。
     */
    inline const x_object_t & front(void) const
    {
        assert(!empty());
        return m_chk_front->xot_array[m_pos_front];
    }

    /**********************************************************/
    /**
     * @brief 返回队列后端对象。
     */
    inline x_object_t & back(void)
    {
        assert(!empty());
        return m_chk_back->xot_array[m_pos_back];
    }

    /**********************************************************/
    /**
     * @brief 返回队列后端对象。
     */
    inline const x_object_t & back(void) const
    {
        assert(!empty());
        return m_chk_back->xot_array[m_pos_back];
    }

    // internal invoking
private:
    /**********************************************************/
    /**
     * @brief 申请一个存储对象节点的内存块。
     */
    x_chunk_ptr_t alloc_chunk(void)
    {
        x_chunk_alloc_t xchunk_allocator(*(x_allocator_t *)this);

        x_chunk_ptr_t xchunk_ptr = xchunk_allocator.allocate(1);
        assert(nullptr != xchunk_ptr);

        if (nullptr != xchunk_ptr)
        {
            xchunk_ptr->xot_array = x_allocator_t::allocate(__chunk_size);
            assert(nullptr != xchunk_ptr->xot_array);

            if (nullptr != xchunk_ptr->xot_array)
            {
                xchunk_ptr->xnext_ptr = nullptr;
            }
            else
            {
                xchunk_allocator.deallocate(xchunk_ptr, 1);
                xchunk_ptr = nullptr;
            }
        }

        return xchunk_ptr;
    }

    /**********************************************************/
    /**
     * @brief 释放一个存储对象节点的内存块。
     */
    void free_chunk(x_chunk_ptr_t xchunk_ptr)
    {
        if (nullptr != xchunk_ptr)
        {
            if (nullptr != xchunk_ptr->xot_array)
            {
                x_allocator_t::deallocate(xchunk_ptr->xot_array, __chunk_size);
            }

            x_chunk_alloc_t xchunk_allocator(*(x_allocator_t *)this);
            xchunk_allocator.deallocate(xchunk_ptr, 1);
        }
    }

    /**********************************************************/
    /**
     * @brief 将前端位置向后移（该接口仅由 pop() 接口调用）。
     */
    void move_front_pos(void)
    {
        if (++m_pos_front == __chunk_size)
        {
            x_chunk_ptr_t xchunk_ptr = m_chk_front;
            m_chk_front = m_chk_front->xnext_ptr;
            assert(nullptr != m_chk_front);
            m_pos_front = 0;

            free_chunk(m_chk_stored.exchange(xchunk_ptr));
        }
    }

    /**********************************************************/
    /**
     * @brief 将后端位置向后移（该接口仅由 push() 接口调用）。
     */
    void move_back_pos(void)
    {
        if (++m_pos_back == __chunk_size)
        {
            x_chunk_ptr_t xchunk_ptr = m_chk_stored.exchange(nullptr);
            if (nullptr != xchunk_ptr)
            {
                xchunk_ptr->xnext_ptr = nullptr;
                m_chk_back->xnext_ptr = xchunk_ptr;
            }
            else
            {
                m_chk_back->xnext_ptr = alloc_chunk();
            }

            m_chk_back = m_chk_back->xnext_ptr;
            m_pos_back = 0;
        }
    }

    // data members
protected:
    x_chunk_ptr_t    m_chk_front;  ///< 内存块链表的前端块
    ssize_t          m_pos_front;  ///< 队列中的前端对象位置
    x_chunk_ptr_t    m_chk_back;   ///< 内存块链表的后端块
    ssize_t          m_pos_back;   ///< 队列中的后端对象位置
    x_atomic_size_t  m_xst_size;   ///< 队列中的有效对象数量
    x_atomic_ptr_t   m_chk_stored; ///< 用于保存临时内存块（备用缓存块）
};

////////////////////////////////////////////////////////////////////////////////

#endif // __XSPSC_QUEUE_H__

3. 使用示例


#include "xspsc_queue.h"
#include <iostream>
#include <thread>
#include <chrono>

#include <list>

////////////////////////////////////////////////////////////////////////////////

int main(int argc, char * argv[])
{
    using x_int_queue_t = x_spsc_queue_t< int, 8 >;

    x_int_queue_t spsc;

    std::cout << "sizeof(x_int_queue_t) : " << sizeof(x_int_queue_t) << std::endl;

    bool b_push_finished = false;
    std::thread xthread_in([&spsc, &b_push_finished](void) -> void
    {
        for (int i = 1; i < 10000; ++i)
        {
            spsc.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }

        b_push_finished = true;
    });

    std::thread xthread_out([&spsc, &b_push_finished](void) -> void
    {
        while (true)
        {
            if (!spsc.empty())
            {
                std::cout << spsc.size() << " : " << spsc.front() << std::endl;
                spsc.pop();
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
            else if (b_push_finished)
            {
                break;
            }
        }
    });

    if (xthread_in.joinable())
    {
        xthread_in.join();
    }

    if (xthread_out.joinable())
    {
        xthread_out.join();
    }

    return 0;
}

原文地址：https://www.cnblogs.com/Gaaagaa/p/12130787.html

时间： 2024-11-23 23:08:38

C++11 —— 单生产者/单消费者 的 FIFO 无锁队列的相关文章

单生产者/单消费者 的 FIFO 无锁队列

??发现 zeromq 的 yqueue_t 模板类,其数据存储理念设计得非常妙.借这一理念,按照 STL 的泛型类 queue 的接口标准,我设计了一个线程安全的 单生产者/单消费者(单线程push/单线程pop) FIFO 队列,以此满足更为广泛的应用. 1. 数据存储理念的结构图 队列的整体结构上,使用链表的方式,将多个固定长度的 chunk 串联起来: 每个 chunk 则可用于存储队列所需要的元素: 增加一个可交换的 chunk 单元,利于内存复用: 队列使用时,支持 单个线程的 pu

单生产者-多消费者模型中遇到的问题

(1)      原始代码 最近使用单生产者-多消费者模型是遇到一个问题,以前既然都没有想到过.生产者线程的代码如下,基本功能就是接收到一个连接之后创建一个Socket对象并放到list中等待处理. void DataManager::InternalStart() { server_socket_ = new ServerSocket(); if (!server_socket_->SetAddress(NetworkUtil::GetIpAddress().c_str(), 9091)) {

并发无锁队列学习之二【单生产者单消费者】

1.前言 最近工作比较忙,加班较多,每天晚上回到家10点多了.我不知道自己还能坚持多久,既然选择了就要做到最好.写博客的少了.总觉得少了点什么,需要继续学习.今天继续上个开篇写,介绍单生产者单消费者模型的队列.根据写入队列的内容是定长还是变长,分为单生产者单消费者定长队列和单生产者单消费者变长队列两种.单生产者单消费者模型的队列操作过程是不需要进行加锁的.生产者通过写索引控制入队操作,消费者通过读索引控制出队列操作.二者相互之间对索引是独享,不存在竞争关系.如下图所示: 2.单生产者单消费者定长

并发无锁队列学习（单生产者单消费者模型）

1.引言 本文介绍单生产者单消费者模型的队列.根据写入队列的内容是定长还是变长,分为单生产者单消费者定长队列和单生产者单消费者变长队列两种.单生产者单消费者模型的队列操作过程是不需要进行加锁的.生产者通过写索引控制入队操作,消费者通过读索引控制出队列操作.二者相互之间对索引是独享,不存在竞争关系.如下图所示: 2.单生产者单消费者定长队列 这种队列要求每次入队和出队的内容是定长的,即生产者写入队列和消费者读取队列的内容大小事相同的.linux内核中的kfifo就是这种队列,提供了读和写两个索引.

disruptor 单生产者多消费者

demo1 单生产者多消费者创建. maven 依赖 <!-- https://mvnrepository.com/artifact/com.lmax/disruptor --> <dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.4.2</version> </dependency&

基于无锁队列和c++11的高性能线程池

基于无锁队列和c++11的高性能线程池线程使用c++11库和线程池之间的消息通讯使用一个简单的无锁消息队列适用于linux平台,gcc 4.6以上 标签: <无> 代码片段(6)[全屏查看所有代码] 1. [代码]lckfree.h ? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 // lckfree.h //

C++11实现生产者消费者问题

生产者消费者问题是多线程并发中一个非常经典的问题.我在这里实现了一个基于C++11的,单生产者单消费者的版本,供大家参考. #include <windows.h> #include <iostream> #include <cstdlib> #include <mutex> #include <thread> #include <condition_variable> const int bufferSize=10; struct

多生产者多消费者问题

1单生产者单消费者 package example; class Resource{ private String name; private int num=1; private boolean flag=false; public synchronized void set(String name){ if(flag){ try { wait(); } catch (InterruptedException e) { // TODO 自动生成的 catch 块 e.printStackTra

Linux：生产者与消费者模式

生产者:生产数据 消费者:消费数据 提供场所:缓冲区,eg:超市 生产者消费者特点:三种关系,两类人,一个场所 三种关系指的是:生产者与生产者之间是互斥关系 消费者与消费者之间是互斥关系 生产者与消费者之间是同步与互斥关系 两类人:生产者,消费者 一个场所:存储数据(此处用带头单链表实现) 单生产者单消费者模式:此例取数据方式为FIFO先进先出. 利用互斥锁实现单生产者单消费者模式. #include<stdio.h> #include<malloc.h> #include<