最佳多線程的方式來處理文件

Question

我有一些分隔的文件，我要處理的文件（依次獨立）的每一行，我希望它是快速的線路數。

所以我寫了一個代碼讀取文件的一大塊入緩衝區的內存大小，然後多線程將競爭讀取緩衝區裏的行並對其進行處理。的僞代碼如下：

do{ 
    do{  

    fread(buffer,500MB,1,file); 
    // creating threads 
    // let the threads compete to read from buffer and PROCESS independently 
    // end of threads 

    while(EOF not reached) 
    file = nextfile; 
while(there is another file to read) 

或者這一個：

void mt_ReadAndProcess(){ 
    lock(); 
    fread(buffer,50MB,1,file); 
    if(EOF reached) 
    file = nextfile; 
    unlock(); 
    process(); 
} 
main(){ 
    // create multi threads 
    // call mt_ReadAndProcess() with multi threads 
} 

過程中的一個（及時）昂貴的過程。

有沒有什麼更好的方法來做到這一點？更快地讀取文件或使用多線程處理文件的更好方法？

謝謝全部，

Ameer。

Answer 1

爲什麼你想有線程「競爭從緩衝區中讀取」？這些數據可以，因爲它是由線程做閱讀讀很容易劃分。爭取從緩衝區獲取數據沒有任何好處，但可能會浪費CPU和掛鐘時間。

由於您正在逐行處理，只需讀取文件中的行，並通過指針將緩衝區傳遞給工作線程。

假設你POSIX兼容的系統上運行，這樣的：

#include <unistd.h> 
#include <pthread.h> 

#define MAX_LINE_LEN 1024 
#define NUM_THREADS 8 

// linePipe holds pointers to lines sent to 
// worker threads 
static int linePipe[ 2 ]; 

// bufferPipe holds pointers to buffers returned 
// from worker threads and used to read data 
static int bufferPipe[ 2 ]; 

// thread function that actually does the work 
void *threadFunc(void *arg) 
{ 
    const char *linePtr; 

    for (;;) 
    { 
     // get a pointer to a line from the pipe 
     read(linePipe[ 1 ], &linePtr, sizeof(linePtr)); 

     // end loop on NULL linePtr value 
     if (!linePtr) 
     { 
      break; 
     } 

     // process line 

     // return the buffer 
     write(bufferPipe[ 0 ], &linePtr, sizeof(linePtr)); 
    } 

    return(NULL); 
} 

int main(int argc, char **argv) 
{ 
    pipe(linePipe); 
    pipe(bufferPipe); 

    // create buffers and load them into the buffer pipe for reading 
    for (int ii = 0; ii < (2 * NUM_THREADS); ii++) 
    { 
     char *buffer = malloc(MAX_LINE_LEN); 
     write(bufferPipe[ 0 ], &buffer, sizeof(buffer)); 
    } 

    pthread_t tids[ NUM_THREADS ]; 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     pthread_create(&(tids[ ii ]), NULL, thread_func, NULL); 
    } 

    FILE *fp = ... 

    for (;;) 
    { 
     char *linePtr; 

     // get the pointer to a buffer from the buffer pipe 
     read(bufferPipe[ 1 ], &linePtr, sizeof(linePtr)); 

     // read a line from the current file into the buffer 
     char *result = fgets(linePtr, MAX_LINE_LEN, fp); 

     if (result) 
     { 
      // send the line to the worker threads 
      write(linePipe, &linePtr, sizeof(linePtr)); 
     } 
     else 
     { 
      // either end loop, or open another file 
      fclose(fp); 
      fp = fopen(...); 
     } 
    } 

    // clean up and exit 

    // send NULL to cause worker threads to stop 
    char *nullPtr = NULL; 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     write(linePipe[ 0 ], &nullPtr, sizeof(nullPtr)); 
    } 

    // wait for worker threads to stop 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     pthread_join(tids[ ii ], NULL); 
    } 

    return(0); 
}