多线程文件读取处理代码偶发异常排查求助

核心问题排查 #

你的代码存在几个关键问题，是导致间歇性异常的主要原因：

• 循环终止条件错误
  生产者和消费者都依赖!gzeof(fileIn)作为循环判断，但gzFile并非线程安全结构，消费者线程调用gzeof会和生产者的gzgets产生竞态。此外，生产者读完所有数据后，缓冲区中可能还有未处理的内容，消费者会因为gzeof返回真而直接退出，导致数据丢失；或者生产者停止后，消费者卡在sem_wait(&reads_full)无法退出。

• 缺少生产者结束后的通知机制
  生产者完成读取后，没有主动唤醒等待的消费者线程，导致部分消费者会一直阻塞在sem_wait(&reads_full)，程序无法正常退出。

• gzgets缓冲区溢出风险
  固定64字节的缓冲区如果遇到超过63字符的行，会被截断，导致后续读取的行内容错乱，引发处理逻辑异常。

代码优化建议 #

1. 重构终止逻辑 #

• 新增原子变量finished标记生产者是否完成读取，避免多线程访问gzFile的gzeof接口。
• 生产者完成读取后，多次调用sem_post(&reads_full)唤醒所有消费者，确保缓冲区剩余数据被处理完毕。
• 消费者循环改为判断!finished或缓冲区还有未处理数据，处理完所有数据后再退出。

2. 确保gzFile操作的独占性 #

仅允许生产者线程对输入gzFile进行读写操作，消费者线程不再访问fileIn，彻底避免竞态。

3. 修复线程ID传递潜在问题 #

将线程ID转换为intptr_t直接传递，避免指针引用可能带来的竞态。

4. 增强缓冲区安全性 #

增大缓冲区或改为动态分配，同时检查gzgets的返回值，确保读取完整有效行。

修改后的代码示例 #

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <semaphore.h>
#include <string.h>
#include <zlib.h>
#include <stdatomic.h>

#define READER_SIZE 30
#define NUM_PROCESSORS 4
#define LINE_BUFFER_SIZE 256  // 增大缓冲区避免截断

// 生产者-消费者共享数据
char reader_buffer[READER_SIZE][LINE_BUFFER_SIZE]; 
int reader_in = 0;
int reader_out = 0;

sem_t reads_full;
sem_t reads_empty;
pthread_mutex_t buffer_mutex;
gzFile fileIn;
atomic_int finished = ATOMIC_VAR_INIT(0);  // 原子标记生产者是否完成

// 写操作共享数据
pthread_mutex_t writes_mutex;
gzFile fileOut;

void *producer()
{
    char line[LINE_BUFFER_SIZE];
    while (gzgets(fileIn, line, LINE_BUFFER_SIZE) != NULL) {
        sem_wait(&reads_empty);
        pthread_mutex_lock(&buffer_mutex);

        strncpy(reader_buffer[reader_in], line, LINE_BUFFER_SIZE);
        reader_in = (reader_in + 1) % READER_SIZE;

        pthread_mutex_unlock(&buffer_mutex);
        sem_post(&reads_full);
    }

    // 标记生产者完成，唤醒所有等待的消费者
    atomic_store(&finished, 1);
    for (int i = 0; i < NUM_PROCESSORS; i++) {
        sem_post(&reads_full);
    }
    return NULL;
}

void *consumer(void *arg)
{
    int consumer_id = (intptr_t)arg;
    char processor_buffer[LINE_BUFFER_SIZE];

    while (1) {
        sem_wait(&reads_full);
        pthread_mutex_lock(&buffer_mutex);

        // 检查是否已经处理完所有数据
        if (atomic_load(&finished) && reader_in == reader_out) {
            pthread_mutex_unlock(&buffer_mutex);
            sem_post(&reads_full);  // 传递唤醒信号给其他消费者
            break;
        }

        strncpy(processor_buffer, reader_buffer[reader_out], LINE_BUFFER_SIZE);
        reader_out = (reader_out + 1) % READER_SIZE;

        pthread_mutex_unlock(&buffer_mutex);
        sem_post(&reads_empty);

        // 处理逻辑：仅保存以@开头的行
        if (processor_buffer[0] == '@') {
            pthread_mutex_lock(&writes_mutex);
            gzputs(fileOut, processor_buffer);
            pthread_mutex_unlock(&writes_mutex);
        }
    }
    return NULL;
}

int main()
{
    fileIn = gzopen("temp.gz", "rb");
    if (!fileIn) {
        perror("Failed to open input file");
        return 1;
    }
    fileOut = gzopen("test.gz", "wb");
    if (!fileOut) {
        perror("Failed to open output file");
        gzclose(fileIn);
        return 1;
    }

    pthread_t reader_thread;
    pthread_t processor_thread[NUM_PROCESSORS];

    // 初始化同步原语
    sem_init(&reads_full, 0, 0);
    sem_init(&reads_empty, 0, READER_SIZE);
    pthread_mutex_init(&buffer_mutex, NULL);
    pthread_mutex_init(&writes_mutex, NULL);

    pthread_create(&reader_thread, NULL, producer, NULL);

    for (int i = 0; i < NUM_PROCESSORS; i++) {
        pthread_create(&processor_thread[i], NULL, consumer, (void *)(intptr_t)i);
    }

    // 等待所有线程结束
    pthread_join(reader_thread, NULL);
    for (int i = 0; i < NUM_PROCESSORS; i++) {
        pthread_join(processor_thread[i], NULL);
    }

    // 清理资源
    sem_destroy(&reads_full);
    sem_destroy(&reads_empty);
    pthread_mutex_destroy(&buffer_mutex);
    pthread_mutex_destroy(&writes_mutex);

    gzclose(fileIn);
    gzclose(fileOut);

    return 0;
}

内容的提问来源于stack exchange，提问作者yuw444