最佳多线程的方式来处理文件

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); 
}