如何用readdir实现多线程目录扫描

使用 readdir 实现多线程目录扫描，可以显著提高扫描速度，特别是在处理包含大量文件和子目录的目录时。以下是使用 C 语言和 POSIX 线程（pthreads）实现多线程目录扫描的基本步骤和示例代码。

步骤概述

1. 主函数初始化：

   • 创建一个线程池或动态分配线程。
   • 将根目录添加到待处理队列中。

2. 工作线程函数：

   • 每个工作线程从队列中获取一个目录路径。
   • 使用 readdir 读取目录内容。
   • 对于每个条目：
     • 如果是子目录，则将其添加到队列中（如果尚未处理）。
     • 如果是文件，则进行处理（例如打印路径）。

3. 同步机制：

   • 使用互斥锁（mutex）保护共享资源（如目录队列）。
   • 使用条件变量通知工作线程有新的任务。

4. 结束条件：

   • 当队列为空且所有线程都完成任务时，主线程退出。

示例代码

以下是一个简单的多线程目录扫描示例：

#include 
#include 
#include 
#include 
#include 

// 定义最大线程数
#define MAX_THREADS 10

// 全局变量
int thread_count = 0;
pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
int done = 0;

// 目录项结构体
typedef struct {
    char path[1024];
} dir_entry;

// 工作线程函数
void* scan_directory(void* arg) {
    while (1) {
        pthread_mutex_lock(&queue_mutex);
        
        // 等待直到有任务或完成
        while (thread_count >= MAX_THREADS && !done) {
            pthread_cond_wait(&queue_cond, &queue_mutex);
        }
        
        if (done && queue_empty()) {
            pthread_mutex_unlock(&queue_mutex);
            pthread_exit(NULL);
        }
        
        // 获取目录项
        dir_entry entry;
        if (!queue_dequeue(&entry)) {
            pthread_mutex_unlock(&queue_mutex);
            break;
        }
        pthread_mutex_unlock(&queue_mutex);
        
        // 打开目录
        DIR* dir = opendir(entry.path);
        if (dir == NULL) {
            perror("opendir");
            continue;
        }
        
        struct dirent* dp;
        while ((dp = readdir(dir)) != NULL) {
            if (strcmp(dp->d_name, ".") == 0 || strcmp(dp->d_name, "..") == 0)
                continue;
            
            char child_path[1024];
            snprintf(child_path, sizeof(child_path), "%s/%s", entry.path, dp->d_name);
            
            pthread_mutex_lock(&queue_mutex);
            if (thread_count < MAX_THREADS) {
                queue_enqueue(child_path);
                thread_count++;
            } else {
                pthread_cond_signal(&queue_cond);
            }
            pthread_mutex_unlock(&queue_mutex);
        }
        
        closedir(dir);
    }
    return NULL;
}

// 简单的队列实现
#define QUEUE_CAPACITY 1024

typedef struct {
    dir_entry items[QUEUE_CAPACITY];
    int front;
    int rear;
    int size;
} queue_t;

queue_t queue = { .front = 0, .rear = -1, .size = 0 };

int queue_empty() {
    return queue.size == 0;
}

int queue_enqueue(dir_entry item) {
    if (queue.size >= QUEUE_CAPACITY) {
        return 0; // 队列满
    }
    queue.rear = (queue.rear + 1) % QUEUE_CAPACITY;
    queue.items[queue.rear] = item;
    queue.size++;
    return 1;
}

int queue_dequeue(dir_entry* item) {
    if (queue_empty()) {
        return 0; // 队列空
    }
    *item = queue.items[queue.front];
    queue.front = (queue.front + 1) % QUEUE_CAPACITY;
    queue.size--;
    return 1;
}

// 添加目录到队列
void add_to_queue(const char* path) {
    pthread_mutex_lock(&queue_mutex);
    if (thread_count < MAX_THREADS) {
        dir_entry entry;
        snprintf(entry.path, sizeof(entry.path), "%s", path);
        queue_enqueue(entry);
        thread_count++;
        pthread_cond_signal(&queue_cond);
    } else {
        pthread_cond_wait(&queue_cond, &queue_mutex);
        add_to_queue(path); // 递归添加
    }
    pthread_mutex_unlock(&queue_mutex);
}

int main(int argc, char* argv[]) {
    if (argc != 2) {
        fprintf(stderr, "Usage: %s \n", argv[0]);
        return EXIT_FAILURE;
    }
    
    const char* start_path = argv[1];
    
    // 添加起始目录到队列
    add_to_queue(start_path);
    
    // 创建工作线程
    pthread_t threads[MAX_THREADS];
    for (int i = 0; i < MAX_THREADS; i++) {
        if (pthread_create(&threads[i], NULL, scan_directory, NULL) != 0) {
            perror("pthread_create");
            exit(EXIT_FAILURE);
        }
    }
    
    // 等待所有线程完成
    pthread_mutex_lock(&queue_mutex);
    done = 1;
    pthread_cond_broadcast(&queue_cond);
    pthread_mutex_unlock(&queue_mutex);
    
    for (int i = 0; i < MAX_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }
    
    return EXIT_SUCCESS;
}

代码说明

1. 队列实现：

   • 使用一个简单的环形队列 queue_t 来存储待扫描的目录路径。
   • 提供 queue_enqueue 和 queue_dequeue 函数来添加和移除目录项。
   • 使用互斥锁 queue_mutex 和条件变量 queue_cond 来同步对队列的访问。

2. 线程管理：

   • 主线程通过 add_to_queue 函数将起始目录添加到队列中。
   • 创建固定数量的工作线程，每个线程执行 scan_directory 函数。
   • 工作线程不断从队列中获取目录路径，使用 readdir 读取内容，并根据需要将子目录添加到队列中。
   • 使用 thread_count 变量跟踪当前活跃的线程数，以控制并发度。

3. 同步与结束：

   • 当所有目录都被处理完毕后，主线程设置 done = 1 并广播条件变量，通知工作线程退出。
   • 工作线程在检测到 done 标志且队列为空时，结束自身。

4. 错误处理：

   • 对于无法打开的目录，打印错误信息并继续处理其他目录。

注意事项

• 递归深度：上述示例没有限制递归深度，如果目录结构非常深，可能会导致大量线程被创建。可以考虑增加递归深度限制或优化线程管理策略。

• 性能优化：

  • 可以根据系统资源动态调整 MAX_THREADS 的值，以达到最佳性能。
  • 使用更高效的数据结构或线程池库（如 pthreadpool）来管理线程。

• 平台兼容性：此示例基于 POSIX 标准，适用于类 Unix 系统。如果在 Windows 上实现，需要使用 Windows 线程 API（如 CreateThread）和相应的同步机制。

• 安全性：确保对共享资源的访问都受到互斥锁的保护，避免竞态条件和数据不一致。

通过上述方法，可以有效地使用 readdir 和多线程技术实现高效的目录扫描。根据具体需求，还可以进一步扩展功能，例如统计文件数量、过滤特定类型的文件等。