2. 程式人生 > >fastdfs上傳

fastdfs上傳

阿新 • • 發佈：2019-01-08

fastdfs是一個輕量級的分散式檔案系統，主要由 tracker server， storage server 以及client組成，這裡主要涉及兩點 ： 
1）客戶端上傳檔案流程和協議分析 
2）實現一個簡單的檔案上傳函式

一: 檔案上傳的基本流程

fastdfs中上傳一個檔案，主要涉及以下幾個步驟:

1）上傳連線請求，客戶端會向tracker server發出上傳檔案的請求 
2）tracker收到請求後，返回storage server的ip和埠 
3）客戶端連線storage,並且上傳檔案 
4）檔案上傳完成後，storage返回路徑資訊

以下具體分析檔案上傳過程中的協議和各種操作

fastdfs協議頭部:

typedef struct
    {
        char pkg_len[FDFS_PROTO_PKG_LEN_SIZE];  //body length, not including header(8個位元組)
        char cmd;    //command code     TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE
        char status; //status code for response
    } TrackerHeader;
1
2
3
4
5
6
fastdfs協議的頭部是由10個位元組大小的結構體構成, 
傳送:傳送資料時，先發送TrackerHeader到伺服器，隨後傳送具體的資料 
接受:接受資料時，先接受sizeof(TrackerHeader)大小的報文頭部，隨後接受pkg_len長度的報文體

status: 傳送的時候設定為0 
cmd: 命令 
pkg_len：一個int64_t的整型，除去TrackerHeader長度的報文長度

二: 客戶端向tracker server傳送獲取storage地址請求 
#define TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE 101

//            協議頭
//            pkg_len | cmd     | status
//            8 bytes | 1 bytes | 1 bytes
//向tracker server請求storage server cmd
#define TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE     101

    TrackerHeader header;//協議頭部
    memset(&header, 0, sizeof(TrackerHeader));
    header.cmd = TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE;

    //向tracker server 請求 storage,tcpsenddata 返回非0，表示傳送成功
    if(tcpsenddata(sockfd, &header, sizeof(TrackerHeader), 10, &count) != 0)
    {
        fprintf(stderr, "tcpsenddata error: %s\n", strerror(errno));
        return 1;
    }
    else//請求傳送成功，等待tracker回覆
    {
        //接收頭部，頭部是一個TrackerHeader型別，10個位元組
        TrackerHeader resp;
        if((ret_code = tcprecvdata(sockfd, &resp, sizeof(TrackerHeader), 10, &count)) != 0) 
        {
            fprintf(stderr, "tcprecvdata error: %s\n", strerror(ret_code));
            return 1;
        }
        //開始接收報文體
        //int64_t read_int64(const char *buff)
        //{
        //      unsigned char *p;
        //      p = (unsigned char *)buff;
        //      return  (((int64_t)(*p)) << 56) | \
        //      (((int64_t)(*(p+1))) << 48) |  \
        //      (((int64_t)(*(p+2))) << 40) |  \
        //      (((int64_t)(*(p+3))) << 32) |  \
        //      (((int64_t)(*(p+4))) << 24) |  \
        //      (((int64_t)(*(p+5))) << 16) |  \
        //      (((int64_t)(*(p+6))) << 8) | \
        //      ((int64_t)(*(p+7)));
        //}
        int size = read_int64(resp.pkg_len);//獲取報體長度
        char *buf = (char*)calloc(size + 1, sizeof(char));

        if((ret_code = tcprecvdata(sockfd, buf, size, 10, &count) != 0)) 
        {
            fprintf(stderr, "tcprecvdata error: %s\n", strerror(ret_code));
            return 1;
        }

        //  報文體
        //  group_name  |ip         |port       |storage_index
        //  16 bytes    |16 bytes   |8 bytes    |
        //#define TRACKER_QUERY_STORAGE_STORE_BODY_LEN 40
        if(count != TRACKER_QUERY_STORAGE_STORE_BODY_LEN)
        {
            fprintf(stderr, "invalid message");
            return 1;
        }
        //group name    
        //#define FDFS_GROUP_NAME_MAX_LEN  16
        char group_name[FDFS_GROUP_NAME_MAX_LEN + 1]  = {0};
        memcpy(group_name, buf, FDFS_GROUP_NAME_MAX_LEN);
        group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';
        //ip: port
        //#define IP_ADDRESS_SIZE 16 
        //port:8 bytes
        char ip[IP_ADDRESS_SIZE + 1] = {0};
        memcpy(ip, buf + FDFS_GROUP_NAME_MAX_LEN, IP_ADDRESS_SIZE - 1);
        char szPort[8] = {0};
        memcpy(szPort, buf + FDFS_GROUP_NAME_MAX_LEN + IP_ADDRESS_SIZE - 1, 8);
        ip[IP_ADDRESS_SIZE] = '\0';
        int port = read_int64(szPort);
        //storage index;
        char *storage_index = buf + FDFS_GROUP_NAME_MAX_LEN + IP_ADDRESS_SIZE - 1 + FDFS_PROTO_PKG_LEN_SIZE;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
三:以上步驟完成後，獲取storage的ip 和 port後，就可以上傳檔案了

在官方的客戶端中，檔案操作有upload，download, append,delete等，這裡只涉及upload 
上傳檔案中，官方給出了三種方式 
1)通過buffer上傳，即將檔案讀取進記憶體，然後在傳送 
2)使用sendfile，sendfile是Linux提供的一個庫函式 
3)通過回撥函式的方式

