如何将基于回调的C API封装为非回调同步API并保证线程安全？

我正尝试在C语言中实现二进制格式与JSON的双向转换（以便实现可脚本化的映射）。我调研了13款C JSON解析器，仅找到2款符合全部需求：一款采用LGPL协议，另一款为回调式API。由于将代码放入独立DLL以避免LGPL与专有代码混合的操作过于复杂，因此我优先选择回调式API。

现有回调式API结构大致如下（经过大幅精简）：

typedef void *(*api_malloc_func)(void *ctx, size_t sz);
/* ... */
typedef struct {
    api_malloc_func malloc;
    /* ... */
} api_config_params;

typedef struct {
    int (* api_on_bool)(void * ctx, int boolVal);
    int (* api_on_int)(void * ctx, long long intVal);
    /* ...*/
    /* "user-defined data", passed to callbacks. */
    void * ctx;
} api_callbacks;

api_handle api_config(const api_callbacks * callbacks, const api_config_params * cfg);

api_status api_parse(api_handle hand, const unsigned char * json, size_t len);

但为了让该API与二进制API风格一致，我需要的是非回调/非事件式（即同步）API，结构如下：

typedef struct {
    api_malloc_func malloc;
    /* ... */
    /* "user-defined data" */
    void * ctx;
} api_config_params;

api_handle api_config(const api_config_params * cfg);

typedef enum api_type_t {
    api_type_bool,
    api_type_int,
    /* ... */
} api_type_t;

api_type_t api_peek_type(api_config_params* reader);

bool api_read_bool(api_config_params* reader, bool* boolVal);

bool api_read_int(api_config_params* reader, long long* intVal);

/* ... */

我现在主要从事Java开发，已经很久没有编写C/C++代码，因此无法理清如何将回调式事件API封装为非回调的同步API。请问该如何实现？理想情况下要保证线程安全，这是否可行？

注：我也可以修改二进制API使其采用回调模式，但这会降低性能，而我仅关注二进制API的速度，对JSON API的速度无要求。

核心思路：用中间缓冲区缓存回调数据 #

回调式JSON解析器是事件驱动的，会在解析到对应类型时立刻触发回调；而你需要的同步API是拉取式的，由调用方主动查询、读取数据。两者的适配核心是在回调中把解析出的数据暂存到一个缓冲区（比如队列或树形结构），之后同步API的接口就从这个缓冲区里读取数据。

具体实现步骤 #

1. 定义中间缓存结构

首先需要一个线程安全的缓存结构，用来存储解析出来的JSON数据节点。每个节点要记录类型（对应api_type_t）和对应的值：

typedef struct json_node {
    api_type_t type;
    union {
        bool bool_val;
        long long int_val;
        // 其他类型：字符串、数组、对象等，按需扩展
        char* str_val;
        struct json_node* array_nodes;
        size_t array_len;
        // ...
    } value;
    struct json_node* next; // 用于队列或链表
} json_node_t;

// 缓存上下文，包含队列头/尾、内存分配函数、用户自定义数据
typedef struct json_parser_cache {
    json_node_t* queue_head;
    json_node_t* queue_tail;
    api_malloc_func malloc;
    void* user_ctx;
    // 线程安全需要的锁（Windows/跨平台兼容）
#ifdef _WIN32
    CRITICAL_SECTION lock;
#else
    pthread_mutex_t lock;
#endif
} json_parser_cache_t;

2. 实现回调函数，将解析数据存入缓存

回调函数的作用是把解析到的数据包装成json_node_t，加入到缓存队列中：

