您现在的位置：程式碼之家 >> 數據庫知識 > >> MySQL 資料庫 >> MySQL 資料庫相關知識

MySQL UDF + libcurl 實現高效異步 HTTP POST 通訊模組

編輯：MySQL 資料庫相關知識

專案背景

本模組用於「逍遙江湖」SNS 與金山自研 TCSQL 資料庫的資料同步：Feed 寫入 SNS 觸發器呼叫 UDF，經 HTTP POST 推送至 TCSQL。採用 `libcurl` 的 `curl_multi` 異步介面，解決傳統 `curl_easy` 同步阻塞、無回應超時難控制等問題。

架構設計

MySQL UDF 介面：提供 `http_post_init`、`http_post`、`http_post_deinit` 三個入口
libcurl multi 異步引擎：非阻塞並發傳輸，配合 `select` 事件循環
記憶體管理策略：靜態 160KB 緩衝 + 大數據動態分配，規避高併發 double-free

核心程式碼實現

1. UDF 入口函數宣告

my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message);
longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error);
void http_post_deinit(UDF_INIT *initid);

2. 初始化函數：參數檢查與緩衝區準備

my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message) {
    if (args->arg_count != 2) {  // url, post_data
        strcpy(message, "Wrong arguments to http_post; need url, post_data");
        return 1;
    }

    if (args->args[1] != NULL) {
        int flexibleLength = strlen(args->args[1]);
        if (flexibleLength > 160000) {
            int allocLength = 200 + flexibleLength;
            if (!(initid->ptr = (char*)malloc(allocLength))) {
                strcpy(message, "Couldn't allocate memory in http_post_init");
                return 1;
            }
            return 0;
        }
    }
    initid->ptr = NULL;
    return 0;
}

3. 主執行函數：curl_multi 異步流程

longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error) {
    char *sendBuffer = NULL;
    CURL *curl = NULL;
    CURLM *multi_handle = NULL;
    int still_running = 0;
    int times = 0;
    const int TRY_TIMES_SMALL = 25;
    const int TRY_TIMES_LARGE = 100;
    struct timeval timeout = {0, 100000};  // 100ms

    // 靜態緩衝區（不可放入 if 區塊，否則生命週期結束）
    static char sendArray[160000] = "\0";

    if (initid->ptr == NULL) {
        sendBuffer = sendArray;
    } else {
        sendBuffer = (char*)initid->ptr;
        // 大數據允許更多重試次數
    }

    strcpy(sendBuffer, args->args[1]);  // POST body

    curl = curl_easy_init();
    multi_handle = curl_multi_init();

    if (curl && multi_handle) {
        curl_easy_setopt(curl, CURLOPT_URL, args->args[0]);
        curl_easy_setopt(curl, CURLOPT_POST, 1L);
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, sendBuffer);
        curl_multi_add_handle(multi_handle, curl);

        // 1. 執行至非 CURLM_CALL_MULTI_PERFORM 狀態
        while (CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi_handle, &still_running));

        // 2. 事件循環：select 等待可讀寫，再執行 perform
        while (still_running && times < (initid->ptr ? TRY_TIMES_LARGE : TRY_TIMES_SMALL)) {
            int maxfd = 0;
            fd_set fdread, fdwrite, fdexcep;
            FD_ZERO(&fdread);
            FD_ZERO(&fdwrite);
            FD_ZERO(&fdexcep);

            curl_multi_fdset(multi_handle, &fdread, &fdwrite, &fdexcep, &maxfd);
            int rc = select(maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout);

            switch (rc) {
                case -1:  // select 錯誤
                    break;
                case 0:   // 超時
                default:
                    while (CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi_handle, &still_running));
                    break;
            }
            times++;
        }

        curl_multi_remove_handle(multi_handle, curl);
        curl_multi_cleanup(multi_handle);
        curl_easy_cleanup(curl);

        if (times >= (initid->ptr ? TRY_TIMES_LARGE : TRY_TIMES_SMALL))
            return 1;  // 超時失敗
        return 0;      // 成功
    }
    return 1;  // 初始化失敗
}

4. 解構函數

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

為何選擇 curl_multi 而非 curl_easy？

維度	curl_easy (同步)	curl_multi (異步)
發送後行為	阻塞等待 Response	非阻塞，可控制超時邏輯
無 Response 服務	永久阻塞或靠超時中斷	發送完成即可返回，無需等待
大數據傳輸	超時難平衡	分塊傳輸，配合事件循環
併發擴展性	需多線程	單線程管理多連線

編譯與安裝指南

# 1. 編譯（需 link libcurl）
gcc -Wall -I/usr/local/webserver/mysql/include/mysql/ \
    -shared http_post.c -o http_post.so -fPIC -lcurl

# 2. 部署
cp http_post.so /usr/local/webserver/mysql/lib/mysql/plugin/http_post.so

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

# 4. 測試（需目標主機啟動監聽 3888 埠）
mysql> SELECT http_post('https://testpost.com/index.php', 'sfasfa');

生產環境優化建議

連線池化：UDF 內頻繁 `curl_easy_init`/`cleanup` 有開銷，可在 `init` 階段預建 `CURL*` 緩存至 `initid->ptr`，`deinit` 統一釋放。
超時策略分級：依業務 SLA 設定連線超時 (`CURLOPT_CONNECTTIMEOUT`)、傳輸超時 (`CURLOPT_TIMEOUT`)、低速限制 (`CURLOPT_LOW_SPEED_LIMIT`/`TIME`)。
錯誤碼擴充：區分 DNS 失敗、連線拒絕、TLS 驗證失敗、HTTP 非 2xx 等，便於上層重試決策。
監控埋點：記錄耗時、重試次數、錯誤分布至系統監控（Prometheus/Grafana）。
替代方案評估：MySQL 8.0+ 建議改用「資料庫觸發器 → 寫入 Kafka/RocketMQ → 消費者推 HTTP」，解耦且更穩健。