使用mysql_udf與curl庫完成http_post通信模塊示例

2020-01-18 23:21:41

字體：大中小

來源：轉載

供稿：網友

使用mysql_udf與curl庫完成http_post通信模塊（mysql_udf,multi_curl,http,post）

這個模塊其目前主要用于xoyo江湖的sns與kingsoft_xoyo自主研發的TCSQL數據庫做數據同步，當有feed插入sns數據庫，使用觸發器調用該模塊，向tcsql數據庫發送同步數據。也可以使用該模塊與其它使用socket接口的數據庫或程序做轉發與同步。
http_post模塊主要使用mysql_udf接口，與curl庫兩部分技術。
mysql_udf是mysql為c語言提供的一個接口，通過這個接口，用戶可以自定義mysql的函數，通過調用這些mysql函數，調用相應的c語言模塊來執行特定功能，實現mysql數據與外部應用的交互。curl庫是一個比較常用的應用層網絡協議庫，主要用到的是其中的curl_multi異步通信api，用來進行網絡傳輸。
首先參考mysql官方提供的udf_example.c文件，建立3個主要的接口函數，分別是初始化函數，執行函數與析構函數。

復制代碼代碼如下:

//args是sql語句傳回的參數，message是返回出錯信息使用這些都是規定好的。
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);
//args是sql語句傳回的參數，message是返回出錯信息,使用這些都是規定好的。
//初始化函數體 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);

在mysql_udf接口中，主函數體中是不允許使用new或malloc動態分配內存，所以如果需要申請內存空間，必須用xxxx_init()函數申請并將申請的地址賦給initid->ptr指針，然后在主函數體中使用，并在xxxx_deinit析構函數體中釋放。另外對于 mysql_udf接口的調用好像當并發量超過一定程度，如果是使用動態分配內存，會出現double free的錯誤，為了避免這個錯誤，所以在我的程序里使用靜態空間與動態申請空間相結合的方式，這樣如果數據較小，并發量較大，不會出現double free錯誤。對于靜態申請空間，最大約在160000~170000byte左右，我這里使用的160000，當mysql傳送的數據大于這個數的時候，才動態申請內存。初始化函數體如下：

復制代碼代碼如下:

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

      if(args->arg_count == 2 && 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;
        }
        else
        {
            initid->ptr=NULL;
        }

}
return 0;

}

其中http_post_init需要返回my_bool型。這個函數目的是給用戶提供一個方式，檢驗由mysql參數傳進來的數據是否正確，如果正確則返回0，則mysql會自動調用定義的主函數，如果返回1，則mysql打印message信息退出，不會調用主函數。所以在設定返回值的時候一定注意。

主函數如下：

復制代碼代碼如下:

longlong http_post( UDF_INIT *initid, UDF_ARGS *args,
                char *is_null __attribute__((unused)),
                char *error __attribute__((unused)))
{
    char* sendBuffer=NULL;
    CURL *curl;
    CURLM *multi_handle;
    int still_running;
    int times=0;//try times if select false
        int TRY_TIMES=25;
    struct timeval timeout;//set a suitable timeout to play around with
    timeout.tv_sec = 0;
    timeout.tv_usec = 100000;

    char sendArray[160000] = "/0";//can not move this into the if
    if(initid->ptr == NULL)
    {
        //char sendArray[160000] = "/0";//error
        sendBuffer=sendArray;
    }
    else
    {
        sendBuffer = initid->ptr;
        TRY_TIMES=100;
    }

    strcpy(sendBuffer,args->args[1]);
    curl = curl_easy_init();
    multi_handle = curl_multi_init();
    if(curl && multi_handle)
    {
        /* what URL that receives this POST */
        curl_easy_setopt(curl, CURLOPT_URL,args->args[0]);
        curl_easy_setopt(curl, CURLOPT_HTTPPOST, 1);
        curl_easy_setopt(curl,CURLOPT_POSTFIELDS,sendBuffer);
        curl_multi_add_handle(multi_handle, curl);
        while(CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi_handle,/ &still_running));
        while(still_running && times< TRY_TIMES)
         {
              int rc;      //select() return code
              int maxfd;
              fd_set fdread;
              fd_set fdwrite;
              fd_set fdexcep;
              FD_ZERO(&fdread);
              FD_ZERO(&fdwrite);
              FD_ZERO(&fdexcep);   //get file descriptors from the transfers
             curl_multi_fdset(multi_handle, &fdread, &fdwrite, &fdexcep,/ &maxfd);
             rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
             switch(rc)
            {
                case -1://select error
                      break;
                case 0:
                default:        // timeout
                     while(CURLM_CALL_MULTI_PERFORM !== curl_multi_perform(multi_handle, &still_running));
                     break;
             }
                times++;
         }//end while
       curl_multi_remove_handle(multi_handle,curl);
       curl_multi_cleanup(multi_handle);//always cleanup
       curl_easy_cleanup(curl);
       if(times>=TRY_TIMES)
       {
            return 1;
       }
        return 0;
}//end if
return 1;
}