這裡主要涉及的是第一種，通過buffer上傳的方式

檔案上傳協議:

//檔案上傳協議頭部
    10 bytes        | 1 bytes        | 8 bytes    | 6 bytes        |
    TrackerHeader   | storage_index  | 檔案長度    | 檔名或者全為0)  |
//storage_index 是客戶端向tracker server申請storage index時候返回的結果
//檔名 如果不為空，那麼取前6位，或者可以全部設定為0

//上傳完成 storage回覆客戶端協議
10 bytes      | 16 bytes    | TrackerHeader.pkg_len - 16bytes 
TrackerHeader | groupname   | remote file name
1
2
3
4
5
6
7
8
9
10
11
12
void uploadfile(int sockfd, const char *filepath, char *storage_index) 
{
    char out_buf[512];
    TrackerHeader *pHeader;
    char *p = out_buf;
    char *buf = NULL;

    //TrackerHeader 10 bytes
    //檔案上傳協議頭部
    //10 bytes        | 1 bytes        | 8 bytes    | 6 bytes        |
    //TrackerHeader   | storage_index  | 檔案長度    | 檔名或者全為0)  |
    pHeader = (TrackerHeader*)out_buf;
    p += sizeof(TrackerHeader);

    //storage index  1 bytes
    *p++ = *storage_index;
    //filesize 8bytes
    long int filesize = 0;
    int ret = 0;
    //讀取檔案到buf,並且返回檔案長度 filesize
    if((ret = getfilebuf(&buf, &filesize, filepath) != 0))
    {
        fprintf(stderr, "getfilebuf failed: %s\n", strerror(ret));
        return;
    }
    //void write_int64(int64_t n, char *buff)
    //{
    //  unsigned char *p;
    //  p = (unsigned char *)buff;
    //  *p++ = (n >> 56) & 0xFF;
    //  *p++ = (n >> 48) & 0xFF;
    //  *p++ = (n >> 40) & 0xFF;
    //  *p++ = (n >> 32) & 0xFF;
    //  *p++ = (n >> 24) & 0xFF;
    //  *p++ = (n >> 16) & 0xFF;
    //  *p++ = (n >> 8) & 0xFF;
    //  *p++ = n & 0xFF;
    //}
    write_int64(filesize, p);
    //#define FDFS_PROTO_PKG_LEN_SIZE 8
    p += FDFS_PROTO_PKG_LEN_SIZE;

    //ext_name
    //#define FDFS_FILE_EXT_NAME_MAX_LEN 6
    memset(p, 0, FDFS_FILE_EXT_NAME_MAX_LEN);
    p += FDFS_FILE_EXT_NAME_MAX_LEN;

    //set TrackerHeader
    write_int64(p - out_buf + filesize - sizeof(TrackerHeader), pHeader->pkg_len);
    //#define STORAGE_PROTO_CMD_UPLOAD_FILE 11
    pHeader->cmd  = STORAGE_PROTO_CMD_UPLOAD_FILE;
    pHeader->status = 0;

    //傳送報文頭部
    int count;
    int ret_code = 0;
    if((ret_code = tcpsenddata(sockfd, out_buf, p - out_buf, 10, &count) != 0)) {
        fprintf(stderr, "tcpsenddata failed: %s\n", strerror(errno));
        return;
    }
    //傳送報文體，具體檔案資料
    if((ret_code = tcpsenddata(sockfd, buf, filesize, 10, &count)) != 0) {
        fprintf(stderr, "tcpsenddata body failed: %s\n", strerror(errno));
        return;
    }

    //接收storage server回覆
    //上傳完成 storage回覆客戶端協議
    //10 bytes      | 16 bytes    | TrackerHeader.pkg_len - 16bytes 
    //TrackerHeader | groupname   | remote file name
    TrackerHeader resp;
    if((ret_code = tcprecvdata(sockfd, &resp, sizeof(TrackerHeader), 1000, &count)) != 0) {
        fprintf(stderr, "tcprecvdata failed: %s\n", strerror(ret_code));
        return;
    }   
    if(count != sizeof(TrackerHeader)) {
        fprintf(stderr, "invalid header");
        return;
    }

    int64_t bodylen = read_int64(resp.pkg_len);
    //接收報文體
    char *in_buf = (char*)calloc(bodylen + 1, sizeof(char));
    if((ret_code = tcprecvdata(sockfd, in_buf, bodylen, 10, &count)) != 0) 
    {
        fprintf(stderr, "read body failed: %s\n", strerror(ret_code));
        return;
    }

    //groupname
    //#define FDFS_GROUP_NAME_MAX_LEN  16
    char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
    memcpy(group_name, in_buf, FDFS_GROUP_NAME_MAX_LEN);
    group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';

    //remote filename
    char remote_filename[bodylen - FDFS_GROUP_NAME_MAX_LEN + 1];
    memcpy(remote_filename, in_buf + FDFS_GROUP_NAME_MAX_LEN, bodylen - FDFS_GROUP_NAME_MAX_LEN + 1);

    cout << "groupname: " << group_name << endl;
    cout << "remote_filename: " << remote_filename << endl;

    char httpaddr[128] = {0};
    sprintf(httpaddr, "http://106.75.129.177:8080/%s/%s", group_name, remote_filename);
    cout << "httpaddr: " <<  httpaddr << endl;//http地址
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
以下附上完整程式碼， ubuntu14位， 編譯器 g++,測試已通過

#include <iostream>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
using namespace std;
#define FDFS_GROUP_NAME_MAX_LEN     16
#define FDFS_PROTO_PKG_LEN_SIZE     8
#define IP_ADDRESS_SIZE         16
//cmd
#define TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE     101
#define STORAGE_PROTO_CMD_UPLOAD_FILE       11

#define TRACKER_QUERY_STORAGE_STORE_BODY_LEN    (FDFS_GROUP_NAME_MAX_LEN \
    + IP_ADDRESS_SIZE - 1 + FDFS_PROTO_PKG_LEN_SIZE + 1)
