怎样在C++中处理并发队列_无锁数据结构

#include <iostream>
#include <thread>
#include <atomic>

template <typename T>
class LockFreeQueue {
private:
    T* buffer;
    int capacity;
    std::atomic<int> head; // 生产者写入位置
    std::atomic<int> tail; // 消费者读取位置

public:
    LockFreeQueue(int capacity) : capacity(capacity), head(0), tail(0) {
        buffer = new T[capacity];
    }

    ~LockFreeQueue() {
        delete[] buffer;
    }

    bool enqueue(const T& item) {
        int current_head = head.load(std::memory_order_relaxed);
        int next_head = (current_head + 1) % capacity;

        if (next_head == tail.load(std::memory_order_acquire)) { // 队列满
            return false;
        }

        buffer[current_head] = item;
        head.store(next_head, std::memory_order_release);
        return true;
    }

    bool dequeue(T& item) {
        int current_tail = tail.load(std::memory_order_relaxed);
        if (current_tail == head.load(std::memory_order_acquire)) { // 队列空
            return false;
        }

        item = buffer[current_tail];
        int next_tail = (current_tail + 1) % capacity;
        tail.store(next_tail, std::memory_order_release);
        return true;
    }
};

int main() {
    LockFreeQueue<int> queue(10);
    std::thread producer([&]() {
        for (int i = 0; i < 20; ++i) {
            while (!queue.enqueue(i));
            std::cout << "Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    std::thread consumer([&]() {
        for (int i = 0; i < 20; ++i) {
            int item;
            while (!queue.dequeue(item));
            std::cout << "Dequeued: " << item << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    });

    producer.join();
    consumer.join();

    return 0;
}

#include <iostream>
#include <thread>
#include <atomic>

template <typename T>
struct Node {
    T data;
    Node<T>* next;
};

template <typename T>
class LockFreeQueue {
private:
    std::atomic<Node<T>*> head;
    std::atomic<Node<T>*> tail;

public:
    LockFreeQueue() {
        Node<T>* dummy = new Node<T>;
        head.store(dummy);
        tail.store(dummy);
    }

    ~LockFreeQueue() {
        Node<T>* current = head.load();
        while (current != nullptr) {
            Node<T>* next = current->next;
            delete current;
            current = next;
        }
    }

    void enqueue(const T& value) {
        Node<T>* newNode = new Node<T>{value, nullptr};
        Node<T>* tailNode;

        while (true) {
            tailNode = tail.load();
            Node<T>* nextNode = tailNode->next;

            if (tailNode == tail.load()) { // 检查 tail 是否被其他线程修改
                if (nextNode == nullptr) {
                    if (tailNode->next.compare_exchange_weak(nextNode, newNode)) {
                        tail.compare_exchange_weak(tailNode, newNode); // 更新 tail，允许失败
                        return;
                    }
                } else {
                    tail.compare_exchange_weak(tailNode, nextNode); // 帮助其他线程完成 tail 的更新
                }
            }
        }
    }

    bool dequeue(T& value) {
        Node<T>* headNode;
        Node<T>* nextNode;

        while (true) {
            headNode = head.load();
            nextNode = headNode->next;

            if (headNode == head.load()) { // 检查 head 是否被其他线程修改
                if (nextNode == nullptr) {
                    return false; // 队列为空
                }

                if (head.compare_exchange_weak(headNode, nextNode)) {
                    value = nextNode->data;
                    delete headNode;
                    return true;
                }
            }
        }
    }
};

int main() {
    LockFreeQueue<int> queue;

    std::thread producer1([&]() {
        for (int i = 0; i < 10; ++i) {
            queue.enqueue(i);
            std::cout << "Producer 1 Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

     std::thread producer2([&]() {
        for (int i = 10; i < 20; ++i) {
            queue.enqueue(i);
            std::cout << "Producer 2 Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    std::thread consumer1([&]() {
        for (int i = 0; i < 10; ++i) {
            int item;
            if(queue.dequeue(item)){
              std::cout << "Consumer 1 Dequeued: " << item << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    });

    std::thread consumer2([&]() {
        for (int i = 0; i < 10; ++i) {
            int item;
             if(queue.dequeue(item)){
              std::cout << "Consumer 2 Dequeued: " << item << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    });

    producer1.join();
    producer2.join();
    consumer1.join();
    consumer2.join();

    return 0;
}