AI 助理
备案控制台
开发者社区 云计算 文章 正文

fastdfs源码阅读：上传和下载（文件客户端逻辑）

2024-01-15 470
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： fastdfs源码阅读：上传和下载（文件客户端逻辑）

一、fastdfs上传

对于fdfs中的不同协议，它们的协议header都是一样的，只是不同功能它们之间协议body格式不同

1、请求和回应的协议格式

协议header部分：

  • pkg_len：body的长度，不包含header长度
  • cmd：命令（比如指明是上传还是下载…）
  • status：表示状态(比如是请求状态还是响应状态)

请求的协议body部分：

  • storage_index:表示 storage的索引号（上传的时候要传到哪个storage）
  • file_size：文件大小
  • file_ext_name：文件后缀名
  • file_content：文件内容

回应的协议body部分：

  • group_name：组名（下图中的group1）
  • remote_filename：文件名称（表示下图中M00开始的一段）

2、源码结构图

fdfs中上传部分的 代码结构

3、源码阅读

1）fastdfs/client/fdfs_upload_file.c

也就是上传命令fdfs_upload_file调用的地方

这部分内容主要有几个部分

1.解析参数

2.如果命令参数设置了storage、指定上传的盘符，就进行设置。如果没有设置，就通过连接tracker，进行自动分配storage和存储位置

3.上传文件(在storage_do_upload_file中实现)

static void usage(char *argv[])
{
  printf("Usage: %s <config_file> <local_filename> " \   
    "[storage_ip:port] [store_path_index]\n", argv[0]);//输出upload_file的用法， 带中括号的是可选的命令参数（可以不指定，系统会默认指定一个storage的ip和端口，如果指定，可以强制选一个storage），store_path_index是上传的盘符（可以指定或者默认）
}
//上传文件命令的用法 : fdfs_upload_file /etc/fdfs/client.conf ./filename.txt        比如会返回：group1/M00/00/00/xxxxxx.txt
int main(int argc, char *argv[])
{
  char *conf_filename;
  char *local_filename;
  char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
  ConnectionInfo *pTrackerServer;
  int result;
  int store_path_index;
  ConnectionInfo storageServer;
  char file_id[128];
  if (argc < 3)
  {
    usage(argv);
    return 1;
  }
  log_init();
  g_log_context.log_level = LOG_ERR;
  ignore_signal_pipe();
  conf_filename = argv[1];
  if ((result=fdfs_client_init(conf_filename)) != 0)
  {
    return result;
  }
  pTrackerServer = tracker_get_connection();//连接traceker
  if (pTrackerServer == NULL)
  {
    fdfs_client_destroy();
    return errno != 0 ? errno : ECONNREFUSED;
  }
  local_filename = argv[2];
  *group_name = '\0';
  if (argc >= 4) //参数大于等于4，说明指定了storage
  {
    const char *pPort;
    const char *pIpAndPort;
    pIpAndPort = argv[3];
    pPort = strchr(pIpAndPort, ':');
    if (pPort == NULL)
    {
      fdfs_client_destroy();
      fprintf(stderr, "invalid storage ip address and " \
        "port: %s\n", pIpAndPort);
      usage(argv);
      return 1;
    }
    storageServer.sock = -1;
    snprintf(storageServer.ip_addr, sizeof(storageServer.ip_addr), \
       "%.*s", (int)(pPort - pIpAndPort), pIpAndPort);
    storageServer.port = atoi(pPort + 1);
    if (argc >= 5)
    {
      store_path_index = atoi(argv[4]);
    }
    else
    {
      store_path_index = -1;
    }
  }
  else if ((result=tracker_query_storage_store(pTrackerServer, \    
                  &storageServer, group_name, &store_path_index)) != 0)     // 请求 // tracker_query_storage_store宏定义，本质tracker_query_storage_store_without_group ，请求tracker去获取storage
  {
    fdfs_client_destroy();
    fprintf(stderr, "tracker_query_storage fail, " \
      "error no: %d, error info: %s\n", \
      result, STRERROR(result));
    return result;
  }
  // 上传文件（最终调用的是storage_do_upload_file）
  result = storage_upload_by_filename1(pTrackerServer, \        
      &storageServer, store_path_index, \
      local_filename, NULL, \
      NULL, 0, group_name, file_id);
  if (result == 0)
  {
    printf("%s\n", file_id);
  }
  else
  {
    fprintf(stderr, "upload file fail, " \
      "error no: %d, error info: %s\n", \
      result, STRERROR(result));
  }
  tracker_close_connection_ex(pTrackerServer, true);
  fdfs_client_destroy();
  return result;
}

