如何拆分大文件数据做多线程单词统计，避免拆分单词导致计数错误？

核心拆分逻辑优化 #

要避免拆分单词，你可以调整块的拆分规则：

• 初始块偏移按固定大小（如1024字节）计算，读取完固定长度的内容后，继续向后读取直到遇到第一个空白分隔符（空格、换行、制表符等）为止
• 当前块的读取范围到该分隔符位置结束，下一个块的起始偏移直接设置为该分隔符的下一位，就能保证所有完整单词都归属某一个块，不会被切分

还有一种更简单的兼容实现：给每个块增加重叠读取区域，每个块读取时多往后读最多100字节（覆盖常规单词的最大长度），处理块内容时主动丢弃开头不是从分隔符开始的残缺单词、结尾不是到分隔符结束的残缺单词，也能避免统计错误。

现有代码其他问题修正 #

你当前的实现还有几个严重问题会导致功能错误、性能失效：

• 你把整个worker函数逻辑都加了全局互斥锁，所有线程实际是串行执行的，完全没有利用到多线程的并行优势，建议每个线程先维护局部单词集合，全部线程执行完毕后再合并到全局集合，仅在合并时加锁即可
• 存在内存越界问题：你申请了BUFER_SIZE长度的memblock数组，却执行了memblock[BUFER_SIZE] = '\0'，写入了数组外的内存，会触发未定义行为
• 宏定义拼写错误：BUFER_SIZE少了一个F，应该为BUFFER_SIZE

修正后参考代码 #

#include <iostream>
#include <fstream>
#include <set>
#include <string>
#include <thread>
#include <mutex>
#include <sstream>
#include <vector>
#include <algorithm>
#include <cctype>
#define BUFFER_SIZE  1024
// 最大单词长度，用于重叠读
#define MAX_WORD_LEN 100

using namespace std;
std::mutex mtx;

void worker(int start, int end, set<std::string> &globalSet, std::string path)
{
    // 每个线程自己维护局部set，避免频繁加锁
    set<std::string> localSet;
    ifstream file (path, ios::in);
    if (file.is_open())
    {
        // 重叠读：多往后读MAX_WORD_LEN个字节，避免结尾单词被切
        int readLen = end - start + MAX_WORD_LEN;
        char *memblock = new char [readLen + 1]();
        file.seekg (start, ios::beg);
        file.read(memblock, readLen);
        std::string blockString(memblock);
        
        // 跳过开头的残缺单词：第一个块不需要跳
        size_t startPos = 0;
        if (start != 0) {
            while (startPos < blockString.size() && !isspace(blockString[startPos])) {
                startPos++;
            }
        }
        // 处理到原本的end位置即可，后面的重叠区不用管
        size_t endPos = end - start;
        if (endPos >= blockString.size()) endPos = blockString.size();
        
        std::string buf;
        stringstream stream(blockString.substr(startPos, endPos - startPos));
        while(stream >> buf) localSet.insert(buf);
        
        file.close();
        delete[] memblock;
    }
    else 
        cout << "Unable to open file" << endl;
    
    // 仅合并全局set的时候加锁
    std::lock_guard<std::mutex> lock(mtx);
    globalSet.insert(localSet.begin(), localSet.end());
}

int main(int argc, char *argv[])
{
    if (argc < 2) {
        std::cout << "Usage: " << argv[0] << " <file_path>" << endl;
        return 1;
    }
    set<std::string> uniqWords;
    int threadCount = 0;
    ifstream file(argv[1], ios::in | ios::binary);
    if(!file){
        std::cout << "Bad path.\n";
        return 1;
    }
    file.seekg(0, ios::end);
    int fileSize = file.tellg();
    file.close();
    std::cout << "Size of the file is " << fileSize << " bytes\n";
    threadCount = fileSize/BUFFER_SIZE + 1;
    std::cout << "Thread count: " << threadCount << std::endl;
    std::vector<std::thread> vec;
    for(int i=0; i < threadCount; i++)
    {
        int start = i * BUFFER_SIZE;
        int end = min((i+1)*BUFFER_SIZE, fileSize);
        vec.push_back(std::thread(worker, start, end, std::ref(uniqWords), argv[1]));
    }
    std::for_each(vec.begin(), vec.end(), [](std::thread& th)
    {
        th.join();
    });
    std::cout << "Unique words count: " << uniqWords.size() << std::endl;
    return 0;
}

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