第7课:fastDFS对应的API

本api使用手册是基于循环处理在线请求的应用场景考虑的,所以有continue或break用来表示本个请求处理结束,继续处理后面的请求;
个人可以根据情况可以设计成tracker和storage短连接的简单方式,也可设计为storage连接池的方式,步骤都一样

文章下面给出了php api的封装,并提供了测试文件,可以测试上传下载删除判断等各个功能
文章下面给出了php api的封装,并提供了测试文件,可以测试上传下载删除判断等各个功能
文章下面给出了php api的封装,并提供了测试文件,可以测试上传下载删除判断等各个功能
重要的事情说三遍!

FastDFS api使用手册之文件上传,其余部分流程类似:

1.1  配置client.conf

1.2  初始化并传入配置文件路径
这一步在程序start时完成,参见附件fdfs_load_test.cpp最后的StartApp().
if ((result=fdfs_client_init(conf_filename)) != 0)
{
    return result;
}

1.3  获取tracker连接
获取跟tracker的连接,只要一个tracker可用,即能获取到连接。
自从FastDFSv1.22开始支持api多线程安全,多线程环境使用api主要是获取tracker连接时不同。多线程的客户端使用这个函数tracker_get_connection_r(),避免各个并发之间互相影响,其余使用tracker_get_connection()。

//为了避免频繁重连,只要不出错,就只连tracker和storage一次        
pTrackerServer = tracker_get_connection();
if (pTrackerServer == NULL)
//tracker_get_connection_r(pTrackerServer);   //多线程安全函数
//if (pTrackerServer->sock <= 0)
{
	//连不上tracker主备机,本次请求处理失败,给用户出错信息,然后继续处理以后的请求,处理以后的请求前调用continue去重连tracker,等tracker只要有一台恢复正常后恢复业务。
	sleep(1);                //可以间隔多少毫秒再试
	continue;
}

1.4询问tracker可用的storage
获取tracker连接后向该tracker查询当前哪个storage可用,使用tracker_query_storage_store函数。
参数store_path_index用来支持1个storage支持多个磁盘分区,返回的值用在下面的函数调用中,storageServer变量返回可用storage的group、ip和port:
store_path_index = 0;
if ((result=tracker_query_storage_store(pTrackerServer, \
		&storageServer, &store_path_index)) != 0) {
	//表示本tracker异常或没有可用storage,本次请求处理失败,给用户出错信息,但继续处理以后的请求,然后返回到step3去重新获取tracker:
	fdfs_quit(pTrackerServer);
	tracker_disconnect_server(pTrackerServer);
	sleep(1);                //可以间隔多少毫秒再试
	continue;
}

1.5连接该storage
根据返回的storage ip等信息连接该storage,并发的程序,每个cgi进程或每个线程可以重复使用storage连接直到连接异常:
if ((result=tracker_connect_server(&storageServer)) != 0) {
	//tracker给了一个连不上的storage,本次请求处理失败,给用户出错信息,重连tracker。
	fdfs_quit(pTrackerServer);
	tracker_disconnect_server(pTrackerServer);
	sleep(1);
	continue;
}

Fdfs_connpool_test.cpp示例了用连接池,适合后台批量导入文件的程序,用数组即可实现连接池,因为每个线程或进程使用的storage连接数最大为所有group的storage总的部署数目,数组的size即等于这个数接口;
查找时根据返回的storage ip等信息遍历这个小数组,比较ip和端口等是否相同,没有找到即连接该storage,连接好后放到连接池中。

1.6上传文件
上传文件可分3种情况,多线程的例子只示例了1种情况的上传,更全的请参见作者自带的例子或fdfs_simple_test.c。
出错处理在3种情况调用的后面。
因为需要专门的db来保存返回的文件索引和应用的信息,可以不用Metadata,下面两个参数meta_list, meta_count填NULL和0即可。

//文件后缀名,FDFS_UPLOAD_BY_FILE分支时这样使用,FDFS_UPLOAD_BY_BUFF或另一种类时直接给file_ext_name赋值,例如赋值”jpeg”
file_ext_name = strrchr(local_filename, '.');
if (file_ext_name != NULL)
{
	file_ext_name++;
}

