如何通过队列实现父进程向子进程传数据并协同处理任务

• fork生成的子进程会复制父进程的内存空间，每个子进程都拥有独立的文件队列副本，因此两个子进程都会读取初始队列的第一个元素str1。
• 当前代码中子进程仅处理一个文件就退出，未实现循环获取下一个文件直到队列清空的逻辑。

要让多个子进程共享同一个队列并安全获取任务，需要共享内存存储队列，同时用信号量保证多进程访问队列的互斥性，避免竞态条件。

以下是修改后的完整代码：

#include <iostream>
#include <unistd.h>
#include <stdlib.h>
#include <sys/wait.h>
#include <vector>
#include <stdio.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <semaphore.h>
#include <queue>
#include <string>
using namespace std;

// 定义共享内存中的结构体，包含队列和信号量
struct SharedData {
    sem_t mutex;          // 互斥信号量，保护队列访问
    sem_t queue_not_empty;// 队列非空信号量，通知子进程有任务
    queue<string> files;  // 共享文件队列
};

int main(int argc, char **argv) {
    vector<pid_t> process;
    pid_t pid = 0;

    // 创建共享内存
    int shm_fd = shm_open("/file_queue_shm", O_CREAT | O_RDWR, 0666);
    ftruncate(shm_fd, sizeof(SharedData));
    SharedData* shared_data = (SharedData*)mmap(NULL, sizeof(SharedData), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);

    // 初始化信号量
    sem_init(&shared_data->mutex, 1, 1);          // 互斥信号量，初始值1（支持多进程共享）
    sem_init(&shared_data->queue_not_empty, 1, 0);// 队列非空信号量，初始值0（先填充队列再通知子进程）

    // 填充文件队列到共享内存
    sem_wait(&shared_data->mutex);
    for (int i = 1; i <= 10; ++i) {
        string str = "str" + to_string(i);
        shared_data->files.push(str);
    }
    sem_post(&shared_data->mutex);

    // 发送10个信号，通知子进程队列有任务
    for (int i = 0; i < 10; ++i) {
        sem_post(&shared_data->queue_not_empty);
    }

    // 创建2个子进程
    for (int i = 0; i < 2; i++) {
        cout << "Prefork child process! " << i << endl;
        pid = fork();
        if (pid == 0) {
            while (true) {
                // 等待队列有任务
                sem_wait(&shared_data->queue_not_empty);
                // 加锁访问队列
                sem_wait(&shared_data->mutex);
                
                if (shared_data->files.empty()) {
                    sem_post(&shared_data->mutex);
                    break;
                }

                string curr = shared_data->files.front();
                shared_data->files.pop();
                cout << "Child " << i << " (PID: " << getpid() << ") got: " << curr << endl;
                
                // 解锁队列
                sem_post(&shared_data->mutex);
            }
            return 0;
        } else {
            process.push_back(pid);
            cout << "Parent for child " << i << " with pid: " << pid << endl;
        }
    }

    // 等待所有子进程结束
    int status;
    while ((pid = waitpid(-1, &status, 0)) != -1) {
        cout << "Process terminated: " << pid << endl;
    }

    // 清理资源
    sem_destroy(&shared_data->mutex);
    sem_destroy(&shared_data->queue_not_empty);
    munmap(shared_data, sizeof(SharedData));
    shm_unlink("/file_queue_shm");

    return 0;
}

• 共享内存：通过shm_open和mmap创建共享内存区域，让父子进程访问同一块内存中的队列，避免内存复制导致的队列副本问题。
• 信号量：
  • mutex：保证同一时间只有一个进程操作队列，防止多个进程同时读写导致数据混乱。
  • queue_not_empty：初始值为0，父进程填充队列后发送对应数量的信号，子进程等待该信号再获取任务，避免空轮询浪费资源。
• 子进程逻辑：子进程循环等待信号，获取并处理队列中的文件，直到队列为空后退出。

编译时需要链接实时库：

g++ -o file_processor file_processor.cpp -lrt -pthread

运行：

./file_processor

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