8、回调函数与线程：在pthread中传递回调函数，实现任务处理

多线程编程里，回调函数是个好东西。

我刚开始用pthread那会儿，总觉得线程就是开个函数跑一跑，完事。后来项目做多了才发现，真正麻烦的不是开线程，而是怎么让线程把结果告诉你，或者怎么让线程按你的要求去处理不同的任务。

说白了，回调函数就是解决这个问题的。

8.1 为什么线程里需要回调？

你想想看，一个线程函数通常是固定的。比如你写了个线程去处理网络数据包，但不同的数据包需要不同的处理逻辑。难道每次都要改线程函数？

当然不行。

我有个项目是做物联网网关的，要同时处理温度、湿度、气压三种传感器数据。每个传感器的解析方式完全不同。如果写三个线程，代码冗余不说，维护起来也头疼。

后来我用了回调函数，一个线程池，传入不同的回调指针，搞定。

核心思想：线程只负责执行框架（比如接收数据、调用回调），具体的业务逻辑由回调函数决定。这样线程代码一次写好，永远不改。

8.2 pthread_create 传递回调的两种方式

pthread_create 的第四个参数是 void *arg，可以传任意指针。这就是我们传递回调的入口。

方式一：直接传函数指针

最简单粗暴。把回调函数地址塞进结构体，传给线程。

#include <pthread.h>
#include <stdio.h>

// 回调函数类型定义
typedef void (*task_callback_t)(int data);

// 传给线程的参数结构体
typedef struct {
    int input_data;
    task_callback_t callback;
} thread_task_t;

// 一个具体的回调实现
void my_callback(int result) {
    printf("回调执行：处理结果 = %d\n", result);
}

// 线程函数
void* worker_thread(void* arg) {
    thread_task_t* task = (thread_task_t*)arg;
    
    // 模拟一些处理
    int processed = task->input_data * 2;
    
    // 调用回调，把结果传回去
    task->callback(processed);
    
    return NULL;
}

int main() {
    pthread_t tid;
    thread_task_t task = {
        .input_data = 42,
        .callback = my_callback
    };
    
    pthread_create(&tid, NULL, worker_thread, &task);
    pthread_join(tid, NULL);
    
    return 0;
}

嗯，这里要注意：回调函数是在线程上下文中执行的。如果你的回调里访问了全局变量，记得加锁。

方式二：用函数指针数组实现多路分发

我在一个嵌入式项目中用过这个模式。当时要处理12种不同的命令码，每个命令码对应一个处理函数。

#include <pthread.h>
#include <stdio.h>

// 定义回调类型
typedef void (*cmd_handler_t)(const char* params);

// 命令处理函数表
void handle_led_on(const char* p)  { printf("LED ON: %s\n", p); }
void handle_led_off(const char* p) { printf("LED OFF: %s\n", p); }
void handle_temp_read(const char* p){ printf("TEMP READ: %s\n", p); }

// 命令码到回调的映射表
typedef struct {
    int cmd_code;
    cmd_handler_t handler;
} cmd_entry_t;

cmd_entry_t cmd_table[] = {
    {1, handle_led_on},
    {2, handle_led_off},
    {3, handle_temp_read}
};
#define CMD_TABLE_SIZE (sizeof(cmd_table)/sizeof(cmd_table[0]))

// 线程参数
typedef struct {
    int cmd;
    char params[64];
} thread_msg_t;

void* cmd_dispatcher(void* arg) {
    thread_msg_t* msg = (thread_msg_t*)arg;
    
    for(int i = 0; i < CMD_TABLE_SIZE; i++) {
        if(cmd_table[i].cmd_code == msg->cmd) {
            cmd_table[i].handler(msg->params);
            return NULL;
        }
    }
    
    printf("未知命令码: %d\n", msg->cmd);
    return NULL;
}

int main() {
    pthread_t tid;
    thread_msg_t msg = {2, "channel=1"};  // 开灯命令
    
    pthread_create(&tid, NULL, cmd_dispatcher, &msg);
    pthread_join(tid, NULL);
    
    return 0;
}

我的习惯：回调函数表用 const 修饰，放在只读区。既安全又省RAM。嵌入式设备上RAM金贵，能省一点是一点。

8.3 回调与线程安全

这里有个坑，我踩过不止一次。

回调函数在线程里执行，如果回调里修改了共享数据，而主线程也在用同一份数据——恭喜你，数据竞争来了。

我曾经写过一个日志系统，主线程往队列里写日志，工作线程通过回调把日志刷到SD卡。结果偶尔丢日志，查了两天才发现是回调里没加锁。

避坑指南：

  • 回调函数里如果访问全局变量或静态变量，必须加互斥锁
  • 回调函数不要阻塞太久，否则会拖死工作线程
  • 回调里不要调用 pthread_exit 或 exit，会导致整个进程退出