static int on_bool(void* ctx, int boolVal) {
    json_parser_cache_t* cache = (json_parser_cache_t*)ctx;
    // 加锁保证线程安全
#ifdef _WIN32
    EnterCriticalSection(&cache->lock);
#else
    pthread_mutex_lock(&cache->lock);
#endif

    // 分配节点内存
    json_node_t* node = (json_node_t*)cache->malloc(cache->user_ctx, sizeof(json_node_t));
    if (!node) {
        // 解锁并返回错误
#ifdef _WIN32
        LeaveCriticalSection(&cache->lock);
#else
        pthread_mutex_unlock(&cache->lock);
#endif
        return -1;
    }
    node->type = api_type_bool;
    node->value.bool_val = (bool)boolVal;
    node->next = NULL;

    // 加入队列
    if (!cache->queue_head) {
        cache->queue_head = node;
        cache->queue_tail = node;
    } else {
        cache->queue_tail->next = node;
        cache->queue_tail = node;
    }

    // 解锁
#ifdef _WIN32
    LeaveCriticalSection(&cache->lock);
#else
    pthread_mutex_unlock(&cache->lock);
#endif
    return 0;
}

// 同理实现on_int等其他类型的回调函数
static int on_int(void* ctx, long long intVal) {
    json_parser_cache_t* cache = (json_parser_cache_t*)ctx;
#ifdef _WIN32
    EnterCriticalSection(&cache->lock);
#else
    pthread_mutex_lock(&cache->lock);
#endif

    json_node_t* node = (json_node_t*)cache->malloc(cache->user_ctx, sizeof(json_node_t));
    if (!node) {
#ifdef _WIN32
        LeaveCriticalSection(&cache->lock);
#else
        pthread_mutex_unlock(&cache->lock);
#endif
        return -1;
    }
    node->type = api_type_int;
    node->value.int_val = intVal;
    node->next = NULL;

    if (!cache->queue_head) {
        cache->queue_head = node;
        cache->queue_tail = node;
    } else {
        cache->queue_tail->next = node;
        cache->queue_tail = node;
    }

#ifdef _WIN32
    LeaveCriticalSection(&cache->lock);
#else
    pthread_mutex_unlock(&cache->lock);
#endif
    return 0;
}

3. 封装同步API的上下文结构

你的同步API需要一个上下文，内部包含回调式API的api_handle和我们定义的缓存结构：

// 同步API的handle，对外隐藏内部细节
typedef struct api_sync_handle {
    api_handle json_handle;
    json_parser_cache_t cache;
} api_sync_handle;

4. 实现同步API的各个接口

（1）初始化接口api_config

api_handle api_config(const api_config_params* cfg) {
    if (!cfg || !cfg->malloc) return NULL;

    // 分配同步handle内存
    api_sync_handle* sync_handle = (api_sync_handle*)cfg->malloc(cfg->ctx, sizeof(api_sync_handle));
    if (!sync_handle) return NULL;

    // 初始化缓存结构
    sync_handle->cache.malloc = cfg->malloc;
    sync_handle->cache.user_ctx = cfg->ctx;
    sync_handle->cache.queue_head = NULL;
    sync_handle->cache.queue_tail = NULL;

    // 初始化线程锁
#ifdef _WIN32
    InitializeCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_init(&sync_handle->cache.lock, NULL);
#endif

    // 配置回调式API的参数
    api_callbacks callbacks = {0};
    callbacks.api_on_bool = on_bool;
    callbacks.api_on_int = on_int;
    // 其他回调函数按需赋值
    callbacks.ctx = &sync_handle->cache;

    // 初始化回调式API的handle
    sync_handle->json_handle = api_config(&callbacks, cfg);
    if (!sync_handle->json_handle) {
        // 初始化失败，清理资源
#ifdef _WIN32
        DeleteCriticalSection(&sync_handle->cache.lock);
#else
        pthread_mutex_destroy(&sync_handle->cache.lock);
#endif
        cfg->malloc(cfg->ctx, sync_handle);
        return NULL;
    }

    return (api_handle)sync_handle;
}

（2）解析JSON（内部触发回调填充缓存）

你需要一个解析接口（可在初始化后调用，用于触发JSON解析并填充缓存）：