2） storage_do_upload_file

主要分为3个步骤

1.将上半部分协议的信息全部打包到buffer中去

2.先发送上传请求的协议的上半部分，将内存中的数据(buffer)进行发送调用tcpsenddata_nb

3.然后发送上传请求的协议的下半部分，通过文件读取并发送调用tcpsendfile

int storage_do_upload_file(ConnectionInfo *pTrackerServer, \
  ConnectionInfo *pStorageServer, const int store_path_index, \
  const char cmd, const int upload_type, const char *file_buff, \
  void *arg, const int64_t file_size, const char *master_filename, \
  const char *prefix_name, const char *file_ext_name, \
  const FDFSMetaData *meta_list, const int meta_count, \
  char *group_name, char *remote_filename)
{
  ...//以上都是对协议进行包装，如storage_index,file_size等
  long2buff(file_size, p);      // 把文件长度写入到buffer中       8字节
  p += FDFS_PROTO_PKG_LEN_SIZE;
  if (bUploadSlave)
  {
    ...
  }
  else
  {
    memset(p, 0, FDFS_FILE_EXT_NAME_MAX_LEN); // 扩展名  固定长度6字节，先置0
  }
  if (file_ext_name != NULL)
  {
    ...//拷贝扩展名
  }
  p += FDFS_FILE_EXT_NAME_MAX_LEN;
  ...
  long2buff((p - out_buff) + file_size - sizeof(TrackerHeader), \
    pHeader->pkg_len);            // body的长度 
  pHeader->cmd = cmd;
  pHeader->status = 0;
  // 请求上传文件（也就是fdfs上传协议的上半部分）
  if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
    p - out_buff, g_fdfs_network_timeout)) != 0)    
  {
    logError("file: "__FILE__", line: %d, " \
      "send data to storage server %s:%d fail, " \
      "errno: %d, error info: %s", __LINE__, \
      pStorageServer->ip_addr, pStorageServer->port, \
      result, STRERROR(result));
    break;
  }
  //请求上传文件（fdfs上传协议的下半部分）（数据部分）
  if (upload_type == FDFS_UPLOAD_BY_FILE)//一般都是按文件进行上传          
  {
    if ((result=tcpsendfile(pStorageServer->sock, file_buff, \  
      file_size, g_fdfs_network_timeout, \
      &total_send_bytes)) != 0)  // 这里才是真正发送文件 （使用了sendfile进行零拷贝）
    {
      break;
    }
  }
  else if (upload_type == FDFS_UPLOAD_BY_BUFF)   
  {
    if ((result=tcpsenddata_nb(pStorageServer->sock, \
      (char *)file_buff, file_size, \
      g_fdfs_network_timeout)) != 0)
    {
      logError("file: "__FILE__", line: %d, " \
        "send data to storage server %s:%d fail, " \
        "errno: %d, error info: %s", __LINE__, \
        pStorageServer->ip_addr, pStorageServer->port, \
        result, STRERROR(result));
      break;
    }
  }
  else //FDFS_UPLOAD_BY_CALLBACK
  {
    UploadCallback callback;
    callback = (UploadCallback)file_buff;
    if ((result=callback(arg, file_size, pStorageServer->sock))!=0)
    {
      break;
    }
  }
  pInBuff = in_buff;
  if ((result=fdfs_recv_response(pStorageServer, \      
    &pInBuff, sizeof(in_buff), &in_bytes)) != 0)  // 获取响应
  {
    logError("file: "__FILE__", line: %d, "
                "fdfs_recv_response fail, result: %d",
                __LINE__, result);
    break;
  }
  //以下是对获取响应的一些处理
  ...
  if (new_connection)
  {
    tracker_close_connection_ex(pStorageServer, result != 0);
  }
  return result;
}

