一种点对点文件断点续传的多线程实现方法

  1 概述

  文件断点续传就是在主机与主机间传输文件时,可以将文件分多次传输,与非断点续传不同的是,每次传输不必传输整个文件。断点续传是大型文件数据传输的核心。本文将以多线程技术和Socket技术为依托,介绍大型文件断点续传的实现方法。

  2 基本实现思想

  多线程断点续传实现的基本思想就是在发送端(也称客户端)将要传输的文件分割为大小相当的多块,用多个线程,将这些块同时向目标服务器端发送;在服务器端的服务程序监听数据传输请求,每当接到新的请求,则创建一个新的线程,与客户端的发送线程对应,接收数据,记录数据传输进程。

  图1是点对点文件断点续传第N块传输过程示意图。在传输发起端(客户端),将大型文件事先分割为大小相当的N块,同时创建N个传输线程,连接目标端服务器。当服务器端接收到每一个连接请求后,告知客户端可以传输文件。当客户端接收到可以传输文件的消息时,首先向服务器发送数据传输信息块(包括第几块、在块中的起始位置)请求,当服务器端接收到该请求后,向客户端发送数据传输信息,客户端然后传输数据传输信息块指定的数据给服务器端,服务器端更新数据传输信息块。

  图1 点对点文件断点续传过程示意图

  3 具体实现

  在实现过程中我使用了MFC的多线程和Windows的Socket,分客户端和服务器端实现。因为数据传输往往是对等的,所以需要将客户端和服务器端集成在一起,在客户端和服务器端分别实现后,这是件非常简单的工作,只需将它们集成在一起即可。下面将分别介绍客户端和服务器端的实现。

  3.1 关键数据结构

  文件信息数据结构 用于在服务器端和客户端之间传递文件第N块的属性信息,详细定义如下:

  structfileinfo

  {

  int fileno; //文件号

  int type; //消息类别

  long len; //文件(块)长度,在客户端向服务器端发送数据时,是文件长度;

  //在服务器端向客户端发送已传部分的信息时,是应该续传部分的长度;

  long seek; //开始位置,标识要传输数据在原文件中的起始位置

  char name[MAX_PATH_LEN];//文件名

  };

  发送进度记录结构 用户记录文件传输进程,详细定义如下:

  structSentInfo

  {

  long totle; //已经成功发送数据的长度;

  int block; //块标识;

  long filelen; //文件总长度;

  int threadno; //负责传输第N块数据的线程标识;

  CString name; //文件名称

  };

  客户端类 客户端文件发送实例封装,用户记录客户端文件发送过程中的属性信息、已发送大小、发送线程句柄、发送线程状态、发送统计信息等,具体定义是:

  classCClient:publicCObject

  {

  protected:

  

  //Attributes

  public:

  CClient(CString ip);

  ~CClient();

  SentInfo doinfo;

  long m_index; //块索引

  BOOL sendOk[BLOCK]; //块发送结束状态

  CString SendFileName;

  CString DestIp; //目标IP地址

  THREADSTATUS SendStatus;

  int GetBlockIndex(); //获取当前要传输文件块的序号,例如0,1,2…

  CCriticalSection m_gCS;

  

  //Sending File Block Thread Handles

  HANDLE m_thread[BLOCK]; //Block Sending Thread Handles Array

  HANDLE m_ExitThread; //WaitToExitThread Handle

  HANDLE m_ProcessTrackThread;

  HANDLE m_hEventKill; //User canceled Event

  long m_IdxInListCtrl; //Index in ListView

  long m_lBeginTimeTick;

  long m_lBytesSent;

  // Operations

  public:

  };

  3.2 客户端

  发送线程 用于发送由lpparam参数指定的Client对象标识的文件块数据。具体实现是:

  //发送线程主函数

  //并行进行的线程,针对每个要发送的文件有BLOCK个该线程

  DWORDWINAPISendThread(LPVOIDlpparam)

  {

  CClient* pClient=(CClient*)lpparam;

  int block_index = pClient->GetBlockIndex();

  //发送文件的第m_index块

  //发送线程

  sockaddr_in local;

  SOCKET m_socket;

  int rc=0;

  

  local.sin_family=AF_INET;

  local.sin_port=htons(pMainFrm->m_FilePort);

  local.sin_addr.S_un.S_addr=inet_addr(pClient->DestIp);

  m_socket=socket(AF_INET,SOCK_STREAM,0);

  int ret;

  char m_buf[MAX_BUFSIZE];

  fileinfo file_info;

  //Connect to Server

  ret = connect(m_socket,(LPSOCKADDR)&local,sizeof(local));

  while((ret == SOCKET_ERROR)&&(WaitForSingleObject(pClient->m_hEventKill, 0)

  ==WAIT_TIMEOUT))

  {

  closesocket (m_socket);

  }

  file_info.len = pClient->doinfo.filelen ;

  if (file_info.len doinfo.name);

  //发送第index块已发送大小请求

  sendn(m_socket,(char*)&file_info,PROTOCOL_HEADSIZE);

  long nRead = readn(m_socket,m_buf,PROTOCOL_HEADSIZE);

  if(nRead len == 0)

  {

  closesocket (m_socket);

  //block_index块是续传,且发送完毕

  pClient->good[block_index] = TRUE;

  return 1;

  }

  CFile SourceFile;

  SourceFile.Open(pClient->SendFileName,

  CFile::modeRead|CFile::typeBinary| CFile::shareDenyNone);

  //Seek right position of sending file

  long apos=SourceFile.Seek(pfile_info->seek,CFile::begin);

  int len1;

  len1=pfile_info->len;

  //取得已经发送的长度

  if (block_index doinfo.totle += pClient->doinfo.filelen / BLOCK - pfile_info->len;

  else

  pClient->doinfo.totle += pClient->doinfo.filelen –

  pClient->doinfo.filelen/BLOCK* (BLOCK - 1) - pfile_info->len;

  int nBegin = GetTickCount();

  int SendTotal = 0;

  while((len1 > 0) && (WaitForSingleObject(pClient->m_hEventKill, 0) == WAIT_TIMEOUT))

  {

  nRead = SourceFile.Read(m_buf, (len1 m_CurrentBlockSize) ?

  len1:pMainFrm->m_CurrentBlockSize);

  int send_count=sendn(m_socket,m_buf,nRead);

  float nLimit;

  if(send_countdoinfo.totle += send_count;

  SendTotal += send_count;

  }

  SourceFile.Close();

  shutdown(m_socket,2);

  closesocket (m_socket);

  if(len1 good[block_index] = TRUE;

  return 1;

  }

  创建发送线程 对应要发送的文件,创建BLOCK个线程,负责发送各自的数据块。

  for(inti=0;i

  {

  //创建线程,状态为停止状态

  pClient->m_thread[i] = ::CreateThread(NULL,0,SendThread,(LPVOID)pClient,

  CREATE_SUSPENDED,&dwthread);

  

  ::SetThreadPriority(pClient->m_thread[i],THREAD_PRIORITY_HIGHEST);

  }

  3.3 服务器端

  监听函数 负责监听文件传输请求,在接受请求后,创建数据块接收线程,具体实现如下:

  //监听函数

  DWORDWINAPIlistenthread(LPVOIDlpparam)

  {

  SOCKET pSocket=(SOCKET)lpparam;

  

  //最多监听MAX_LISTENCOUNT个客户端连接

  int rc=listen(pSocket,MAX_LISTENCOUNT);

  if (rcm_DestPath + "\\IP" + PeerName + "-" + DestFileName,

  CFile::modeWrite|CFile::typeBinary| CFile::shareDenyNone))

  {

  DWORD error = GetLastError();

  return;

  };

  

  long apos = DestFile.Seek(fileseek,CFile::begin);

  char m_buf[MAX_BUFSIZE];

  CString m_temp;

  m_temp.Format("%s\\Temp\\IP%s-%s.%d",pWnd->m_appPath,PeerName,

  DestFileName,block_index);

  CFile BlockFile;

  BlockFile.Open(m_temp, CFile::modeWrite | CFile::typeBinary);

  while((filelen>0) && (WaitForSingleObject(pWnd->m_hEventKill, 0) == WAIT_TIMEOUT))

  {

  int nRead = readn(so, m_buf, (filelen > MAX_BUFSIZE) ? MAX_BUFSIZE : filelen);

  

  //Exited

  if (nRead == 0)

  {

  CString msg;

  msg.Format("[SOCKET %d] Receiving file %s break(Recv 0).",so,DestFileName);

  pWnd->SendMessage(WM_NEWMSG,(LPARAM)msg.GetBuffer(0),3);

  msg.ReleaseBuffer();

  break;

  }

  if(nReadSendMessage(WM_NEWMSG,(LPARAM)msg.GetBuffer(0),3);

  msg.ReleaseBuffer();

  break;

  }

  DestFile.Write(m_buf,nRead);

  filelen -= nRead;

  fileseek += nRead;

  BlockFile.Seek(0,CFile::begin);

  BlockFile.Write(&fileseek,sizeof(long));

  BlockFile.Write(&filelen,sizeof(long));

  pfileinfo->m_seek = fileseek;

  pfileinfo->m_len = filelen;

  }

  closesocket (so);

  //Current Block Received OK

  BlockFile.Close();

  DestFile.Close();

  if (filelen canDel = true;

  }

  初始化文件数据块信息 负责初始化文件传输时数据块的大小、开始位置等信息。具体实现是:

  //初始化文件数据块信息

  CFileInfoRecving*GetRecvedFileBlock(CStringPeerName,CStringFileName,long& filelen,

  long&fileseek,intblock_index)

  {

  CString m_temp;

  CMainFrame* pWnd = (CMainFrame*) AfxGetMainWnd();

  m_temp.Format("%s\\temp\\IP%s-%s.%d",pWnd->m_appPath,PeerName,FileName,block_index);

  FILE* fp=NULL;

  CFile Blockfile;

  CFile RecvFile;

  long seek = fileseek;

  long flen = filelen;

  CString DestFile = pWnd->m_DestPath + "\\IP" + PeerName + "-" + FileName;

  

  bool DestFileNoExist = false;

  if((fp=fopen(DestFile,"r"))==NULL)

  {

  RecvFile.Open(DestFile,

  CFile::modeCreate|CFile::modeWrite|CFile::typeBinary| CFile::shareDenyNone);

  RecvFile.Close();

  DestFileNoExist = true;

  }

  if (fp) fclose(fp);

  CFileInfoRecving* pfileinfo = NULL;

  //如果目标文件不存在或者长度记录文件不存在,重新传

  if( ((fp=fopen(m_temp,"r"))==NULL) || (DestFileNoExist))

  {

  //第一次接收

  fileseek = block_index*(flen/BLOCK);

  filelen = flen/BLOCK;

  if (block_index == (BLOCK - 1))

  filelen = flen - fileseek;

  

  //Add To Recving File List

  pfileinfo = new CFileInfoRecving();

  pfileinfo->m_sFileName = FileName;

  pfileinfo->m_sPeerName = PeerName;

  pfileinfo->m_nBlockIndex = block_index;

  

  pfileinfo->m_sFileSize.Format("%d",flen);

  

  seek = fileseek;

  flen = filelen;

  pfileinfo->m_seek = seek;

  pfileinfo->m_len = flen;

  pWnd->m_RecvingFileList.AddTail(pfileinfo);

  Blockfile.Open(m_temp,

  CFile::modeCreate|CFile::modeWrite| CFile::typeBinary | CFile::shareDenyNone);

  Blockfile.Write(&seek,sizeof(long));

  Blockfile.Write(&flen,sizeof(long));

  Blockfile.Close();

  }

  else

  {

  POSITION pos = pWnd->m_RecvingFileList.GetHeadPosition();

  while (pos)

  {

  pfileinfo = pWnd->m_RecvingFileList.GetAt(pos);

  if (pfileinfo)

  {

  if ((pfileinfo->m_sFileName == FileName) &&

  (pfileinfo->m_sPeerName == PeerName) &&

  (pfileinfo->m_nBlockIndex == block_index))

  break;

  }

  pWnd->m_RecvingFileList.GetNext(pos);

  }

  

  Blockfile.Open(m_temp,CFile::modeRead | CFile::typeBinary | CFile::shareDenyNone);

  Blockfile.Read(&fileseek,sizeof(long));

  Blockfile.Read(&filelen,sizeof(long));

  if ((pfileinfo == NULL) || (!pos))

  {

  pfileinfo = new CFileInfoRecving();

  pfileinfo->m_sFileName = FileName;

  pfileinfo->m_sPeerName = PeerName;

  pfileinfo->m_nBlockIndex = block_index;

  pfileinfo->m_sFileSize.Format("%d",flen);

  //Add To Recving File List

  //pWnd->m_RecvingFileListCS.Lock();

  pWnd->m_RecvingFileList.AddTail(pfileinfo);

  //pWnd->m_RecvingFileListCS.Unlock();

  }

  Blockfile.Close();

  pfileinfo->m_seek = fileseek;

  pfileinfo->m_len = filelen;

  }

  if (fp) fclose(fp);

  return pfileinfo;

  }

  数据接收线程 负责从客户端接收数据信息:

  //数据接收线程

  DWORDWINAPIServerthread(LPVOIDlpparam)

  {

  fileinfo* pFileInfo;

  

  //

  char* m_buf = new char[PROTOCOL_HEADSIZE];

  SOCKET pthis=(SOCKET)lpparam;

  int nRead = readn(pthis,m_buf,PROTOCOL_HEADSIZE);

  if( nRead type)

  {

  case 1:

  //获取文件块的已接收大小

  pfileinfo =

  GetRecvedFileBlock(FileID,pFileInfo->name,pFileInfo->len,pFileInfo->seek,pFileInfo->fileno);

  

  nRead = sendn(pthis,(char*)pFileInfo,PROTOCOL_HEADSIZE);

  if(nRead name,pFileInfo->seek,

  pFileInfo->len,pFileInfo->fileno,pfileinfo);

  closesocket(pthis);

  break;

  default:

  break;

  }

  delete [] m_buf;

  return 0;

  }

  4 小结

  本文简要介绍了点对点文件断点续传的多线程实现方法,有一定的使用性。限于水平,文中不当之处,敬请批评指正。

  作者简介:冯常青(1974-),男,北京,硕士,工程师,主要研究方向:计算机网络

       师明珠(1976-),男,北京,硕士,工程师,主要研究方向:计算机应用技术;

----《通信世界》


微信扫描分享本文到朋友圈
扫码关注5G通信官方公众号,免费领取以下5G精品资料
  • 1、回复“YD5GAI”免费领取《中国移动:5G网络AI应用典型场景技术解决方案白皮书
  • 2、回复“5G6G”免费领取《5G_6G毫米波测试技术白皮书-2022_03-21
  • 3、回复“YD6G”免费领取《中国移动:6G至简无线接入网白皮书
  • 4、回复“LTBPS”免费领取《《中国联通5G终端白皮书》
  • 5、回复“ZGDX”免费领取《中国电信5GNTN技术白皮书
  • 6、回复“TXSB”免费领取《通信设备安装工程施工工艺图解
  • 7、回复“YDSL”免费领取《中国移动算力并网白皮书
  • 8、回复“5GX3”免费领取《R1623501-g605G的系统架构1
  • 本周热点本月热点

     

      最热通信招聘

      最新招聘信息

    最新技术文章

    最新论坛贴子