C++多线程并发读取ifstream时getline冲突及并行化方案咨询

问题根源 #

你之前用mutex保护getline仍出错，核心原因是**std::ifstream本身不是线程安全的**——即使加锁，多个线程操作同一个流对象时，流内部的状态（如文件指针位置、错误标记）可能出现未定义的交互。锁只能保护getline调用本身，无法覆盖流状态的所有共享修改，最终导致读取错乱。

方案1：预拆分文件，线程独立处理分片 #

这是最直接的无竞争方案：先将lineitem文件按行拆分为多个独立分片，每个线程单独处理一个分片（或原文件的指定区间），最后汇总各线程的计算结果。

实现步骤 #

1. 主线程先遍历原文件，记录所有换行符的位置，将文件划分为N个包含完整行的区间（N为CPU核心数）。
2. 每个线程打开原文件，通过seekg定位到自己负责区间的起始位置，逐行读取处理，计算局部总和。
3. 所有线程完成后，累加局部总和得到最终结果。

代码示例片段 #

// 预计算所有行的起始位置（主线程执行）
std::vector<std::streampos> line_positions;
std::ifstream temp_file(this->lineitem);
std::string temp_line;
line_positions.push_back(0);
while (std::getline(temp_file, temp_line)) {
    line_positions.push_back(temp_file.tellg());
}
size_t total_lines = line_positions.size() - 1;
size_t lines_per_thread = total_lines / std::thread::hardware_concurrency();

// 启动线程处理分片
std::vector<std::thread> threads;
std::vector<long long> local_sums(std::thread::hardware_concurrency(), 0);
std::vector<int> local_counts(std::thread::hardware_concurrency(), 0);

for (size_t i = 0; i < std::thread::hardware_concurrency(); ++i) {
    size_t start_line = i * lines_per_thread;
    size_t end_line = (i == std::thread::hardware_concurrency() - 1) ? total_lines : (i+1)*lines_per_thread;
    
    threads.emplace_back([this, start_line, end_line, &line_positions, &local_sums, &local_counts, i]() {
        std::ifstream l_file(this->lineitem);
        l_file.seekg(line_positions[start_line]);
        
        std::string l_line;
        long long local_sum = 0;
        int local_n = 0;
        
        for (size_t line_idx = start_line; line_idx < end_line; ++line_idx) {
            if (!std::getline(l_file, l_line)) break;
            
            std::istringstream iss(l_line);
            std::string l_orderkey, l_quantity;
            
            std::getline(iss, l_orderkey, '|');
            for (int skip = 0; skip < 3; ++skip) {
                std::getline(iss, l_quantity, '|');
            }
            std::getline(iss, l_quantity, '|');
            
            try {
                int ok = std::stoi(l_orderkey);
                auto cust_it = customerMap.find(orderMap[ok]);
                if (cust_it != customerMap.end()) {
                    local_sum += std::stoi(l_quantity);
                    local_n += 1;
                }
            } catch (const std::exception& e) {
                // 跳过转换错误的行，可根据需求添加日志
                continue;
            }
        }
        
        local_sums[i] = local_sum;
        local_counts[i] = local_n;
    });
}

// 等待线程完成并汇总结果
for (auto& t : threads) {
    t.join();
}

long long total_sum = 0;
int total_n = 0;
for (size_t i = 0; i < local_sums.size(); ++i) {
    total_sum += local_sums[i];
    total_n += local_counts[i];
}

方案2：生产者-消费者模式 #

用一个单独的生产者线程负责读取所有行，将行数据放入线程安全的队列；多个消费者线程从队列取行处理，计算局部总和，最后合并结果。这种方式彻底避免了多线程操作同一个文件流的问题。

代码示例片段 #

#include <queue>
#include <mutex>
#include <condition_variable>

// 线程安全队列实现
template<typename T>
class SafeQueue {
private:
    std::queue<T> queue_;
    std::mutex mutex_;
    std::condition_variable cv_;
    bool done_ = false;

public:
    void push(T item) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(item));
        cv_.notify_one();
    }
    
    bool pop(T& item) {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this]() { return done_ || !queue_.empty(); });
        if (done_ && queue_.empty()) return false;
        item = std::move(queue_.front());
        queue_.pop();
        return true;
    }
    
    void finish() {
        std::lock_guard<std::mutex> lock(mutex_);
        done_ = true;
        cv_.notify_all();
    }
};

// 并行处理逻辑
SafeQueue<std::string> line_queue;
std::vector<std::thread> consumer_threads;
std::vector<long long> local_sums;
std::vector<int> local_counts;
size_t num_consumers = std::thread::hardware_concurrency();

local_sums.resize(num_consumers, 0);
local_counts.resize(num_consumers, 0);

// 启动消费者线程
for (size_t i = 0; i < num_consumers; ++i) {
    consumer_threads.emplace_back([this, &line_queue, &local_sums, &local_counts, i]() {
        std::string l_line;
        long long local_sum = 0;
        int local_n = 0;
        
        while (line_queue.pop(l_line)) {
            std::istringstream iss(l_line);
            std::string l_orderkey, l_quantity;
            
            std::getline(iss, l_orderkey, '|');
            for (int skip = 0; skip < 3; ++skip) {
                std::getline(iss, l_quantity, '|');
            }
            std::getline(iss, l_quantity, '|');
            
            try {
                int ok = std::stoi(l_orderkey);
                auto cust_it = customerMap.find(orderMap[ok]);
                if (cust_it != customerMap.end()) {
                    local_sum += std::stoi(l_quantity);
                    local_n += 1;
                }
            } catch (const std::exception& e) {
                continue;
            }
        }
        
        local_sums[i] = local_sum;
        local_counts[i] = local_n;
    });
}

// 生产者线程读取文件
std::thread producer([this, &line_queue]() {
    std::ifstream l_file(this->lineitem);
    std::string l_line;
    
    while (std::getline(l_file, l_line)) {
        line_queue.push(std::move(l_line));
    }
    
    line_queue.finish(); // 通知消费者无更多数据
});

// 等待所有线程完成
producer.join();
for (auto& t : consumer_threads) {
    t.join();
}

// 汇总结果
long long total_sum = 0;
int total_n = 0;
for (size_t i = 0; i < num_consumers; ++i) {
    total_sum += local_sums[i];
    total_n += local_counts[i];
}

关键注意事项 #

• customerMap和orderMap必须是只读的（并行处理前已完全构建），否则需要额外加锁保护。
• 建议捕获std::stoi的异常，避免因脏数据导致程序崩溃。
• 线程数建议设置为CPU核心数，避免过度调度降低性能。

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