从上面可以看到有三种发送的方式，但是一般都是默认选择FDFS_UPLOAD_BY_FILE,按文件上传

二、fastdfs下载

1、请求和回应的协议格式

  • file_offset：下载的文件起始偏移位置（比如偏移为0，那么就是从头开始下载）
  • download_bytes：要下载的字节数量(比如文件10MB，先下载2MB)

2、源码结构图

3、源码阅读

1）fastdfs/client/fdfs_download_file.c

使用命令fdfs_download_file下载文件，就会调用下面这部分内容。

这一部分主要是解析参数，比如设置下载的起始偏移位置和下载的字节数

如果没有设置，默认是从头开始下载

最后调用storage_do_download_file1_ex进行下载（发送下载请求和接受数据）

int main(int argc, char *argv[])
{
  char *conf_filename;
  char *local_filename;
  ConnectionInfo *pTrackerServer;
  int result;
  char file_id[128];
  int64_t file_size;
  int64_t file_offset;
  int64_t download_bytes;
  if (argc < 3)
  {
    printf("Usage: %s <config_file> <file_id> " \
      "[local_filename] [<download_offset> " \
      "<download_bytes>]\n", argv[0]);
    return 1;
  }
  log_init();
  g_log_context.log_level = LOG_ERR;
  ignore_signal_pipe();
  conf_filename = argv[1];
  if ((result=fdfs_client_init(conf_filename)) != 0)
  {
    return result;
  }
  pTrackerServer = tracker_get_connection();
  if (pTrackerServer == NULL)
  {
    fdfs_client_destroy();
    return errno != 0 ? errno : ECONNREFUSED;
  }
  snprintf(file_id, sizeof(file_id), "%s", argv[2]);
  file_offset = 0;
  download_bytes = 0;
  if (argc >= 4)
  {
    local_filename = argv[3];
    if (argc >= 6)
    {
      file_offset = strtoll(argv[4], NULL, 10);//文件的字节偏移
      download_bytes = strtoll(argv[5], NULL, 10);//下载字节数
    }
  }
  else
  {
    local_filename = strrchr(file_id, '/');
    if (local_filename != NULL)
    {
      local_filename++;  //skip /
    }
    else
    {
      local_filename = file_id;
    }
  }
  //在storage_do_download_file_ex实现下载
  result = storage_do_download_file1_ex(pTrackerServer, \
                NULL, FDFS_DOWNLOAD_TO_FILE, file_id, \
                file_offset, download_bytes, \
                &local_filename, NULL, &file_size);
  if (result != 0)
  {
    printf("download file fail, " \
      "error no: %d, error info: %s\n", \
      result, STRERROR(result));
  }
  tracker_close_connection_ex(pTrackerServer, true);
  fdfs_client_destroy();
  return 0;
}

2）storage_do_download_file_ex

主要有三部分

1.连接stoarge

2.打包协议信息，发送下载请求

3.下载文件（可以下载到磁盘，内存，获得通过回调函数的方式，默认是下载到磁盘FDFS_DOWNLOAD_TO_FILE）

