C++线程安全的队列你了解嘛

C++线程安全的队列

什么是线程安全的队列？

线程安全的队列是可以在多个线程同时读写时保证数据一致性和正确性的队列。在多个线程同时对同一个队列进行读写操作时，若不进行同步控制，就会出现数据异常和不一致的情况。线程安全的队列就是为了解决这个问题而设计的一种数据结构。

如何设计线程安全的队列？

设计线程安全的队列主要需要解决以下两个问题：

1. 如何对队列进行同步控制：对队列的读写操作必须进行同步控制，保证线程之间的数据一致性和正确性。
2. 如何实现队列的性能优化：一个高效的队列需要具备一定的性能优化，比如延迟申请内存、预申请内存空间等。

一般而言，可以使用互斥锁或者自旋锁来对队列进行同步控制。具体实现细节可参考以下示例代码：

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <chrono>

template<typename T>
class SafeQueue
{
private:
    std::queue<T> _queue;               // 队列
    std::mutex _mutex;                  // 互斥锁

public:
    bool empty() { std::lock_guard<std::mutex> lock(_mutex); return _queue.empty(); }         // 判断队列是否为空
    int size() { std::lock_guard<std::mutex> lock(_mutex); return _queue.size(); }            // 获取队列的大小
    void push(T value) { std::lock_guard<std::mutex> lock(_mutex); _queue.push(value); }      // 将元素入队列
    T pop() { std::lock_guard<std::mutex> lock(_mutex); T value = _queue.front(); _queue.pop(); return value; }                 // 将元素出队列
};

// 测试
int main()
{
    SafeQueue<int> q;

    // 生产者
    std::thread producer([&]() {
        for (int i = 1; i <= 10; i++) {
            q.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(10));  // 休眠10毫秒
        }
    });

    // 消费者
    std::thread consumer([&]() {
        while (true) {
            if (q.empty()) continue;

            int value = q.pop();
            std::cout << "消费者消费: " << value << std::endl;

            if (value == 10) break;         // 生产者线程结束循环后，退出消费者线程
        }
    });

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

    return 0;
}

线程安全的队列的应用示例

应用示例1：多线程文件IO操作

多线程支持高效的文件IO操作是一个非常常见的应用场景，通常在一个线程中读取文件数据，然后将读取到的数据写入到线程安全的队列中，之后再在另外的线程中从队列中取出数据进行处理。以下是一个简单的多线程文件IO操作示例：

#include <iostream>
#include <fstream>
#include <thread>
#include <mutex>
#include <condition_variable>

template<typename T>
class SafeQueue
{
private:
    std::queue<T> _queue;                                   // 队列
    mutable std::mutex _mutex;                              // 互斥锁
    std::condition_variable _cond;                          // 条件变量

public:
    bool empty() const { std::lock_guard<std::mutex> lock(_mutex); return _queue.empty(); }         // 判断队列是否为空
    int size() const { std::lock_guard<std::mutex> lock(_mutex); return _queue.size(); }            // 获取队列的大小
    void push(T value) { std::lock_guard<std::mutex> lock(_mutex); _queue.push(value); _cond.notify_all(); }      // 将元素入队列
    T pop() { std::unique_lock<std::mutex> lock(_mutex); _cond.wait(lock, [=] { return !_queue.empty(); }); T value = _queue.front(); _queue.pop(); return value; }           // 将元素出队列并等待队列非空通知
};

void readFile(std::ifstream& ifs, SafeQueue<std::string>& q)
{
    std::string line;

    while (std::getline(ifs, line)) {
        q.push(line);
    }
}

void writeFile(std::ofstream& ofs, SafeQueue<std::string>& q)
{
    while (true) {
        if (q.empty()) continue;

        std::string line = q.pop();
        ofs << line << std::endl;

        if (line == "exit") break;
    }
}

int main()
{
    std::ifstream ifs("input.txt");
    std::ofstream ofs("output.txt");

    SafeQueue<std::string> q;

    // 读取文件
    std::thread reader([&]() { readFile(ifs, q); });

    // 写入文件
    std::thread writer([&]() { writeFile(ofs, q); });

    reader.join();
    writer.join();

    return 0;
}

应用示例2：多线程爬虫

多线程爬虫是另一个常见的应用场景，可以使用线程安全的队列来存储待爬取的URL，然后多个爬虫线程从队列中取出URL进行解析和爬取。以下是一个简单的多线程爬虫示例：

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <curl/curl.h>

template<typename T>
class SafeQueue
{
private:
    std::queue<T> _queue;                                   // 队列
    mutable std::mutex _mutex;                              // 互斥锁
    std::condition_variable _cond;                          // 条件变量

public:
    bool empty() const { std::lock_guard<std::mutex> lock(_mutex); return _queue.empty(); }         // 判断队列是否为空
    int size() const { std::lock_guard<std::mutex> lock(_mutex); return _queue.size(); }            // 获取队列的大小
    void push(T value) { std::lock_guard<std::mutex> lock(_mutex); _queue.push(value); _cond.notify_all(); }      // 将元素入队列
    T pop() { std::unique_lock<std::mutex> lock(_mutex); _cond.wait(lock, [=] { return !_queue.empty(); }); T value = _queue.front(); _queue.pop(); return value; }           // 将元素出队列并等待队列非空通知
};

size_t writeFunction(void* ptr, size_t size, size_t nmemb, std::string* data)
{
    data->append((char*)ptr, size * nmemb);

    return size * nmemb;
}

void worker(SafeQueue<std::string>& q)
{
    CURL* curl = curl_easy_init();
    std::string url;

    while (true) {
        if (q.empty()) continue;

        url = q.pop();

        if (url == "exit") break;

        curl_easy_setopt(curl, CURLOPT_URL, url.c_str());

        std::string response;
        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeFunction);
        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response);

        CURLcode res = curl_easy_perform(curl);

        if (res == CURLE_OK) {
            std::cout << url << ": " << response.size() << std::endl;
        } else {
            std::cerr << "HTTP error code: " << res << std::endl;
        }
    }

    curl_easy_cleanup(curl);
}

int main()
{
    SafeQueue<std::string> q;
    std::vector<std::thread> threads;

    threads.push_back(std::thread([&]() { worker(q); }));
    threads.push_back(std::thread([&]() { worker(q); }));
    threads.push_back(std::thread([&]() { worker(q); }));

    q.push("https://www.google.com/");
    q.push("https://www.baidu.com/");
    q.push("https://www.bing.com/");
    q.push("exit");

    for (auto& thread : threads) {
        thread.join();
    }

    return 0;
}

以上是线程安全队列的基本知识点和应用实例，希望对你有所帮助。