#define FDFS_FILE_EXT_NAME_MAX_LEN 6
typedef struct {
    char pkg_len[FDFS_PROTO_PKG_LEN_SIZE];
    char cmd;
    char status;
}TrackerHeader;

//set socketfd nonblocking
int setnonblocking(int sockfd);
int tcprecvdata(int sockfd, void *data, const int size,\
    const int timeout_ms, int *count);
int tcpsenddata(int sockfd, void *data, const int size,\
const int timeout_ms, int *count);

int64_t read_int64(const char* buf);
void write_int64(int64_t n, char* buf);

void uploadfile(int sockfd, const char *filepath, char *storage_index);
int getfilebuf(char **buf, long int *filesize, const char* filepath);
//apply storage address from tracker server
int main() {
    const char *ip = "127.0.0.1";
    uint16_t port = 22122;
    int ret_code = 0;
    int sockfd = -1;
    int count = 0;
    //connect tracker server
    if((sockfd = socket(PF_INET, SOCK_STREAM, 0)) < 0) 
    {
        fprintf(stderr, "socket errnor: %s\n", strerror(errno));
        return 1;
    }
    if((ret_code = setnonblocking(sockfd)) != 0) 
    {
        fprintf(stderr, "setnonblocking error: %s\n", strerror(ret_code));
        return 1;
    }
    struct sockaddr_in addr;
    addr.sin_addr.s_addr = inet_addr(ip);
    addr.sin_port = htons(port);
    addr.sin_family = AF_INET;

    socklen_t len = sizeof(struct sockaddr);
    if(connect(sockfd, (struct sockaddr*)&addr, len) < 0) 
    {
        fprintf(stderr, "connect error: %s\n", strerror(errno));
        return 1;
    }

    TrackerHeader header;
    memset(&header, 0, sizeof(TrackerHeader));
    header.cmd = TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE;

    if(tcpsenddata(sockfd, &header, sizeof(TrackerHeader), 10, &count) != 0)
    {
        fprintf(stderr, "tcpsenddata error: %s\n", strerror(errno));
        return 1;
    }
    else
    {
        //recv header
        TrackerHeader resp;
        if((ret_code = tcprecvdata(sockfd, &resp, sizeof(TrackerHeader), 10, &count)) != 0) 
        {
            fprintf(stderr, "tcprecvdata error: %s\n", strerror(ret_code));
            return 1;
        }

        cout << "recv header: " << count << endl;

        //read body;
        int size = read_int64(resp.pkg_len);
        char *buf = (char*)calloc(size + 1, sizeof(char));

        if((ret_code = tcprecvdata(sockfd, buf, size, 10, &count) != 0)) 
        {
            fprintf(stderr, "tcprecvdata error: %s\n", strerror(ret_code));
            return 1;
        }

        //body
        //  group_name  |ip     |port       |storage_index
        //  16bytes     |16bytes    |8bytes     |
        cout << "read body: " << count << endl;
        if(count != TRACKER_QUERY_STORAGE_STORE_BODY_LEN)
        {
            fprintf(stderr, "invalid message");
            return 1;
        }
        //group name    
        char group_name[FDFS_GROUP_NAME_MAX_LEN + 1]  = {0};
        memcpy(group_name, buf, FDFS_GROUP_NAME_MAX_LEN);
        group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';
        cout << "group name: " << group_name << endl;
        //ip: port
        char ip[IP_ADDRESS_SIZE + 1] = {0};
        memcpy(ip, buf + FDFS_GROUP_NAME_MAX_LEN, IP_ADDRESS_SIZE - 1);
        char szPort[8] = {0};
        memcpy(szPort, buf + FDFS_GROUP_NAME_MAX_LEN + IP_ADDRESS_SIZE - 1, 8);
        ip[IP_ADDRESS_SIZE] = '\0';
        int port = read_int64(szPort);
        cout << "address: " << ip << ":" << port << endl;
        //storage index;
        char *storage_index = buf + FDFS_GROUP_NAME_MAX_LEN + IP_ADDRESS_SIZE - 1 + FDFS_PROTO_PKG_LEN_SIZE;
        cout << "storage_index: " << storage_index << endl;
        free(buf);

        //connect storage server
        sockaddr_in st_addr;
        st_addr.sin_addr.s_addr = inet_addr(ip);
        st_addr.sin_family = AF_INET;
        st_addr.sin_port = htons(port);

        int storage_fd = socket(AF_INET, SOCK_STREAM, 0);
        if(storage_fd < 0) {
            fprintf(stderr, "socket failed: %s\n", strerror(errno));
            return 1;
        }
        socklen_t len2 = sizeof(sockaddr_in);
        if(connect(storage_fd, (struct sockaddr*)&st_addr, len2) < 0) {
            fprintf(stderr, "connect failed: %s\n", strerror(errno));
            return 1;
        }

        uploadfile(storage_fd, "1.jpg", storage_index);
    }
    return 0;
}