int storage_do_download_file_ex(ConnectionInfo *pTrackerServer, \
    ConnectionInfo *pStorageServer, \
    const int download_type, \
    const char *group_name, const char *remote_filename, \
    const int64_t file_offset, const int64_t download_bytes, \
    char **file_buff, void *arg, int64_t *file_size)
{
  ...
  *file_size = 0;
  if ((result=storage_get_read_connection(pTrackerServer, \
    &pStorageServer, group_name, remote_filename, \
    &storageServer, &new_connection)) != 0)     // 连接storage
  {
    return result;
  }
  do
  {
  ...//将一些请求协议的信息打包到buffer中
  pHeader->cmd = STORAGE_PROTO_CMD_DOWNLOAD_FILE;   // 请求storage下载
  if ((result=tcpsenddata_nb(pStorageServer->sock, out_buff, \
    out_bytes, g_fdfs_network_timeout)) != 0)   //发送下载请求
  {
    logError("file: "__FILE__", line: %d, " \
      "send data to storage server %s:%d fail, " \
      "errno: %d, error info: %s", __LINE__, \
      pStorageServer->ip_addr, pStorageServer->port, \
      result, STRERROR(result));
    break;
  }
  if (download_type == FDFS_DOWNLOAD_TO_FILE)//storage回应数据
  { // 请求接收header
    if ((result=fdfs_recv_header(pStorageServer, \    
      &in_bytes)) != 0)  //in_bytes 一般就是说我们body的长度（解析获取的header，可以获得body的长度）
    {
            logError("file: "__FILE__", line: %d, "
                    "fdfs_recv_header fail, result: %d",
                    __LINE__, result);
      break;
    }
    //in_bytes 是获取到的文件的大小
    if ((result=tcprecvfile(pStorageServer->sock, \   
        *file_buff, in_bytes, 0, \
        g_fdfs_network_timeout, \
        &total_recv_bytes)) != 0)  // 接收文件
    {
      break;
    }
  }
  else if (download_type == FDFS_DOWNLOAD_TO_BUFF)//如果要把数据读到内存里去
  {
    *file_buff = NULL;
    if ((result=fdfs_recv_response(pStorageServer, \
      file_buff, 0, &in_bytes)) != 0)
    {
            logError("file: "__FILE__", line: %d, "
                    "fdfs_recv_response fail, result: %d",
                    __LINE__, result);
      break;
    }
  }
  else//通过设置的回调函数去下载
  {
    DownloadCallback callback;
    char buff[2048];
    int recv_bytes;
    int64_t remain_bytes;
    if ((result=fdfs_recv_header(pStorageServer, \
      &in_bytes)) != 0)
    {
            logError("file: "__FILE__", line: %d, "
                    "fdfs_recv_header fail, result: %d",
                    __LINE__, result);
      break;
    }
    callback = (DownloadCallback)*file_buff;
    remain_bytes = in_bytes;
    while (remain_bytes > 0)
    {
      if (remain_bytes > sizeof(buff))
      {
        recv_bytes = sizeof(buff);
      }
      else
      {
        recv_bytes = remain_bytes;
      }
      if ((result=tcprecvdata_nb(pStorageServer->sock, buff, \
        recv_bytes, g_fdfs_network_timeout)) != 0)
      {
        logError("file: "__FILE__", line: %d, " \
          "recv data from storage server " \
          "%s:%d fail, " \
          "errno: %d, error info: %s", __LINE__, \
          pStorageServer->ip_addr, \
          pStorageServer->port, \
          result, STRERROR(result));
        break;
      }
      result = callback(arg, in_bytes, buff, recv_bytes);
      if (result != 0)
      {
        logError("file: "__FILE__", line: %d, " \
          "call callback function fail, " \
          "error code: %d", __LINE__, result);
        break;
      }
      remain_bytes -= recv_bytes;
    }
    if (remain_bytes != 0)
    {
      break;
    }
  }
  *file_size = in_bytes;
  } while (0);
  if (new_connection)
  {
    tracker_close_connection_ex(pStorageServer, result != 0);
  }
  return result;
}

4、补充

多线程下载

由于协议中有偏移和下载字节数，因此可以自己去实现使用多线程进程下载，分成很多段，用不同的线程去下载

这样做有什么好处？

1、比如服务器对每个下载请求都限速了，那么可以用多线程来提速

2、服务器如果此时负荷比较高，多个连接，可以多占用资源进行下载

断点续传

这是fdfs内部就实现的功能，将文件追加到上一次上传的文件的末尾

echo hello > test1.txt
echo world > test2.txt
fdfs_upload_appender /etc/fdfs/client.conf test1.txt
得到：group1/M00/00/00/rBMYd2Id2FmEay2-AAAAADY6MCA286.txt ，在fdfs_append_file的时候需要
fdfs_append_file /etc/fdfs/client.conf group1/M00/00/00/rBMYd2Id2FmEay2-AAAAADY6MCA286.txt test2.txt
在服务器相应的⽬录下查找对应的⽂件，⽤cat读取⽂件内容。
root@iZbp1h2l856zgoegc8rvnhZ:/home/fastdfs/storage/data/00/00# cat rBMYd2Id2FmEay2-AAAAADY6MCA286.txt
hello
world
  • fdfs_upload_appender 上传第⼀部分⽂件
  • fdfs_append_file 上传其他部分的⽂件

注意：

注意断点续传的顺序性