8.4 回调 + 线程池的经典模式

实际项目中，我们很少单独开一个线程就跑一个回调。更常见的做法是：线程池 + 任务队列 + 回调。

下面是一个简化版的线程池回调模型：

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

#define POOL_SIZE 3

// 任务结构体
typedef struct task {
    void (*func)(void* arg);
    void* arg;
    struct task* next;
} task_t;

// 线程池
typedef struct {
    pthread_t threads[POOL_SIZE];
    task_t* task_queue;
    pthread_mutex_t lock;
    pthread_cond_t cond;
    int shutdown;
} thread_pool_t;

thread_pool_t pool;

// 工作线程
void* worker(void* arg) {
    (void)arg;
    
    while(1) {
        pthread_mutex_lock(&pool.lock);
        
        while(pool.task_queue == NULL && !pool.shutdown) {
            pthread_cond_wait(&pool.cond, &pool.lock);
        }
        
        if(pool.shutdown) {
            pthread_mutex_unlock(&pool.lock);
            pthread_exit(NULL);
        }
        
        // 取出任务
        task_t* task = pool.task_queue;
        pool.task_queue = task->next;
        
        pthread_mutex_unlock(&pool.lock);
        
        // 执行回调
        task->func(task->arg);
        
        free(task);
    }
    
    return NULL;
}

// 提交任务
void submit_task(void (*func)(void*), void* arg) {
    task_t* task = malloc(sizeof(task_t));
    task->func = func;
    task->arg = arg;
    task->next = NULL;
    
    pthread_mutex_lock(&pool.lock);
    
    // 插入队列尾部
    if(pool.task_queue == NULL) {
        pool.task_queue = task;
    } else {
        task_t* p = pool.task_queue;
        while(p->next) p = p->next;
        p->next = task;
    }
    
    pthread_cond_signal(&pool.cond);
    pthread_mutex_unlock(&pool.lock);
}

// 一个具体的回调
void sensor_process(void* arg) {
    int sensor_id = *(int*)arg;
    printf("线程 %lu 处理传感器 %d 数据\n", 
           pthread_self(), sensor_id);
    // 实际处理...
}

int main() {
    // 初始化线程池
    pthread_mutex_init(&pool.lock, NULL);
    pthread_cond_init(&pool.cond, NULL);
    pool.shutdown = 0;
    pool.task_queue = NULL;
    
    for(int i = 0; i < POOL_SIZE; i++) {
        pthread_create(&pool.threads[i], NULL, worker, NULL);
    }
    
    // 提交3个传感器任务
    int ids[3] = {1, 2, 3};
    for(int i = 0; i < 3; i++) {
        submit_task(sensor_process, &ids[i]);
    }
    
    // 等待一会儿
    sleep(1);
    
    // 清理
    pool.shutdown = 1;
    pthread_cond_broadcast(&pool.cond);
    for(int i = 0; i < POOL_SIZE; i++) {
        pthread_join(pool.threads[i], NULL);
    }
    
    pthread_mutex_destroy(&pool.lock);
    pthread_cond_destroy(&pool.cond);
    
    return 0;
}

8.5 回调在线程中的生命周期

我画了一张图，帮你理清回调从提交到执行的全过程：

回调函数在线程中的生命周期 主线程 任务队列 工作线程 ① 提交回调 ② 取出回调 执行回调函数 ③ 调用 返回结果/通知 ④ 回调完成 ⑤ 主线程通过同步机制（锁/条件变量）获取结果 回调函数本身不关心它在哪个线程执行，只关心被调用时参数正确、环境安全

8.6 实际项目中的注意事项

最后，分享几个我在实战中总结的经验：

  • 回调函数要短小精悍——工作线程不是给你做复杂计算的。如果回调逻辑很重，考虑再开一个线程或者用异步队列。
  • 参数生命周期要管好——传给回调的指针，在线程执行完之前不能释放。我见过有人传了栈上变量的地址，线程还没跑完变量就销毁了，结果回调里拿到的是野指针。
  • 错误处理不能省——回调里如果出错了，最好通过返回值或者错误码通知调用方，而不是直接 abort。
  • 调试时加个日志——多线程回调的调用链很难追踪。我习惯在每个回调入口和出口加一行调试日志，线上出问题能快速定位。

一个小技巧：如果你不确定回调是否会在多线程中被重入，可以在回调开头加一个 pthread_mutex_trylock，如果锁失败说明有重入，直接返回错误。这个做法帮我抓出过好几个隐蔽的bug。

好了，关于pthread中传递回调函数实现任务处理，就聊这么多。记住一句话：回调让线程变得灵活，但灵活的背后需要更严谨的设计。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →