C语言中MySQL查询行传入线程池的内存优化问题

核心思路 #

1. 避免全量加载结果：用mysql_use_result替代mysql_store_result，逐行从MySQL服务器拉取数据，而非一次性将所有结果缓存到客户端内存，从根源降低内存占用。
2. 动态分配任务参数：放弃固定大小数组，为每一行数据单独动态分配任务参数结构体，处理完成后在任务函数内释放内存，避免固定数组的容量限制和内存浪费。
3. 平衡取数与处理速度：通过控制线程池待处理任务的最大数量，当队列满时暂停从数据库取数，等待线程处理部分任务后再继续，防止任务队列无限制膨胀。

修改后的代码 #

#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <pthread.h>
#include <unistd.h>
#include "thpool.h"

#define THREADS 10
#define MAX_PENDING_TASKS 20  // 待处理任务的最大阈值

struct fparam
{
  int id;
  char *data;
};

void process(void *arg)
{
  struct fparam *args = arg;
  // 自定义数据处理逻辑
  printf("%d - %s\n", args->id, args->data);
  
  // 释放动态分配的内存
  free(args->data);
  free(args);
}

int main(int argc, char **argv)
{
  threadpool thpool = thpool_init(THREADS);

  // MySQL连接初始化（需补充实际连接参数）
  MYSQL *con = mysql_init(NULL);
  if (!mysql_real_connect(con, "localhost", "user", "password", "db_name", 0, NULL, 0)) {
    fprintf(stderr, "%s\n", mysql_error(con));
    exit(EXIT_FAILURE);
  }

  // 执行查询语句
  if (mysql_query(con, "SELECT id, data FROM target_table")) {
    fprintf(stderr, "%s\n", mysql_error(con));
    mysql_close(con);
    exit(EXIT_FAILURE);
  }

  // 逐行获取结果，不缓存全量数据
  MYSQL_RES *result = mysql_use_result(con);
  if (!result) {
    fprintf(stderr, "%s\n", mysql_error(con));
    mysql_close(con);
    exit(EXIT_FAILURE);
  }

  MYSQL_ROW row;
  while ((row = mysql_fetch_row(result)))
  {
    // 等待待处理任务数降到阈值以下，避免队列爆内存
    while (thpool_num_waiting(thpool) >= MAX_PENDING_TASKS) {
      usleep(10000);  // 休眠10ms后重新检查
    }

    // 动态分配任务参数
    struct fparam *param = malloc(sizeof(struct fparam));
    if (!param) {
      fprintf(stderr, "Malloc failed\n");
      break;
    }

    // 复制数据：mysql_use_result的row内存会被下一次fetch覆盖，必须单独存储
    param->id = atoi(row[0]);
    param->data = strdup(row[1]);
    if (!param->data) {
      free(param);
      fprintf(stderr, "Strdup failed\n");
      break;
    }

    // 提交任务到线程池
    thpool_add_work(thpool, process, param);
  }

  // 检查取数过程中的错误
  if (mysql_errno(con)) {
    fprintf(stderr, "%s\n", mysql_error(con));
  }

  mysql_free_result(result);
  mysql_close(con);

  thpool_wait(thpool);
  thpool_destroy(thpool);

  exit(EXIT_SUCCESS);
}

关键细节说明 #

• mysql_use_result的特性：该函数不会在客户端缓存全量结果，每次mysql_fetch_row才从服务器获取一行数据，大幅降低客户端内存压力；但需注意，在调用mysql_free_result前不能执行其他MySQL命令，且row指向的内存会被下一次fetch覆盖，因此必须复制数据到自己分配的内存中。
• 内存回收逻辑：每个任务的参数结构体和数据字符串都是动态分配的，处理完成后在process函数内释放，确保内存及时回收，不会产生内存泄漏。
• 任务队列控制：通过thpool_num_waiting获取待处理任务数量，当达到阈值时暂停取数，平衡数据库取数速度和线程处理速度，避免任务队列无限增长占用过多内存。

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