19、多线程与数据库：SQLite线程模式、连接池设计、线程安全操作、互斥锁使用

多线程环境下操作数据库，是很多C程序员头疼的事。说实话，我早年做嵌入式项目时，就因为在多线程里直接共享同一个SQLite连接，导致数据错乱，排查了整整两天。从那以后，我对线程安全这块就格外上心。

这一章，咱们就聊聊SQLite在多线程下的那些事。包括它的线程模式、怎么设计一个轻量级的连接池、以及互斥锁的正确用法。

19.1 SQLite的三种线程模式

SQLite本身支持三种线程模式，你可以在编译时或运行时选择。我个人习惯在编译时直接指定，省得运行时还要操心。

模式	宏定义	说明
单线程模式	SQLITE_THREADSAFE=0	所有互斥锁被禁用，性能最高，但只能在单线程中使用
多线程模式	SQLITE_THREADSAFE=1	单个连接不能被多线程共享，但不同线程可用不同连接
序列化模式	SQLITE_THREADSAFE=2	完全线程安全，一个连接可以被多个线程安全使用

19.2 连接池设计思路

你想想看，如果每个线程都自己打开一个数据库连接，那连接数会随着线程数暴涨。嵌入式设备资源有限，扛不住的。

连接池的核心思想很简单：预先创建一批连接，线程需要时借，用完还回去。这样既控制了连接总数，又避免了频繁打开关闭的开销。

我设计过一个轻量级连接池，结构大概是这样：

typedef struct {
    sqlite3 **conns;      // 连接数组
    int *available;       // 可用标记，1表示空闲
    int total;            // 总连接数
    pthread_mutex_t mutex; // 互斥锁
    pthread_cond_t cond;   // 条件变量
} ConnectionPool;

初始化时，我一次性创建 total 个连接。每个线程调用 pool_get() 时，从池里找一个空闲连接；用完调用 pool_put() 归还。

19.3 线程安全操作与互斥锁

连接池本身需要互斥锁保护，不然多个线程同时取连接，数据就乱了。我习惯用 pthread_mutex_t，简单可靠。

下面是一个取连接的示例：

sqlite3* pool_get(ConnectionPool *pool) {
    pthread_mutex_lock(&pool->mutex);
    for (int i = 0; i < pool->total; i++) {
        if (pool->available[i]) {
            pool->available[i] = 0;
            pthread_mutex_unlock(&pool->mutex);
            return pool->conns[i];
        }
    }
    // 没有空闲连接，等待
    pthread_cond_wait(&pool->cond, &pool->mutex);
    // 唤醒后递归获取
    pthread_mutex_unlock(&pool->mutex);
    return pool_get(pool);
}

嗯，这里要注意：pthread_cond_wait 会释放锁，被唤醒后重新获取。这个机制能避免忙等待，节省CPU。

19.4 核心知识体系

下面这张图，是我梳理的多线程与SQLite的核心逻辑。你看一遍，基本就心里有数了。

19.5 完整示例：多线程写入

最后，我贴一个实际用过的多线程写入示例。两个线程同时往数据库里写数据，通过连接池保证安全。

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

#define NUM_THREADS 2
#define POOL_SIZE 3

typedef struct {
    sqlite3 *conns[POOL_SIZE];
    int available[POOL_SIZE];
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} ConnectionPool;

ConnectionPool pool;

void* worker(void *arg) {
    int id = *(int*)arg;
    sqlite3 *db = pool_get(&pool);
    
    char sql[128];
    snprintf(sql, sizeof(sql), 
             "INSERT INTO logs VALUES (%d, 'thread %d');", id, id);
    
    char *err = NULL;
    sqlite3_exec(db, sql, NULL, NULL, &err);
    if (err) {
        fprintf(stderr, "SQL error: %s\n", err);
        sqlite3_free(err);
    }
    
    pool_put(&pool, db);
    return NULL;
}

int main() {
    // 初始化连接池
    pthread_mutex_init(&pool.mutex, NULL);
    pthread_cond_init(&pool.cond, NULL);
    for (int i = 0; i < POOL_SIZE; i++) {
        sqlite3_open("test.db", &pool.conns[i]);
        pool.available[i] = 1;
    }
    
    pthread_t threads[NUM_THREADS];
    int ids[NUM_THREADS];
    for (int i = 0; i < NUM_THREADS; i++) {
        ids[i] = i;
        pthread_create(&threads[i], NULL, worker, &ids[i]);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }
    
    // 清理
    for (int i = 0; i < POOL_SIZE; i++) {
        sqlite3_close(pool.conns[i]);
    }
    pthread_mutex_destroy(&pool.mutex);
    pthread_cond_destroy(&pool.cond);
    return 0;
}