线程安全编译期数组：std::array多线程操作的安全问题及优化咨询

核心问题解答 #

虽然SafeQueue是线程安全的，std::shared_ptr的引用计数操作是原子的，但**std::array容器本身不是线程安全的**。当不同线程同时调用broadcast（遍历/读取数组元素）和addQueue（修改数组元素）时，会存在以下线程安全隐患：

• broadcast中拷贝shared_ptr的操作，与addQueue中给数组元素赋值shared_ptr的操作不是原子的，可能导致数据竞争，拿到不完整的shared_ptr。
• 遍历过程中数组元素被修改，会导致遍历到的元素状态不一致，后续调用push可能引发未定义行为。

现有代码的线程安全隐患 #

你的Broadcaster类中，m_queues是普通的std::array，没有任何同步机制：

• broadcast的范围for循环会逐个拷贝数组中的shared_ptr，如果此时addQueue正在修改某个元素，拷贝操作和赋值操作会冲突，因为shared_ptr的拷贝/赋值（除原子特化版本外）不是线程安全的。
• 遍历过程中元素被修改，可能出现遍历到的shared_ptr处于半初始化状态，调用其成员函数会导致崩溃或未定义行为。

优化实现方案 #

方案一：互斥锁（最简单可靠） #

给Broadcaster添加一个全局互斥锁，确保同一时间只有一个线程操作数组，完全避免数据竞争。

#include <array>
#include <memory>
#include <mutex>

struct SafeQueue
{
    void push(const double d) {};
    bool try_pop(double& d) {};
};

struct Broadcaster
{
    std::array<std::shared_ptr<SafeQueue>, 10> m_queues;
    std::mutex m_mutex; // 全局互斥锁

    void broadcast(const double d)
    {
        std::lock_guard<std::mutex> lock(m_mutex); // 自动加锁/解锁
        for(auto p : m_queues)
            if(p) p->push(d);
    }

    bool addQueue(const std::shared_ptr<SafeQueue>& q)
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        for(auto& p : m_queues)
            if(!p) 
            {
                p = q;
                return true;
            }
        return false;
    }
};

优点：实现简单，逻辑清晰，能覆盖所有场景，完全消除数据竞争。
缺点：高并发场景下锁竞争会带来一定性能损耗，但如果SafeQueue::push耗时不长，这个影响可以忽略。

方案二：C++20原子shared_ptr（高并发场景） #

使用std::atomic<std::shared_ptr<T>>（C++20引入）将数组元素改为原子类型，每个元素的读写都是原子操作，无需全局锁。

#include <array>
#include <memory>
#include <atomic>

struct SafeQueue
{
    void push(const double d) {};
    bool try_pop(double& d) {};
};

struct Broadcaster
{
    std::array<std::atomic<std::shared_ptr<SafeQueue>>, 10> m_queues;

    void broadcast(const double d)
    {
        for(auto& atomic_p : m_queues)
        {
            // 原子加载shared_ptr，确保拿到完整的对象
            auto p = atomic_p.load(std::memory_order_acquire);
            if(p) p->push(d);
        }
    }

    bool addQueue(const std::shared_ptr<SafeQueue>& q)
    {
        for(auto& atomic_p : m_queues)
        {
            std::shared_ptr<SafeQueue> expected;
            // 比较并交换，只有当元素为空时才赋值
            if(atomic_p.compare_exchange_strong(expected, q, 
                std::memory_order_release, std::memory_order_acquire))
            {
                return true;
            }
        }
        return false;
    }
};

优点：无全局锁，并发性能更好，适合高频率调用broadcast和addQueue的场景。
缺点：依赖C++20标准，需要熟悉内存序的使用，且遍历无法保证数组的整体一致性（但在广播场景下，发送到当前存在的队列即可，这个不一致是可接受的）。

方案三：原子索引（仅限顺序添加场景） #

如果你的场景只需要按顺序添加队列，不需要填充数组中的空位，可以用原子变量记录下一个可用索引，实现无锁逻辑。

#include <array>
#include <memory>
#include <atomic>

struct SafeQueue
{
    void push(const double d) {};
    bool try_pop(double& d) {};
};

struct Broadcaster
{
    std::array<std::shared_ptr<SafeQueue>, 10> m_queues;
    std::atomic<int> m_next_index = 0;

    void broadcast(const double d)
    {
        int current_idx = m_next_index.load(std::memory_order_acquire);
        // 只遍历已添加的队列
        for(int i = 0; i < current_idx; ++i)
        {
            auto p = m_queues[i];
            if(p) p->push(d);
        }
    }

    bool addQueue(const std::shared_ptr<SafeQueue>& q)
    {
        // 原子获取下一个索引
        int idx = m_next_index.fetch_add(1, std::memory_order_release);
        if(idx >= m_queues.size())
        {
            // 索引超出范围，回滚
            m_next_index.fetch_sub(1, std::memory_order_release);
            return false;
        }
        m_queues[idx] = q;
        return true;
    }
};

优点：完全无锁，并发性能最优。
缺点：局限性大，只能按顺序添加队列，无法填充已有的空位；如果有队列移除需求，该方案不适用。

方案选择建议 #

• 若不需要极致性能，优先选择互斥锁方案，实现简单且可靠。
• 若处于高并发场景且使用C++20及以上标准，选择原子shared_ptr方案。
• 若仅需顺序添加队列且无空位填充需求，可考虑原子索引方案。

内容的提问来源于stack exchange，提问作者SpeakX