int getfilebuf(char **buf, long int *filesize, const char *filepath) 
{
    int ret_code = 0;
    FILE *fp = fopen(filepath, "rb+");
    if(fp == NULL)
    {
        ret_code = errno;
        return ret_code;
    }
    //get filesize;
    fseek(fp, 0, SEEK_END);
    *filesize = ftell(fp);
    fseek(fp, 0, SEEK_SET);
    cout << "get filesize: " <<*filesize << endl;
    //malloc buf
    *buf = (char*)calloc(*filesize + 1, sizeof(char));
    if(*buf == NULL) {
        ret_code = errno;
        return ret_code;
    }

    int read_bytes = 0;
    int left_bytes = *filesize;
    char *p = *buf;
    while(left_bytes > 0) {
        read_bytes = fread(p, sizeof(char), left_bytes, fp);
        left_bytes -= read_bytes;
        p += read_bytes;
    }   
    return ret_code;
}
void uploadfile(int sockfd, const char *filepath, char *storage_index) 
{
    char out_buf[512];
    TrackerHeader *pHeader;
    char *p = out_buf;
    char *buf = NULL;

    //TrackerHeader 10 bytes
    pHeader = (TrackerHeader*)out_buf;
    p += sizeof(TrackerHeader);

    //storage index  1 bytes
    *p++ = *storage_index;
    //filesize 8bytes
    long int filesize = 0;
    int ret = 0;
    if((ret = getfilebuf(&buf, &filesize, filepath) != 0))
    {
        fprintf(stderr, "getfilebuf failed: %s\n", strerror(ret));
        return;
    }
    printf("filesize: %ld\n", filesize);
    write_int64(filesize, p);
    p += FDFS_PROTO_PKG_LEN_SIZE;

    //ext_name
    memset(p, 0, FDFS_FILE_EXT_NAME_MAX_LEN);
    p += FDFS_FILE_EXT_NAME_MAX_LEN;

    //set TrackerHeader
    write_int64(p - out_buf + filesize - sizeof(TrackerHeader), pHeader->pkg_len);
    pHeader->cmd  = STORAGE_PROTO_CMD_UPLOAD_FILE;
    pHeader->status = 0;

    //send header
    int count;
    int ret_code = 0;
    if((ret_code = tcpsenddata(sockfd, out_buf, p - out_buf, 10, &count) != 0)) {
        fprintf(stderr, "tcpsenddata failed: %s\n", strerror(errno));
        return;
    }
    //send body
    if((ret_code = tcpsenddata(sockfd, buf, filesize, 10, &count)) != 0) {
        fprintf(stderr, "tcpsenddata body failed: %s\n", strerror(errno));
        return;
    }

    //recv response
    TrackerHeader resp;
    if((ret_code = tcprecvdata(sockfd, &resp, sizeof(TrackerHeader), 1000, &count)) != 0) {
        fprintf(stderr, "tcprecvdata failed: %s\n", strerror(ret_code));
        return;
    }   
    if(count != sizeof(TrackerHeader)) {
        fprintf(stderr, "invalid header");
        return;
    }

    int64_t bodylen = read_int64(resp.pkg_len);

    char *in_buf = (char*)calloc(bodylen + 1, sizeof(char));
    if((ret_code = tcprecvdata(sockfd, in_buf, bodylen, 10, &count)) != 0) 
    {
        fprintf(stderr, "read body failed: %s\n", strerror(ret_code));
        return;
    }

    //groupname
    char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
    memcpy(group_name, in_buf, FDFS_GROUP_NAME_MAX_LEN);
    group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';

    //remote filename
    char remote_filename[bodylen - FDFS_GROUP_NAME_MAX_LEN + 1];
    memcpy(remote_filename, in_buf + FDFS_GROUP_NAME_MAX_LEN, bodylen - FDFS_GROUP_NAME_MAX_LEN + 1);

    cout << "groupname: " << group_name << endl;
    cout << "remote_filename: " << remote_filename << endl;

    char httpaddr[128] = {0};
    sprintf(httpaddr, "http://127.0.0.1:8080/%s/%s", group_name, remote_filename);
    cout << "httpaddr" <<  httpaddr << endl;
}

