使用Pthreads实现任务队列：多线程并行处理

本文将演示如何使用Pthreads库在C语言中实现一个任务队列，利用多线程并行处理任务，从而提高程序效率。

代码示例

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

// 任务结构体
typedef struct Task {
    int data;
    struct Task* next;
} Task;

// 全局变量
Task* taskQueue = NULL;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int tasksRemaining = 0;

// 添加任务到队列
void addTask(int data) {
    Task* newTask = (Task*)malloc(sizeof(Task));
    newTask->data = data;
    newTask->next = NULL;

    pthread_mutex_lock(&mutex);
    if (taskQueue == NULL) {
        taskQueue = newTask;
    } else {
        Task* current = taskQueue;
        while (current->next != NULL) {
            current = current->next;
        }
        current->next = newTask;
    }
    tasksRemaining++;
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
}

// 从队列中获取任务
Task* getTask() {
    Task* task = NULL;
    pthread_mutex_lock(&mutex);
    while (taskQueue == NULL && tasksRemaining > 0) {
        pthread_cond_wait(&cond, &mutex);
    }
    if (taskQueue != NULL) {
        task = taskQueue;
        taskQueue = taskQueue->next;
        tasksRemaining--;
    }
    pthread_mutex_unlock(&mutex);
    return task;
}

// 执行任务
void* processTask(void* arg) {
    while (1) {
        Task* task = getTask();
        if (task == NULL) {
            break;
        }

        // 执行任务
        printf("Processing task with data: %d\n", task->data);
        free(task);
    }
    return NULL;
}

int main() {
    int numThreads;
    int numTasks;

    printf("Enter the number of threads: ");
    scanf("%d", &numThreads);
    printf("Enter the number of tasks: ");
    scanf("%d", &numTasks);

    pthread_t* threads = (pthread_t*)malloc(numThreads * sizeof(pthread_t));

    // 创建线程
    for (int i = 0; i < numThreads; i++) {
        pthread_create(&threads[i], NULL, processTask, NULL);
    }

    // 生成任务
    for (int i = 0; i < numTasks; i++) {
        addTask(i);
    }

    // 等待任务执行完毕
    while (tasksRemaining > 0) {
        pthread_cond_broadcast(&cond);
    }

    // 等待线程结束
    for (int i = 0; i < numThreads; i++) {
        pthread_join(threads[i], NULL);
    }

    free(threads);

    return 0;
}

工作原理

任务结构体： Task 结构体包含一个数据成员 (data) 和一个指向下一个任务的指针 (next)，用于构建链表形式的任务队列。
全局变量： taskQueue 指向任务队列的头节点，mutex 是一个互斥锁，用于保护任务队列的访问，cond 是一个条件变量，用于线程间同步，tasksRemaining 记录待处理的任务数量。
添加任务： addTask 函数将新的任务添加到任务队列的末尾，并使用 pthread_cond_signal 唤醒一个等待的线程。
获取任务： getTask 函数从任务队列中获取一个任务，并使用 pthread_cond_wait 进入条件等待状态，直到队列中存在任务或所有任务都已完成。
执行任务： processTask 函数循环获取任务并执行，如果获取不到任务，则退出循环。
主线程： 主线程创建指定数量的工作线程，生成任务并添加到队列中。在所有任务生成完毕后，使用 pthread_cond_broadcast 唤醒所有线程，并等待所有线程结束。

注意事项

代码示例中没有进行内存分配错误处理，实际应用中应该进行相应的处理。
为了简化代码，没有对条件变量的等待进行超时处理，实际应用中可以根据需要进行相应的处理。
为了提高程序效率，可以考虑使用更快的链表操作算法，例如双向链表。

总结

本文介绍了使用 Pthreads 库实现一个任务队列，并使用多线程并行处理任务，提高程序效率。代码示例展示了如何创建线程、添加任务、获取任务和执行任务，以及如何使用条件变量和互斥锁实现线程同步和安全访问共享资源。

希望这篇文章对你有所帮助！