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

MySQL UDF 非阻塞超時重傳進階實現：原生 Socket + select 模式

編輯：MySQL 綜合教學

設計目標

在不依賴 libcurl 的環境下，使用原生 Socket 實現 HTTP POST 發送，支援**非阻塞連接、發送超時重試、自定義協議構造**，適合極簡依賴部署場景。

完整 C 代碼實現 (http_post_udf.c)

#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 

#ifdef HAVE_DLOPEN

// --- UDF 接口聲明 ---
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);

// --- 初始化：參數校驗、大數據動態分配 ---
my_bool http_post_init(UDF_INIT *initid, UDF_ARGS *args, char *message) {
    if (args->arg_count < 3) {  // ip, port, path, [post_data]
        strcpy(message, "http_post requires 3 or 4 args: ip, port, path, [post_data]");
        return 1;
    }
    
    if (args->arg_count == 4 && args->args[3] != NULL) {
        size_t len = strlen(args->args[3]);
        if (len > 160000) {
            initid->ptr = malloc(len + 200);
            if (!initid->ptr) {
                strcpy(message, "Memory allocation failed");
                return 1;
            }
        } else {
            initid->ptr = NULL;
        }
    }
    return 0;
}

// --- 反初始化：釋放動態內存 ---
void http_post_deinit(UDF_INIT *initid) {
    if (initid && initid->ptr) {
        free(initid->ptr);
        initid->ptr = NULL;
    }
}

// --- 主函數：非阻塞 Socket + select 超時控制 ---
longlong http_post(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error) {
    int sockfd = -1, numbytes = 0, flags = 0, cycles = 0;
    char *send_buf = NULL;
    fd_set wset;
    struct timeval tv = {0, 300000};  // 300ms select 超時
    
    // 緩衝區選擇
    if (initid->ptr == NULL) {
        static char static_buf[160000] = {0};
        send_buf = static_buf;
    } else {
        send_buf = (char*)initid->ptr;
    }
    
    // 目標地址解析
    struct sockaddr_in serv_addr = {0};
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(atoi(args->args[1]));  // port
    serv_addr.sin_addr.s_addr = inet_addr(args->args[0]);  // ip
    
    // 構造 HTTP POST 請求
    if (args->arg_count == 4 && args->args[3] != NULL) {
        int body_len = strlen(args->args[3]);
        sprintf(send_buf, 
            "POST /?%s HTTP/1.1\r\n"
            "Host: %s\r\n"
            "Content-Length: %d\r\n"
            "Connection: Close\r\n\r\n"
            "%s", 
            args->args[2], args->args[0], body_len, args->args[3]);
    } else {
        sprintf(send_buf, 
            "POST /?%s HTTP/1.1\r\n"
            "Host: %s\r\n"
            "Connection: Close\r\n\r\n", 
            args->args[2], args->args[0]);
    }
    
    // 建立 Socket
    if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) return 2;
    
    // 設置非阻塞
    flags = fcntl(sockfd, F_GETFL, 0);
    fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
    
    // 非阻塞連接 + select 等待可寫 (連接建立)
    do {
        connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr));
        FD_ZERO(&wset);
        FD_SET(sockfd, &wset);
        
        if (select(sockfd + 1, NULL, &wset, NULL, &tv) <= 0 && cycles == 5) {
            close(sockfd);
            return 5;  // 連接超時
        }
        
        // 嘗試發送
        numbytes = send(sockfd, send_buf, strlen(send_buf), 0);
        if (numbytes < 0) {
            usleep(20000);  // 20ms 退讓
        }
        cycles++;
    } while (numbytes < 0 && cycles != 5);
    
    if (numbytes < 0) { close(sockfd); return 4; }  // 發送失敗
    
    close(sockfd);
    return 0;  // 成功
}

#endif /* HAVE_DLOPEN */

編譯與部署 (Linux)

# 編譯
gcc -Wall -I/usr/local/mysql/include/mysql -shared http_post_udf.c -o http_post_udf.so -fPIC

# 部署
cp http_post_udf.so /usr/local/mysql/lib/mysql/plugin/

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

# 測試 (需目標端監聽 3888 埠)
mysql> SELECT http_post('192.168.1.100', '3888', 'api/sync', 'test_data');
+--------------------------------------------------------------+
| http_post('192.168.1.100', '3888', 'api/sync', 'test_data') |
+--------------------------------------------------------------+
|                                                            0 |
+--------------------------------------------------------------+

關鍵技術點深度解析

技術點	實現細節	作用
非阻塞 Socket	`fcntl(sockfd, F_SETFL, flags \| O_NONBLOCK)`	connect/send 不阻塞線程，配合 select 多路復用
連接超時控制	`select(sockfd+1, NULL, &wset, NULL, &tv)` 監聽可寫事件	非阻塞 connect 後，可寫表示連接建立 (或失敗)，避免無限等待
發送重試機制	`do-while` 循環 + `usleep(20000)` + 最大 5 次	應對網絡抖動、接收端緩衝區滿導致的 EAGAIN/EWOULDBLOCK
靜態/動態緩衝區	`initid->ptr` 區分，<160KB 用靜態數組	滿足 UDF 禁 malloc 限制，高併發小包零分配，大包動態分配
HTTP 協議手工構造	`sprintf` 拼接 Request Line、Headers、Body	無 libcurl 依賴，完全可控，適合嵌入式/極簡環境
錯誤碼設計	0 成功，2 Socket 創建失敗，4 發送失敗，5 連接超時	便於上層業務區分失敗類型，針對性重試/告警

對比：libcurl vs 原生 Socket

維度	libcurl (multi)	原生 Socket + select
依賴	需安裝 libcurl-dev、運行時庫	零依賴，系統調用級
協議支援	HTTP/HTTPS/FTP/WS 等完整	僅手工實現的 HTTP 子集
HTTPS	原生支援 OpenSSL/mbedTLS	極難正確實現 (需引入 TLS 庫)
代碼複雜度	高級 API 封裝，邏輯清晰	手工處理協議細節，易出錯
效能	成熟優化，連接復用、管線化	單連接無復用，高頻場景需自行實現池化
適用場景	生產環境標準選擇	無法安裝第三方庫、極簡容器、教學演示

生產級避坑指南

DNS 解析阻塞：`inet_addr` 僅支援 IP。域名需 `getaddrinfo` (阻塞) 或預解析緩存 IP。
大數據分片發送：`send` 不保證一次發完，需循環發送剩餘數據 (代碼中已隱含在重試邏輯，但嚴謹寫法應處理部分發送)。
接收響應可選：若需讀取 HTTP Response，需 `select` 監聽可讀事件並 `recv` 循環讀取，當前代碼發送即關閉 (Connection: Close)。
信號處理：`send` 可能觸發 `SIGPIPE` (對端關閉)，需 `signal(SIGPIPE, SIG_IGN)` 或 `send(..., MSG_NOSIGNAL)`。
UDF 線程安全：MySQL 線程調用 UDF，靜態緩衝區 `static char buf[]` 非線程安全！**必須改用 `initid->ptr` 線程私有存儲**，或加鎖 (性能損)。本代碼靜態數組僅作演示，**生產嚴禁**。

總結

原生 Socket 實現的 UDF HTTP POST，展示了**非阻塞網絡編程在 MySQL 擴展中的核心模式**：`O_NONBLOCK` + `select` + 狀態機。但受限於 HTTPS 複雜度、線程安全、協議完整性，**生產環境強烈建議使用 libcurl**。本方案僅作為「零依賴、純 HTTP、內網可控」場景的備選技術儲備。