如何用readdir實現多線程目錄掃描

使用 readdir 實現多線程目錄掃描，可以顯著提高掃描速度，特別是在處理包含大量文件和子目錄的目錄時。以下是使用 C 語言和 POSIX 線程（pthreads）實現多線程目錄掃描的基本步驟和示例代碼。

步驟概述

1. 主函數初始化：

   • 創建一個線程池或動態分配線程。
   • 將根目錄添加到待處理隊列中。

2. 工作線程函數：

   • 每個工作線程從隊列中獲取一個目錄路徑。
   • 使用 readdir 讀取目錄內容。
   • 對于每個條目：
     • 如果是子目錄，則將其添加到隊列中（如果尚未處理）。
     • 如果是文件，則進行處理（例如打印路徑）。

3. 同步機制：

   • 使用互斥鎖（mutex）保護共享資源（如目錄隊列）。
   • 使用條件變量通知工作線程有新的任務。

4. 結束條件：

   • 當隊列為空且所有線程都完成任務時，主線程退出。

示例代碼

以下是一個簡單的多線程目錄掃描示例：

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

// 定義最大線程數
#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 <directory>\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 和多線程技術實現高效的目錄掃描。根據具體需求，還可以進一步擴展功能，例如統計文件數量、過濾特定類型的文件等。