api_status api_parse_sync(api_handle hand, const unsigned char* json, size_t len) {
    if (!hand) return API_STATUS_INVALID_HANDLE;

    api_sync_handle* sync_handle = (api_sync_handle*)hand;
    // 解析前清空缓存（如果需要重复解析，这里要实现缓存清理逻辑避免内存泄漏）
    // ...

    return api_parse(sync_handle->json_handle, json, len);
}

（3）api_peek_type：查看下一个数据类型

api_type_t api_peek_type(api_handle reader) {
    if (!reader) return api_type_invalid; // 需要提前定义invalid类型

    api_sync_handle* sync_handle = (api_sync_handle*)reader;
    api_type_t type = api_type_invalid;

#ifdef _WIN32
    EnterCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_lock(&sync_handle->cache.lock);
#endif

    if (sync_handle->cache.queue_head) {
        type = sync_handle->cache.queue_head->type;
    }

#ifdef _WIN32
    LeaveCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_unlock(&sync_handle->cache.lock);
#endif

    return type;
}

（4）api_read_bool等读取接口

bool api_read_bool(api_handle reader, bool* boolVal) {
    if (!reader || !boolVal) return false;

    api_sync_handle* sync_handle = (api_sync_handle*)reader;
    bool success = false;

#ifdef _WIN32
    EnterCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_lock(&sync_handle->cache.lock);
#endif

    if (sync_handle->cache.queue_head && sync_handle->cache.queue_head->type == api_type_bool) {
        *boolVal = sync_handle->cache.queue_head->value.bool_val;
        // 从队列中移除节点并释放内存
        json_node_t* temp = sync_handle->cache.queue_head;
        sync_handle->cache.queue_head = temp->next;
        if (!sync_handle->cache.queue_head) {
            sync_handle->cache.queue_tail = NULL;
        }
        sync_handle->cache.malloc(sync_handle->cache.user_ctx, temp);
        success = true;
    }

#ifdef _WIN32
    LeaveCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_unlock(&sync_handle->cache.lock);
#endif

    return success;
}

// api_read_int同理实现
bool api_read_int(api_handle reader, long long* intVal) {
    if (!reader || !intVal) return false;

    api_sync_handle* sync_handle = (api_sync_handle*)reader;
    bool success = false;

#ifdef _WIN32
    EnterCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_lock(&sync_handle->cache.lock);
#endif

    if (sync_handle->cache.queue_head && sync_handle->cache.queue_head->type == api_type_int) {
        *intVal = sync_handle->cache.queue_head->value.int_val;
        json_node_t* temp = sync_handle->cache.queue_head;
        sync_handle->cache.queue_head = temp->next;
        if (!sync_handle->cache.queue_head) {
            sync_handle->cache.queue_tail = NULL;
        }
        sync_handle->cache.malloc(sync_handle->cache.user_ctx, temp);
        success = true;
    }

#ifdef _WIN32
    LeaveCriticalSection(&sync_handle->cache.lock);
#else
    pthread_mutex_unlock(&sync_handle->cache.lock);
#endif

    return success;
}

5. 线程安全的实现说明

• 每个同步api_handle对应独立的缓存和锁，不同线程使用不同handle时天然线程安全；
• 同一handle被多线程调用时，通过CRITICAL_SECTION（Windows）或pthread_mutex_t（POSIX）保护缓存的读写操作，避免竞态条件；
• 如果需要支持多线程同时解析不同的JSON，只要每个线程使用独立的api_sync_handle即可。

额外注意事项 #

• 内存管理：要确保所有分配的节点都被正确释放，比如在销毁handle时遍历缓存队列释放所有节点；
• 复杂类型支持：如果需要处理JSON数组、对象等复杂类型，缓存结构需要扩展（比如数组节点存储子节点列表），同步API也要对应添加读取数组/对象的接口；
• 性能权衡：因为你对JSON API的速度无要求，这种缓存方式的性能损耗完全可接受，不会影响二进制API的速度。

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