OpenMP 并行化生产者-消费者模式：多线程文件解析

#include <stdio.h> #include <stdlib.h> #include <string.h> #include <omp.h> #define MAX_LINE_SIZE 1024 #define MAX_WORD_SIZE 128 #define QUEUE_SIZE 100 // 共享队列 char* queue[QUEUE_SIZE]; int front = 0; int rear = 0; int count = 0; // 生产者 void producer(char* filename) { FILE* file = fopen(filename, "r"); if (file == NULL) { printf("Error opening file: %s\n", filename); return; } char line[MAX_LINE_SIZE]; while (fgets(line, sizeof(line), file)) { char* new_line = strdup(line); // 复制一份文本行 #pragma omp critical // 保证队列操作的原子性 { while (count == QUEUE_SIZE) { #pragma omp flush(count) } queue[rear] = new_line; // 插入队列 rear = (rear + 1) % QUEUE_SIZE; count++; #pragma omp flush(rear, count) } } fclose(file); } // 消费者 void consumer() { while (1) { char* line; #pragma omp critical // 保证队列操作的原子性 { while (count == 0) { #pragma omp flush(count) } line = queue[front]; // 取出队列首元素 front = (front + 1) % QUEUE_SIZE; count--; #pragma omp flush(front, count) } char* word = strtok(line, " \t\n"); while (word != NULL) { printf("Word: %s\n", word); // 输出单词 word = strtok(NULL, " \t\n"); } free(line); } } int main() { char* filenames[] = {"file1.txt", "file2.txt", "file3.txt"}; int num_files = sizeof(filenames) / sizeof(filenames[0]);

#pragma omp parallel num_threads(num_files + 1)
{
    int thread_id &#x3D; omp_get_thread_num();
    if (thread_id &#x3C; num_files) {
        producer(filenames[thread_id]);
    } else {
        consumer();
    }
}

return 0;

} 该程序使用 OpenMP 并行编程模型实现了生产者-消费者模式。程序中包含一个生产者函数producer和一个消费者函数consumer，以及一个主函数main。主函数创建了多个线程，其中每个线程扮演生产者或消费者的角色。

生产者函数producer负责读取文件中的每一行文本，并将其复制到共享队列中。在复制之前，使用#pragma omp critical指令确保队列操作的原子性，以避免竞态条件。如果队列已满，则使用#pragma omp flush指令等待队列有空位。

消费者函数consumer负责从共享队列中取出文本行，并将其拆分为单词进行处理。使用#pragma omp critical指令确保队列操作的原子性，以避免竞态条件。如果队列为空，则使用#pragma omp flush指令等待队列有新的元素。

主函数main创建了多个线程，其中每个线程根据文件数量扮演生产者或消费者的角色。使用#pragma omp parallel指令并指定num_threads参数来创建线程。在主函数中，生产者线程负责调用producer函数读取文件内容并将其存入队列，而消费者线程负责调用consumer函数从队列中取出文本行并处理。