//这个参数填””
strcpy(group_name, "");
//这里是分类上传,具体应用时选择一类即可,第一类是文件已经存放在客户端本次磁盘了,指定文件名(含路径)上传。
if (upload_type == FDFS_UPLOAD_BY_FILE)
{
	result = storage_upload_by_filename(pTrackerServer, \
		&storageServer, store_path_index, \
		local_filename, file_ext_name, \
		meta_list, meta_count, \
		group_name, remote_filename);

	printf("storage_upload_by_filename\n");
}

//第二类是完整的文件已由用户上传到web程序内存中的file_content这个buffer内,file_size参数指明buffer内收到的文件长度。
else if (upload_type == FDFS_UPLOAD_BY_BUFF)
{
	char *file_content;
	if ((result=getFileContent(local_filename, \
			&file_content, &file_size)) == 0)
	{
	result = storage_upload_by_filebuff(pTrackerServer, \
		&storageServer, store_path_index, \
		file_content, file_size, file_ext_name, \
		meta_list, meta_count, \
		group_name, remote_filename);
	free(file_content);
	}

	printf("storage_upload_by_filebuff\n");
}

//第三类跟第一类类似,也需要本地磁盘有文件,只是上传完成后可以执行一个业务定义的回调函数。
else
{
	struct stat stat_buf;

	if (stat(local_filename, &stat_buf) == 0 && \
		S_ISREG(stat_buf.st_mode))
	{
	file_size = stat_buf.st_size;
	result = storage_upload_by_callback(pTrackerServer, \
		&storageServer, store_path_index, \
		uploadFileCallback, local_filename, \
		file_size, file_ext_name, \
		meta_list, meta_count, \
		group_name, remote_filename);
	}

	printf("storage_upload_by_callback\n");
}

if (result != 0)
{
	//上传失败,本次请求处理失败,给用户出错信息,清理该storage的连接,可以返回循环体开始处重连tracker和storage。
	 fdfs_quit(&storageServer);
		      tracker_disconnect_server(&storageServer);
		      fdfs_quit(pTrackerServer);
		      tracker_disconnect_server(pTrackerServer);
		      break;
}

上传成功则保存文件索引信息group_name和 remote_filename,后面的timestamp和file size可以不要。
sprintf(file_id, "%s/%s", group_name, remote_filename);
url_len = sprintf(file_url, "http://%s:%d/%s", \
pTrackerServer->ip_addr, \
g_tracker_server_http_port, file_id);
memset(buff, 0, sizeof(buff));
base64_decode_auto(&context, remote_filename + FDFS_FILE_PATH_LEN, \
	strlen(remote_filename) - FDFS_FILE_PATH_LEN \
	- (FDFS_FILE_EXT_NAME_MAX_LEN + 1), buff, &len);
printf("group_name=%s, remote_filename=%s\n", \
	group_name, remote_filename);
printf("file timestamp=%d\n", buff2int(buff+sizeof(int)));
printf("file size="INT64_PRINTF_FORMAT"\n", \
	buff2long(buff+sizeof(int)*2));
printf("file url: %s\n", file_url);
//成功则不要关闭tracker和storage连接,下次处理请求时继续使用

1.7关闭连接,清理客户端
客户端退出时调用一次下面函数即可,注意在上面的循环处理出错时不要调用fdfs_client_destroy();清理整个客户端,注意下面函数的说明和作用:

//循环处理请求时没关闭则关闭storage连接
fdfs_quit(&storageServer);
tracker_disconnect_server(&storageServer);

//循环处理请求时没关闭则要关闭tracker连接
fdfs_quit(pTrackerServer);
tracker_close_all_connections();

