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

MySQL UDF 開發實戰：非阻塞超時重傳 HTTP POST 實現

編輯：MySQL 資料庫相關知識

前言

MySQL UDF（User Defined Function）允許使用 C/C++ 擴展資料庫功能。本文展示一個完整的 UDF 範例：`http_post`，實現非阻塞 Socket 連線、超時重試機制，可在觸發器或存儲過程中直接呼叫外部 HTTP 服務，適用於異步資料同步、事件通知等場景。

UDF 開發規範回顧

需實作三個入口函數：`xxx_init()`、`xxx()`、`xxx_deinit()`
編譯為動態鏈接庫，放置於 plugin 目錄，`CREATE FUNCTION` 載入
主函數內**禁止**直接 `malloc`/`new`，動態記憶體必須在 `init` 階段分配並掛載至 `initid->ptr`，`deint` 釋放
高併發下頻繁 malloc 易觸發 double-free，採「靜態緩衝 + 大數據動態分配」混合策略

完整原始碼解析

1. 標頭檔與跨平台型別定義

#ifdef STANDARD
#include 
#include 
#include 
#ifdef __WIN__
typedef unsigned __int64 ulonglong;
typedef __int64 longlong;
#else
typedef unsigned long long ulonglong;
typedef long long longlong;
#endif
#else
#include 
#include 
#if defined(MYSQL_SERVER)
#include 
#else
#include 
#endif
#endif

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include

2. 函數宣告

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

3. 初始化函數：參數校驗與記憶體預分配

my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message) {
    if (args->arg_count < 3) {
        strcpy(message, "Wrong arguments to http_post; need host, port, path [,body]");
        return 1;
    }

    // 第 4 參數為 POST body，超過 160KB 則動態分配
    if (args->arg_count == 4 && args->args[3] != NULL) {
        int flexibleLength = strlen(args->args[3]);
        if (flexibleLength > 160000) {
            int allocLength = 200 + flexibleLength;  // 預留 HTTP 標頭空間
            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;
}

4. 核心執行函數：非阻塞 Socket + select 超時重試

longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error) {
    int sockfd = 0, numbytes = 0, flags = 0, cycletimes = 0;
    char *sendBuffer = NULL;
    fd_set wset;
    struct timeval tval = {0, 300000};  // 300ms 超時

    // 選擇發送緩衝區
    if (initid->ptr == NULL) {
        static char sendArray[160000] = "\0";
        sendBuffer = sendArray;
    } else {
        sendBuffer = (char*)initid->ptr;
    }

    // 解析目標地址
    struct sockaddr_in serv_addr;
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(atoi(args->args[1]));
    serv_addr.sin_addr.s_addr = inet_addr(args->args[0]);
    bzero(&(serv_addr.sin_zero), 8);

    // 組裝 HTTP POST 請求
    if (args->arg_count == 4 && args->args[3] != NULL) {
        int argsNum = strlen(args->args[3]);
        sprintf(sendBuffer, "POST /?%s HTTP/1.1\r\nContent-Length:%d\r\n\r\n%s",
                args->args[2], argsNum, args->args[3]);
    } else {
        sprintf(sendBuffer, "POST /?%s HTTP/1.1\r\n", args->args[2]);
    }

    // 建立非阻塞 Socket
    if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        close(sockfd);
        return 2;  // Socket 建立失敗
    }
    flags = fcntl(sockfd, F_GETFL, 0);
    fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);

    // 連線 + 發送重試迴圈（最多 5 次）
    do {
        connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(struct sockaddr));
        FD_ZERO(&wset);
        FD_SET(sockfd, &wset);

        // select 監控可寫事件
        if (select(sockfd + 1, NULL, &wset, NULL, &tval) <= 0 && cycletimes == 5) {
            close(sockfd);
            return 5;  // 連線超時
        }

        numbytes = send(sockfd, sendBuffer, strlen(sendBuffer), 0);
        if (numbytes < 0) {
            usleep(20000);  // 20ms 退避
        }
        cycletimes++;
    } while (numbytes < 0 && cycletimes != 5);

    if (numbytes < 0) {
        close(sockfd);
        return 4;  // 發送失敗
    }

    close(sockfd);
    return 0;  // 成功
}

5. 解構函數：釋放動態記憶體

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

編譯與部署步驟

# 1. 編譯為 .so（路徑依實際 MySQL 安裝位置調整）
gcc -Wall -I/usr/local/mysql/include/mysql/ -shared http_post.c -o http_post.so -fPIC

# 2. 複製至 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('192.168.1.100', '8080', '/api/sync', '{"id":123}');

關鍵設計點總結

非阻塞 + select：避免 `connect`/`send` 卡死 MySQL 執行緒，設定 300ms 輪詢超時，最多重試 5 次（約 1.5 秒）。
混合記憶體策略：≤160KB 用棧上靜態陣列，>160KB 在 `init` 動態分配，`deinit` 釋放，規避高併發 double-free。
錯誤碼語義化：2=Socket失敗、4=發送失敗、5=連線超時，便於上層業務判斷重試或告警。
適用場景：觸發器異步推送、資料同步至異構系統、外部服務回調。不適用於需即時回應的同步業務。