void write_int64(int64_t n, char *buff)
{
    unsigned char *p;
    p = (unsigned char *)buff;
    *p++ = (n >> 56) & 0xFF;
    *p++ = (n >> 48) & 0xFF;
    *p++ = (n >> 40) & 0xFF;
    *p++ = (n >> 32) & 0xFF;
    *p++ = (n >> 24) & 0xFF;
    *p++ = (n >> 16) & 0xFF;
    *p++ = (n >> 8) & 0xFF;
    *p++ = n & 0xFF;
}
int64_t read_int64(const char *buff)
{
    unsigned char *p;
    p = (unsigned char *)buff;
    return  (((int64_t)(*p)) << 56) | \
        (((int64_t)(*(p+1))) << 48) |  \
        (((int64_t)(*(p+2))) << 40) |  \
        (((int64_t)(*(p+3))) << 32) |  \
        (((int64_t)(*(p+4))) << 24) |  \
        (((int64_t)(*(p+5))) << 16) |  \
        (((int64_t)(*(p+6))) << 8) | \
        ((int64_t)(*(p+7)));
}

int setnonblocking(int sockfd)
{
    int ret_code = 0;
    if(fcntl(sockfd, F_SETFD, O_NONBLOCK) < 0) {
        ret_code = errno;
    }
    return ret_code;
}
int tcpsenddata(int sockfd, void *data, const int size,\
const int timeout_ms, int *count) 
{
    int left_bytes = size;
    int write_bytes = 0;
    int ret_code = 0;
    int res = 0;
    char *p = (char*)data;

    fd_set rfds;
    FD_ZERO(&rfds);
    FD_SET(sockfd, &rfds);
    while(left_bytes > 0) {
        write_bytes = send(sockfd, p, left_bytes, 0);
        if(write_bytes > 0) 
        {
            left_bytes -= write_bytes;
            p += write_bytes;
            continue;
        }
        else if(write_bytes < 0) 
        {
            if(!(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK))
            {
                ret_code = errno == 0 ? errno : EINTR;
                break;
            }

        }
        else
        {
            ret_code = ENOTCONN;
            break;
        }

        if(timeout_ms <= 0) 
        {
            res = select(sockfd + 1, &rfds, NULL, NULL, NULL); 
        }
        else 
        {
            struct timeval tv;
            tv.tv_usec = timeout_ms;
            tv.tv_sec = 0;
            res = select(sockfd + 1, &rfds, NULL, NULL, &tv);
        }
        if(res == 0)
        {
            ret_code = ETIMEDOUT;
            break;
        }
        if(res < 0) 
        {
            if(errno == EINTR)
            {
                continue;
            }
            ret_code = errno == 0 ? errno : EINTR;
        }
    }
    if(count != NULL)
    {
        *count = size - left_bytes;
    }
    return ret_code;
}

