您现在的位置：程式碼之家 >> 數據庫知識 > >> MySQL 資料庫 >> MySQL 綜合教學

MySQL UDF + libcurl 實現 HTTP POST 通信模塊：從 C 開發到生產部署全流程

編輯：MySQL 綜合教學

架構背景

某遊戲項目需將 SNS 社交餵流數據實時同步至自研 TCSQL 數據庫。方案：MySQL Trigger 觸發 → UDF 函數調用 → libcurl 異步 HTTP POST → 目標系統 REST API。核心技術棧：**MySQL UDF API (C/C++) + libcurl multi 接口 (異步非阻塞)**。

UDF 三大入口函數規範

// 初始化：參數校驗、記憶體分配 (返回 0 成功，1 失敗)
my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message);

// 主執行函數：核心業務邏輯 (返回 longlong 結果)
longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error);

// 反初始化：釋放資源
void http_post_deinit(UDF_INIT *initid);

核心代碼實現精要

1. 初始化函數：靜態/動態記憶體混合策略防 Double Free

my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message) {
    if (args->arg_count != 2) {  // url, post_data
        strcpy(message, "http_post requires 2 arguments: url, post_data");
        return 1;
    }
    
    // 靜態緩衝區 160KB，超過則動態分配
    if (args->arg_count == 2 && args->args[1] != NULL) {
        size_t len = strlen(args->args[1]);
        if (len > 160000) {
            initid->ptr = malloc(len + 200);
            if (!initid->ptr) {
                strcpy(message, "Memory allocation failed");
                return 1;
            }
        } else {
            initid->ptr = NULL;  // 使用棧上靜態數組
        }
    }
    return 0;
}

2. 主函數：libcurl Multi 異步非阻塞發送

longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error) {
    char *send_buf = initid->ptr ? (char*)initid->ptr : (char[160000]){0};
    strcpy(send_buf, args->args[1]);  // post_data
    
    CURL *curl = curl_easy_init();
    CURLM *multi = curl_multi_init();
    int still_running = 1, tries = 0, max_tries = (initid->ptr ? 100 : 25);
    struct timeval timeout = {0, 100000};  // 100ms
    
    if (curl && multi) {
        curl_easy_setopt(curl, CURLOPT_URL, args->args[0]);        // url
        curl_easy_setopt(curl, CURLOPT_POST, 1L);
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, send_buf);
        curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, 5000);          // 總超時 5s
        curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, 2000);   // 連接超時 2s
        curl_multi_add_handle(multi, curl);
        
        // 驅動傳輸
        while (CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi, &still_running));
        
        while (still_running && tries < max_tries) {
            fd_set fdread, fdwrite, fdexcep;
            int maxfd = -1;
            FD_ZERO(&fdread); FD_ZERO(&fdwrite); FD_ZERO(&fdexcep);
            curl_multi_fdset(multi, &fdread, &fdwrite, &fdexcep, &maxfd);
            
            int rc = select(maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout);
            if (rc == -1) break;  // select error
            
            while (CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi, &still_running));
            tries++;
        }
        
        curl_multi_remove_handle(multi, curl);
        curl_multi_cleanup(multi);
        curl_easy_cleanup(curl);
        
        return (tries >= max_tries) ? 1 : 0;  // 0 成功，1 超時/失敗
    }
    return 1;  // init failed
}

3. 反初始化：安全釋放

void http_post_deinit(UDF_INIT *initid) {
    if (initid && initid->ptr) {
        free(initid->ptr);
        initid->ptr = NULL;
    }
}

編譯部署流程

# 1. 編譯生成共享庫 (路徑按實際調整)
gcc -Wall -I/usr/local/mysql/include/mysql -shared http_post.c -o http_post.so -fPIC -lcurl

# 2. 複製到 MySQL plugin 目錄
cp http_post.so /usr/local/mysql/lib/mysql/plugin/

# 3. MySQL 命令行註冊函數
mysql> DROP FUNCTION IF EXISTS http_post;
mysql> CREATE FUNCTION http_post RETURNS INTEGER SONAME 'http_post.so';

# 4. 測試調用
mysql> SELECT http_post('https://target.com/api/sync', '{"feed_id":123,"action":"like"}');
+--------------------------------------------------------------+
| http_post('https://target.com/api/sync', '{"feed_id":123...') |
+--------------------------------------------------------------+
|                                                            0 |
+--------------------------------------------------------------+

Trigger 整合示例

CREATE TRIGGER `trg_feed_sync` AFTER INSERT ON `sns_feed`
FOR EACH ROW
BEGIN
    DECLARE v_payload JSON;
    SET v_payload = JSON_OBJECT(
        'feed_id', NEW.id,
        'user_id', NEW.user_id,
        'content', NEW.content,
        'ts', NEW.created_at
    );
    -- 異步發送，不阻塞主業務
    SELECT http_post('https://tcsql-gateway:8080/sync/feed', JSON_UNQUOTE(JSON_EXTRACT(v_payload, '$')));
END;

關鍵設計決策解析

決策點	選擇	理由
HTTP 庫	libcurl multi (異步)	easy 同步模式會阻塞等待 Response，multi 發送完即可返回，適合「發後即忘」場景
記憶體策略	靜態 160KB + 動態分配	UDF 主函數禁用 malloc/new；高併發下動態分配易觸發 Double Free，小數據走棧內存零分配
超時控制	select + 最大重試次數	防止目標端無 Response 導致死循環，小包 25 次、大包 100 次動態調整
返回值	INTEGER (0 成功，非 0 失敗)	簡單明確，Trigger 中可用 `SELECT ... INTO @var` 捕獲但不阻塞

生產級強化建議

連接池化：UDF 無狀態，每次調用重建連接開銷大。可在 `init` 階段建立連接池 (全域靜態)，`http_post` 復用 `easy handle`，`deinit` 清理。
熔斷降級：統計連續失敗次數，超閾值直接返回錯誤不再嘗試，保護目標服務。
異步隊列化：高頻寫入場景，UDF 僅寫入本地內存隊列/Redis Stream，獨立 Worker 消費發送，徹底解耦。
監控指標：QPS、延遲分位 (P50/P99)、錯誤率、目標端可用性。
安全加固：HTTPS + 憑證驗證 (`CURLOPT_SSL_VERIFYPEER=1`)、簽名鑒權、IP 白名單。

總結

MySQL UDF + libcurl 實現了資料庫層面的主動異構推送能力，解決了「Trigger 中無法直接發 HTTP」的難題。關鍵在於異步非阻塞發送、記憶體安全管理、超時熔斷機制。但需認清：**資料庫不應承擔網絡 I/O 職責**，極高併發下建議遷移至應用層異步隊列架構 (Kafka/RocketMQ + Consumer)。