在主函數中，主要使用curl庫進行通信，curl庫分成3部分，easy是同步模式，multi是異步模式，share模式是多線程共享數據的模式。

對于easy發送完數據后，會阻塞等待服務器的response，如果沒有返回，就會一直阻塞，當然可以設置一個timeout，但如果這個時間設小了，easy發送大數據的時候就會中斷，設太大了影響時間效率，另外當接收端不發送response的時候，easy庫即使發送完了數據，也會阻塞等待，有些時候對于發送端來講不需要等待接收端的respons，當發送完畢就可以結束了，這個時候easy就不適用。所以最后選擇multi庫。

如程序所示，首先得初始化，并設置easy句柄為post模式，指定需要post的數據，如下：

curl = curl_easy_init();

multi_handle = curl_multi_init();

curl_easy_setopt(curl, CURLOPT_URL,args->args[0]);

curl_easy_setopt(curl, CURLOPT_HTTPPOST, 1);

curl_easy_setopt(curl,CURLOPT_POSTFIELDS,sendBuffer);

由于要使用multi模式，必須也要初始化一個easy模式，并將這個easy模式的句柄放入所謂的multi函數執行棧：

curl_multi_add_handle(multi_handle, curl);

使用curl_multi_perform(multi_handle, &still_running)，來進行異步傳輸，但如果該函數返回的不是CURLM_CALL_MULTI_PERFORM，則需要重新執行。直到循環while(CURLM_CALL_MULTI_PERFORM == curl_multi_perform(multi_handle, &still_running));結束。此時如果剛才函數體中的still_running被置為1，表明連接建立，正在發送數據。需要配合select機制來進行數據發送。

函數 curl_multi_fdset(multi_handle, &fdread, &fdwrite, &fdexcep, &maxfd);會將最大的描述符寫入maxfd，

然后用select進行等待：rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);

最后如果select返回值不為-1（error）0（timeout）時候再次進行異步傳輸，即執行curl_multi_perform函數，直到

still_running為0，程序結束退出。

這里設置了一個最大執行次數的限制，如果服務器出現了問題，不能發送response，則still_running不會變為0，程序會死循環，

所以，設置一個最大循環次數TRY_TIMES，防止這種情況發生。但是這個次數設小了，數據可能沒有發送完，就退出了，如設置太大了，程序發送完了，服務器沒有response就會多執行多余循環。所以這個TRY_TIMES需要根據數據的大小和網絡狀況來設置，比正常

傳輸數據的次數略長。這里我小數據的時候循環設次數25，大數據循環設為100.

最后是析構函數體：

復制代碼代碼如下:

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

將初始化函數設置的內存釋放。

編譯執行過程如下：

將程序保存為http_post.c編譯如下(請根據機器上的mysql路徑進行調整)：

復制代碼代碼如下:

gcc -wall -I/usr/local/webserver/mysql/include/mysql/ -shared http_post.c -o http_post.so -fPIC
//使用mysql提供的頭文件生成動態鏈接庫
cp -f http_post.so /usr/local/webserver/mysql/lib/mysql/plugin/http_post.so
//將生成的.so文件放入mysql的plugin文件夾下
//進入mysql對動態鏈接庫中的函數進行安裝
cd /usr/local/webserver/mysql/bin/mysql
./mysql
//在mysql命令行下輸入如下命令：
mysql> DROP FUNCTION IF EXISTS http_post;
//其目的是如果系統內安裝了同名函數先進性drop。
mysql> CREATE FUNCTION http_post RETURNS INTEGER SONAME ‘http_post.so';
//生成http_post函數，并指明調用來源是http_post.so。
//最后調用函數，其目的是向指定ip和端口發送post數據。調用前先打開指定ip主機上的網絡調試助手，并監聽3888端。
mysql> select http_post(‘testpost.com/index.php','sfasfa');

在網絡助手中可以看到如下結果：

上一篇：mysql的udf編程之非阻塞超時重傳

下一篇：Windows下修改mysql的data文件夾存放位置的方法