int tcprecvdata(int sockfd, void *data, const int size,\
    const int timeout_ms, int *count) {

    int left_bytes = size;
    int read_bytes = 0;
    int ret_code = 0;
    int res = 0;
    char *p = (char*)data;

    fd_set rfds;
    FD_ZERO(&rfds);
    FD_SET(sockfd, &rfds);

    while(left_bytes > 0) {
        read_bytes = recv(sockfd, p, left_bytes, 0);
        if(read_bytes > 0) 
        {
            left_bytes -= read_bytes;
            p += read_bytes;
            continue;
        } 
        else if(read_bytes < 0) 
        {
            if(!(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) 
            {
                ret_code = errno != 0 ? errno : EINTR;
                break;  
            }
        }
        else 
        {
            ret_code = ENOTCONN;
            break;
        }

        if(timeout_ms <= 0) 
        {
            res = select(sockfd + 1, &rfds, NULL, NULL, NULL);
        }   
        else 
        {
            struct timeval tv;
            tv.tv_usec = timeout_ms;
            tv.tv_sec  = 0;
            res = select(sockfd + 1, &rfds, NULL, NULL, &tv);
        }

        if(res == 0) 
        {
            ret_code = ETIMEDOUT;
            break;
        }
        if(res < 0)
        {
            if(errno == EINTR)
            {
                continue;
            }
            ret_code = errno == 0 ? errno : EINTR;
            break;          
        }
    }

    if(count != NULL)
    {
        *count = size - left_bytes;
    }
    return ret_code;
--------------------- 
作者：Ada_baby 
來源：CSDN 
原文：https://blog.csdn.net/Ada_baby/article/details/51274157 
版權宣告：本文為博主原創文章，轉載請附上博文連結！

相關推薦

Fastdfs上傳失敗 tracker_query_storage fail

本地上傳失敗 [[email protected] ~]# fdfs_test /etc/fdfs/client.conf.sample upload /tmp/test.jpg This is FastDFS client test program v5.09 Copyrig

使用java操作FastDFS上傳,下載,刪除檔案

經過我的測試，直接把程式碼copy就能用了！ 前提條件：將原始碼生成為jar包,install到本地maven倉庫 1.從github上down下原始碼：https://github.com/happyfish100/fastdfs-client-java在myeclipse中建立maven專案

分享知識-快樂自己：FastDFS 上傳 java 原始碼

FastDFS 上傳 java 原始碼：點我下載原始碼 首先匯入 POM 檔案：解決 maven 不能下載 fastdfs-client-java JAR <dependency> <groupId>org.csource</groupId>

fastdfs上傳

fastdfs是一個輕量級的分散式檔案系統，主要由 tracker server， storage server 以及client組成，這裡主要涉及兩點 ：  1）客戶端上傳檔案流程和協議分析  2）實現一個簡單的檔案上傳函式 一: 檔案上傳的基本流程 fastdfs中上

解決:使用java連線Fastdfs,上傳檔案時報:SocketTimeOutException的問題.

最近研究了下分散式儲存Fastdfs,在centOS上配置完後,使用centOS或putty連線並上傳圖片,然後用瀏覽器讀取storage server返回的URL,一切正常. 但是,使用eclipse上傳圖片時,報錯:SocketTimeOutException. 報錯

FastDFS上傳簡單Demo

  FastDFS是一個輕量級的分散式檔案系統，它對檔案的管理功能包括：檔案儲存、檔案同步、檔案訪問等，解決了大容量儲存和負載均衡的問題。下面是Demo的程式碼，採用的是Spring MVC框架。 開發工具：Eclipse jar包管理：Maven JDK：1.8 1、

（轉） 淘淘商城系列——使用FastDFS-Client客戶端進行上傳圖片的測試

row 構造方法 無法 空間 依賴 ron 文件下載 信息 utils http://blog.csdn.net/yerenyuan_pku/article/details/72804018 不久之前，我們實現了商品的類目選擇這個功能，但這只是萬裏長征的第一步，我們還有很

【轉載】【JAVA秒會技術之圖片上傳】基於Nginx及FastDFS，完成圖片的上傳及展示

相互 沒有 con 性能 ext 存儲服務器 網絡 管理 代理配置 基於Nginx及FastDFS，完成商品圖片的上傳及展示 一、傳統圖片存儲及展示方式 存在問題： 1）大並發量上傳訪問圖片時，需要對web應用做負載均衡，但是會存在圖片共享問題 2）web應

FastDFS 工具類實現文件上傳_02

works storage scl imp pre nec style ret ima 一、jar 包　 　　jar包下載：https://pan.baidu.com/s/1nwkAHU5 密碼：tlv6 　　或者 　　下載工程，安裝到 maven 本地倉庫 　　工程下載：

FastDFS重啟後上傳文件報錯的解決方案

fastdfs報錯：ERROR - file: tracker_proto.c, line: 48, server: 109.24.249.40:22122, response status 2 != 0tracker_query_storage fail, error no: 2, error info:

fastDFS與java整合文件上傳下載

snap spa generate mapping oba exe try file getclass 準備 下載fastdfs-client-java源碼 源碼地址 密碼：s3sw 修改pom.xml 第一個plugins是必需要的，是maven用來編譯的插件，第二

FastDFS初步認識--上傳下載流程介紹

冗余 size ID server 存在 上傳文件 都是 客戶端訪問 認識 什麽是FastDFS FastDFS 是用 c 語言編寫的一款開源的分布式文件系統。FastDFS 為互聯網量身定制，充分考慮了冗余備份、負載均衡、線性擴容等機制，並註重高可用、高性能等指標，使用

分布式搭建-簡易版文件上傳下載服務器FastDFS

centos 提取 新的 一份 xms x64 子目錄 archive 加權 一、FastDFS介紹 FastDFS開源地址：https://github.com/happyfish100 參考：分布式文件系統FastDFS設計原理 參考：FastDFS分布式文件系統 個

一次替換FastDFS數據目錄引發的文件上傳和訪問異常

org prot 因此 0.10 51cto 找到 tor copy name 1、圖片訪問異常 問題描述 搭建一臺新環境的FastDFS文件服務器，剛搭建好的時候，上傳了一張圖片，正常，但是因為當時端口沒有開，沒有驗證訪問的問題。環境暫時擱置等待測試。後來在測試環節，因為

Java Springboot結合FastDFS實現檔案上傳以及根據圖片url將圖片上傳至圖片伺服器

上一篇文章我們已經講解了如何搭建FastDFS圖片伺服器，環境我們準備好了現在就讓我們開始與Java結合將他應用到實際的專案中吧。本篇文章我們將會展示上傳圖片到FastDFS圖片伺服器以及通過外網的圖片url將圖片上傳至我們自己的圖片伺服器中。 1.建立springbo

FastDFS的配置、部署與API使用解讀（8）FastDFS多種檔案上傳介面詳解

《Poechant的FastDFS的配置、部署與API使用教程》 FastDFS的配置、部署與API使用解讀——（1）入門使用教程 FastDFS的配置、部署與API使用解讀——（2）Java API：根據InputStream、檔名、檔案長度上傳檔案 FastDFS的配置、部署與API

FastDFS：Java客戶都實現檔案的上傳、下載、修改、刪除

客戶端版本：fastdfs_client_v1.24.jar   配置檔案 connect_timeout = 200 network_timeout = 3000 charset = UTF-8 http.tracker_http_port = 8080 http.anti_steal

《商城專案03》--用FastDFS實現圖片上傳

一, 圖片伺服器的下載安裝配置 1, 下載資源-image-server.7z 連結： https://pan.baidu.com/s/1jtfpBtlocrhfmdFb5q3nCg 提取碼：0hjw 2, 下載解壓至自定義位置 3, 雙擊開啟Redis_02.v

fastDFS+LibreOffice多檔案上傳(二)前端部分：修改上傳按鈕樣式和上傳後在頁面顯示

一：修改上傳按鈕樣式 原來的<input type="file"/>太醜了，所以我自己設定了樣式。方法：在<input type="file"/>上面加一層div，遮住它。 .upload{ padding: 0px 25px; line-height

fastDFS+LibreOffice多檔案上傳(二)後端部分：檔案資訊轉json字串儲存資料庫(Gson和org.json兩種方式)

需要注意的地方： 1）如果你複製我的程式碼到你的程式上報錯，可以看看我第一篇文章實體類跟配置檔案的設定：https://blog.csdn.net/qq_36688143/article/details/84162924 第二篇檔案上傳前端頁面的程式碼： https://blog.c