⽀持断点续传，但fastdfs并不⽀持多线程分⽚上传同⼀个⽂件

文章标签：
存储
菊头蝙蝠
目录
相关文章
三分钟热度的鱼
|
监控 物联网 开发工具
MQTT常见问题之MQTT云端sdk消费者 出现重复消费如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
905 2
fanstuck
|
Kubernetes 数据可视化 数据安全/隐私保护
云服务器CentOS8.2Docker图形化管理工具Portainer安装部署
云服务器CentOS8.2Docker图形化管理工具Portainer安装部署
fanstuck
689 0
云服务器CentOS8.2Docker图形化管理工具Portainer安装部署
yuanzhengme
|
Java 关系型数据库 MySQL
Elasticsearch【问题记录 01】启动服务&停止服务的2类方法【及 java.nio.file.AccessDeniedException: xx/pid 问题解决】（含shell脚本文件）
【4月更文挑战第12天】Elasticsearch【问题记录 01】启动服务&停止服务的2类方法【及 java.nio.file.AccessDeniedException: xx/pid 问题解决】（含shell脚本文件）
yuanzhengme
1094 3
上云活动指导
|
缓存 Linux 开发工具
CentOS 7- 配置阿里镜像源
阿里镜像官方地址http://mirrors.aliyun.com/ 1、点击官方提供的相应系统的帮助 ：2、查看不同版本的系统操作： 下载源1、安装wget yum install -y wget2、下载CentOS 7的repo文件wget -O /etc/yum.
上云活动指导
258040 0
GoodTime
|
8月前
|
人工智能 C# C++
在 Visual Studio 上体验腾讯云 AI 代码助手
本文介绍如何在 Visual Studio 2022 中安装和使用腾讯云 AI 代码助手。首先，通过扩展管理器搜索并安装插件，重启 VS 完成安装。接着，在工具栏的“工具”中找到并启动助手，快捷键为 ALT + T。测试时输入问题如冒泡排序，助手会生成 C# 代码并可直接运行。建议优化默认代码语言为 C#、增加代码插入功能，并统一显示位置以提升用户体验。
GoodTime
423 1
梦笔生花伟
|
算法 NoSQL 编译器
使用CLion进行ROS开发
1.安装基本的ROS环境 　　ROS环境的安装请参考安装ROS。 安装CLion 下载CLion 　　Linux的下载地址如下：CLion 解压CLion 　　将下载的CLion复制到/opt目录下（你可以解压到适合自己的文件夹，只要保证后续使用的路径一致即可），然后在这个目录下使用终端将其解压，命令如下所示(其中CLion-xxxx.xx.xx.tar.gx为下载的安装包名)： sudo tar -zxvf CLion-xxxx.xx.xx.tar.gx 使用脚本安装CLion 运行解压后的文件夹中的clion.sh脚本，进行安装： 1.运行安装 2.登录激活可以选择教
梦笔生花伟
397 6
热爱技术的小郑
|
存储 Java 测试技术
阿里巴巴java开发手册
这篇文章是关于阿里巴巴Java开发手册的整理，内容包括编程规约、异常日志、单元测试、安全规约、MySQL数据库使用以及工程结构等方面的详细规范和建议，旨在帮助开发者编写更加规范、高效和安全的代码。
热爱技术的小郑
18490 3
月木天上
|
Java
java操作fastdfs包括文件上传、下载、删除
java操作fastdfs包括文件上传、下载、删除
月木天上
632 0
java操作fastdfs包括文件上传、下载、删除
游客ubxfrsiqpy57c
|
存储 固态存储 Linux
如何看电脑的配置
**电脑配置关乎日常使用体验，包括CPU、内存、硬盘、显卡、主板和操作系统等。要查看配置，可右击“此电脑”选“属性”查看基础信息，使用任务管理器检查性能，运行&quot;msinfo32&quot;获取详细系统信息，或借助如CPU-Z等第三方工具。了解配置助于选购和优化电脑。**
游客ubxfrsiqpy57c
992 0
如何看电脑的配置
阿甘兄
|
Oracle Java 关系型数据库
深入理解JVM系列教程（01） - HotSpot VM
深入理解JVM系列教程（01） - HotSpot VM
阿甘兄
375 0