Unix环境C++实现MapReduce时拆分大文本为8MB分块的方案咨询

1. 8MB文件分块的可选实现方案 #

• 逻辑分块（推荐最佳实践）：不实际切割物理文件，仅为每个分块记录文件路径、起始偏移量、分块长度三个元数据。拆分时需要额外处理跨行问题：如果8MB切割点落在某一行中间，需要将当前行剩余内容全部划入当前分块，下一个分块的起始偏移量移至该行结束的下一个字节，避免单词被切分导致统计错误。这种方案没有额外磁盘IO开销，是工业界MapReduce实现的通用方案。
• 物理分块：提前将大于8MB的文件切割为多个独立的小文件存储。仅适合需要重复处理同一批文件的场景，单次作业使用会增加不必要的磁盘读写开销，不推荐。

2. Map线程判断分块读取终止的实现 #

放弃原来用EOF作为结束标记的方案，改为分块级哨兵标记机制：
每个Split线程完成对应分块的所有行写入操作后，往绑定的线程安全队列中写入一个约定好的特殊哨兵值（比如空字符串指针、自定义的SplitEnd类型标记对象）。Map线程每次从队列读取到内容后先判断是否为哨兵值，如果是则代表当前分块读取完成，进入后续处理逻辑。
对于大文件拆分出的多个分块，每个分块对应独立的Split线程、独立的队列、独立的哨兵标记，互不干扰，天然适配多拆分场景。

3. Map线程无数据时的等待逻辑 #

基于条件变量实现线程安全队列的阻塞读取逻辑，这是多生产者消费者场景的标准实现：

1. 线程安全队列内部维护互斥锁、条件变量、以及数据存储容器
2. Map线程调用队列的pop()接口时，先加锁，判断如果队列为空则调用条件变量的wait()接口阻塞，释放锁等待唤醒
3. Split线程写入数据到队列后，调用条件变量的notify_one()接口唤醒阻塞的Map线程
4. 注意处理虚假唤醒问题：wait()接口第二个参数传入谓词，判断队列不为空时才结束等待，伪代码如下：

std::unique_lock<std::mutex> lock(queue_mutex);
cond.wait(lock, [this]{ return !queue.empty() || is_closed; });

这种方案完全避免轮询空等的CPU占用，性能最优。

分块描述符生成函数改造方案 #

首先定义分块描述结构体存储每个分块的元数据，再将原计数函数改造为返回分块描述数组的实现：

// 首先定义分块描述结构体
struct SplitInfo {
    char file_path[264];
    off_t start_offset; // 分块在文件中的起始偏移
    size_t split_size;  // 分块的大小
};

class MapReduce {
    // ... 原有类定义
public:
    // 改造后的函数，返回所有分块的描述数组
    std::vector<SplitInfo> generateSplitInfos() {
        std::vector<SplitInfo> split_infos;
        char file_path[264];

        DIR* dir = opendir(InputPath);
        if (!dir) return split_infos; // 目录打开失败容错
        
        struct dirent* entity;
        unsigned char isFile = 0x8;

        while ((entity = readdir(dir)) != NULL) {
            if (strcmp(entity->d_name, ".") != 0 && strcmp(entity->d_name, "..") != 0 && entity->d_type == isFile) {
                struct stat file_status;
                snprintf(file_path, sizeof(file_path), "%s/%s", InputPath, entity->d_name);
                if (stat(file_path, &file_status) != 0) continue; // stat失败容错

                long file_size = file_status.st_size;
                // 简化分块数计算逻辑
                int chunk_count = (file_size + MAX_SPLIT_SIZE - 1) / MAX_SPLIT_SIZE;
                off_t current_offset = 0;

                for (int i = 0; i < chunk_count; i++) {
                    SplitInfo info;
                    strncpy(info.file_path, file_path, sizeof(info.file_path) - 1);
                    info.start_offset = current_offset;
                    // 最后一个分块大小取剩余长度
                    info.split_size = (i == chunk_count - 1) ? (file_size - current_offset) : MAX_SPLIT_SIZE;
                    split_infos.push_back(info);
                    current_offset += MAX_SPLIT_SIZE;
                }
            }
        }
        closedir(dir);
        return split_infos;
    }
};

改造后返回的vector的大小就是原来的split_num，每个元素对应一个分块的完整元数据，Split线程可以直接读取对应偏移的内容处理。

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