//清理客户端
fdfs_client_destroy();
=============================================
void * MultiThreadTest( void *pParam )
{
    int                                 MessageSum=0;
    memcpy(&MessageSum, pParam,4);
    
    int nStartTime=time(NULL);
    int nReqStartTime=0,nReqEndTime=0,nTimeout=0;

    TrackerServerInfo theTrackerServer;
    memset(&theTrackerServer,0,sizeof(theTrackerServer));
    TrackerServerInfo *pTrackerServer = &theTrackerServer;
    
    int result=0;
    TrackerServerInfo storageServer;
    char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
    char remote_filename[256];
    char buff[32];
    char token[32 + 1];
    char file_id[128];
    char file_url[256];
    int len;
    int url_len;
    time_t ts;
    char *file_buff;
    int64_t file_size;
    char *meta_buff;
    int store_path_index;
    struct base64_context context;
    nReqStartTime=time(NULL);

    base64_init_ex(&context, 0, '-', '_', '.');//为了最后解析出文件大小需要初始化这个变量
   
    int i=0;//发送数目
    
    //处理在线请求时可以改成while(Running)
    while(i<MessageSum)
    {
        //pTrackerServer = tracker_get_connection();
        tracker_get_connection_r(pTrackerServer);   //多线程安全函数
        if (pTrackerServer->sock <= 0)
        {
            sleep(1);
            continue;
        }

        store_path_index = 0;
        if ((result=tracker_query_storage_store(pTrackerServer, \
                        &storageServer, &store_path_index)) != 0)
        {
            //查询tracker获取storage失败,重连tracker
            printf("tracker_query_storage fail, " \
                    "error no: %d, error info: %s\n", \
                    result, strerror(result));
            fdfs_quit(pTrackerServer);
            tracker_disconnect_server(pTrackerServer);
            sleep(1);
            continue;
        }

        printf("group_name=%s, ip_addr=%s, port=%d\n", \
                storageServer.group_name, \
                storageServer.ip_addr, \
                storageServer.port);

        if ((result=tracker_connect_server(&storageServer)) != 0)
        {
            //tracker给了一个连不上的storage,重连tracker
            fdfs_quit(pTrackerServer);
            tracker_disconnect_server(pTrackerServer);
            sleep(1);
            continue;
        }
        
        //1下面保持tracker和storage的各1个长连接,本线程固定跟一个storage交互,出错后才返回重连进行容灾
        //2实际中如果是大文件没必要保持长连接,则下面这个for循环去掉成顺序执行,出错时continue回while循环去重连tracker和storage
        for(;i<MessageSum;i++)
        {
            //以上传为例,上传目前有3类,根据应用自己选一类即可,参见版本自带的fdfs_test.c
            //if (strcmp(operation, "upload") == 0)
            {
                strcpy(group_name, "");
                result = storage_upload_by_filename(pTrackerServer, \
                                &storageServer, store_path_index, \
                                local_filename, NULL, \
                                NULL, 0, \
                                group_name, remote_filename);     

                //下面的出错或成功处理,跟上述上传种类无关,通用的
                if (result != 0)
                {
                    printf("storage_upload_by_filename fail, " \
                                "error no: %d, error info: %s\n", \
                                result, strerror(result));
                    fdfs_quit(&storageServer);
                    tracker_disconnect_server(&storageServer);
                    fdfs_quit(pTrackerServer);
                    tracker_disconnect_server(pTrackerServer);

                    //上传次数的统计,测试用
                    pthread_mutex_lock(&theSumMutex);
                    nTradeFailed ++;
                    nSum++;
                    pthread_mutex_unlock(&theSumMutex);        
                    break;
                }

                sprintf(file_id, "%s/%s", group_name, remote_filename);
                url_len = sprintf(file_url, "http://%s:%d/%s", \
                                pTrackerServer->ip_addr, \
                                g_tracker_server_http_port, file_id);
               

                memset(buff, 0, sizeof(buff));
                base64_decode_auto(&context, remote_filename + FDFS_FILE_PATH_LEN, \
                        strlen(remote_filename) - FDFS_FILE_PATH_LEN \
                         - (FDFS_FILE_EXT_NAME_MAX_LEN + 1), buff, &len);
                printf("group_name=%s, remote_filename=%s\n", \
                        group_name, remote_filename);
                printf("file timestamp=%d\n", buff2int(buff+sizeof(int)));
                printf("file size="INT64_PRINTF_FORMAT"\n", \
                        buff2long(buff+sizeof(int)*2));
                printf("file url: %s\n", file_url);
            
                //上传次数的统计,测试用
                pthread_mutex_lock(&theSumMutex);
                nTradeOK ++;
                nSum++;
                pthread_mutex_unlock(&theSumMutex);
            }                  
        }

        fdfs_quit(&storageServer);
        tracker_disconnect_server(&storageServer);
        fdfs_quit(pTrackerServer);
        tracker_disconnect_server(pTrackerServer);
    }
    //上传次数的统计,测试用
    int nEndTime=time(NULL);
    pthread_mutex_lock(&theSumMutex);
    OverThread ++;
    nTimeAllUsed += (nEndTime - nStartTime);
    pthread_mutex_unlock(&theSumMutex);        

    return NULL;
}

下面是PHP封装的API下载,如果你fastdfs安装配置好了,直接放在tracker服务器里用

点击下载