QT应用编程: 基于UDP协议设计的大文件传输软件

Posted 2021-06-12 DS小龙哥

一、环境介绍

QT版本: 5.12.6

编译器:   MinGW 32

传输协议: UDP

功能介绍:  软件由客户端和服务器组成，客户端通过 UDP协议不断循环地向服务端发送文件,文件传输速率可以达到10MB/s以上，文件传输后支持自动删除，客户端上可以支持每分钟创建一个文件并以时间戳命名,每个生成的文件可以设置大小，默认大小为6GB; 服务端收到文件之后，将文件进行存储到本地，可以指定时间自动删除文件; 服务端可以动态计算传输速率，并写入日志文件; 服务器可以支持同时接收多个客户端的文件上传。 

完整的项目源码下载地址(包含客户端与服务器)-打开即可编译运行: https://download.csdn.net/download/xiaolong1126626497/19006507

二、软件运行效果

三、传输协议介绍

本软件使用的网络传输协议为UDP协议，UDP本身是一个无连接协议，传输数据之前源端和终端不建立连接，当它想传送时就简单地去抓取来自应用程序的数据，并尽可能快地把它扔到网络上。在发送端，UDP传送数据的速度仅仅是受应用程序生成数据的速度、计算机的能力和传输带宽的限制；在接收端，UDP把每个消息段放在队列中，应用程序每次从队列中读一个消息段，由于传输数据不建立连接，因此也就不需要维护连接状态，包括收发状态等，因此一台服务机可同时向多个客户机传输相同的消息。

UDP信息包的标题很短，只有8个字节，相对于TCP的20个字节信息包而言UDP的额外开销很小。

UDP提供不可靠服务，具有TCP所没有的优势：

UDP无连接，时间上不存在建立连接需要的时延。空间上，TCP需要在端系统中维护连接状态，需要一定的开销。此连接装入包括接收和发送缓存，拥塞控制参数和序号与确认号的参数。UCP不维护连接状态，也不跟踪这些参数，开销小。空间和时间上都具有优势。

本软件的传输层框架采用的是UDT协议，UDT是基于UDP的数据传输协议，UDT是开源软件，主要目的是针对“TCP在高带宽长距离网络上的传输性能差”的问题，尽可能全面支持BDP网络上的海量数据传输。UDT是建立与UDP之上的面向双向的应用层协议，引入了新的拥塞控制算法和数据可靠性控制机制。它不仅可以支持可靠的数据流传输（STREAM 类型TCP）和部分可靠的数据报（DGRAM类似网络上发广播消息）传输，也可以应用在点对点技术，防火墙穿透，多媒体数据传输等领域。

UDT的特性

UDT的特性主要包括在以下几个方面：

1）基于UDP的应用层协议

2）面向连接的可靠协议

3）双工的协议

4）拥有新的拥塞控制算法，并具有可拓展的拥塞控制框架。

此外UDT协议在高BDP网络相对于TCP协议的优势，可以用下面几点来表示：

1）UDT是基于UDP协议，并且是定时器做的发送，不像tcp需要等待ack后才能开始下一轮发送

2）UDT的拥塞控制算法，能够实现在慢启动阶段快速增长抢占带宽，而在接近饱和时逐渐降低增长速度，使它趋于稳定。

3）UDT对包丢失的处理算法，和对噪声链路的容忍性，使得在网络波动比较大的环境中，它比传统的TCP协议更加的稳定

引入UDT的原因

互联网上的标准数据传输协议TCP在高带宽长距离网络上性能很差，且无法充分的利用带宽。其原因主要有一下几点：

1）现行的tcp拥塞窗口机制在高带宽长距离的环境下无法很好的工作，拥塞窗口太小，而且增加过于缓慢直接导致吞吐率不高，无法充分利用带宽。

此外TCP的AIMD拥塞控制算法过激地降低拥塞窗口的大小，但是不能快速回复到高位充分利用带宽。

2）目前的tcp拥塞控制算法在BDP网络下具有较差的RTT公平性，rtt会影响拥塞窗口的增长，越不容易达的链接的拥塞 

窗口增加得越慢，其发送速度越慢，因此会导致越远的链接发送速率越慢。

UDT网站链接:  https://udt.sourceforge.io/

UDT 项目源码官方下载地址:  https://sourceforge.net/projects/udt/

UDT协议移植到QT工程: https://blog.csdn.net/xiaolong1126626497/article/details/116118320

 四、软件逻辑部分源码

4.1  UDP客户端文件发送线程

#include "udp_file_send.h"

#ifndef WIN32
void* sendfile(void*);
#else
DWORD WINAPI sendfile(LPVOID);
DWORD WINAPI monitor(LPVOID s);
#endif

UDP_FILE_SEND_THREAD::UDP_FILE_SEND_THREAD()
{
    qDebug()<<"startup---->";
    UDT::startup();
}

UDP_FILE_SEND_THREAD::~UDP_FILE_SEND_THREAD()
{

}


void UDP_FILE_SEND_THREAD::run()
{
    FileInfoSendSuccess=0;
    Send_TotalBytes=0;
    File_TotalBytes=0;
    qint64 send_len=0;

    qint64 time1;
    qint64 time2;

    LogSend(QString("UDP文件发送线程开始运行.\\n"));

    //判断文件是否存在
    QFileInfo file(send_file);
    if(file.exists()==false)
    {
        LogSend(QString("%1 文件不存在.停止发送.\\n").arg(send_file));
      //  mSocket->close();
       // delete mSocket;
        return;
    }

    //打开文件
    QFile SrcFile(send_file);
    if(!SrcFile.open(QIODevice::ReadOnly))
    {
        LogSend(QString("%1 文件打开失败.停止发送.\\n").arg(send_file));
        return;
    }

    //得到文件大小信息
    File_TotalBytes=SrcFile.size();

    //判断端口号
    if(server_port<=0)
    {
        LogSend("没有填写端口号.暂停发送.\\n");
        return;
    }

    //判断IP地址
    if(server_ip_addr.isEmpty())
    {
        LogSend("没有填写IP地址.暂停发送.\\n");
        return;
    }

    struct addrinfo hints, *local, *peer;

    memset(&hints, 0, sizeof(struct addrinfo));

    hints.ai_flags = AI_PASSIVE;
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;

    if (0 != getaddrinfo(nullptr,"9000", &hints, &local))
    {
        LogSend("非法端口号或端口正忙.\\n");
        return;
    }

   client = UDT::socket(local->ai_family, local->ai_socktype, local->ai_protocol);
#ifdef WIN32
   UDT::setsockopt(client, 0, UDT_MSS, new int(1052), sizeof(int));
#endif
   freeaddrinfo(local);

     if (0 != getaddrinfo(QString("%1").arg(server_ip_addr).toStdString().c_str(),QString("%1").arg(server_port).toStdString().c_str(), &hints, &peer))
     {
        qDebug() << "incorrect server/peer address. " <<server_ip_addr << ":" << server_port << endl;
        return;
     }

     // 连接到服务器，隐式绑定
     if (UDT::ERROR == UDT::connect(client, peer->ai_addr, peer->ai_addrlen))
     {
        LogSend("连接到服务器，隐式绑定. ERROR\\n");
        //qDebug()<< "connect: " << UDT::getlasterror().getErrorMessage() << endl;
        return;
     }

     freeaddrinfo(peer);
     int ssize = 0;
     int ss;

     QString str;
     QFileInfo send_file_info(send_file);
     str=QString("image#%1#%2").arg(send_file_info.fileName()).arg(send_file_info.size());
     LogSend(str+"\\n");

     qDebug()<<"str.toStdString().c_str():"<<str.toStdString().c_str()<<",len:"<<str.size();

     if (UDT::ERROR == (ss = UDT::send(client, str.toStdString().c_str(),strlen(str.toStdString().c_str()), 0)))
     {
          // qDebug() << "send_error:" << UDT::getlasterror().getErrorMessage() << endl;
         LogSend("send_error\\n");
         return;
     }

     //设置运行状态
     run_flag=true;

     LogSend(QString("准备发送%1文件.目的地:%2:%3\\n").arg(send_file).arg(server_ip_addr).arg(server_port));

     //获取系统当前时间
     time1= QDateTime::currentMSecsSinceEpoch();

    //开始发送
    while(run_flag)
    {
        QByteArray byte;

        byte=SrcFile.read(MAX_READ_LEN);

        if (UDT::ERROR == (send_len = UDT::send(client,byte.data(),byte.size(), 0)))
        {
            LogSend("send_error....\\n");
            break;
        }
        //记录发送的长度
         Send_TotalBytes+=send_len;

         //获取本地时间
         current_ms_time=QDateTime::currentMSecsSinceEpoch();

         //时间经过了1s
         if(current_ms_time-old_ms_time>1000)
         {
             old_ms_time=current_ms_time;

             //更新状态
             ss_FileSendState(Send_TotalBytes,File_TotalBytes);
         }

        if(Send_TotalBytes>=File_TotalBytes || (byte.size()<MAX_READ_LEN))
        {
            break;
        }

        QThread::msleep(1);
    }

    UDT::close(client);

    //获取系统当前时间
    time2= QDateTime::currentMSecsSinceEpoch();

    //消耗的时间
    time1=time2-time1;

    //更新状态
    ss_FileSendState(Send_TotalBytes,File_TotalBytes);

    LogSend(QString("UDP文件发送完成. 总大小:%1 Byte. 耗时:%2\\n").arg(File_TotalBytes).arg(QTime(0, 0, 0,0).addMSecs(int(time1)).toString(QString::fromLatin1("HH:mm:ss:zzz"))));

    Send_TotalBytes=0;
    File_TotalBytes=0;

    //关闭文件
    SrcFile.close();

    //发送完毕
    emit ss_SendComplete();

    LogSend("发送线程已正常退出...\\n");
}


void UDP_FILE_SEND_THREAD::exit_thread()
{
    run_flag=false;
    //退出事件循环
    this->exit();
}

4.2 UDP服务器文件接收线程

#include "udp_file_recv.h"
QString recv_save_path=""; //保存接收文件存储的路径
class StringDataQueue log_queue;

UDP_FILE_RECV_THREAD::UDP_FILE_RECV_THREAD()
{
    qDebug()<<"startup---->";
    UDT::startup();
}

UDP_FILE_RECV_THREAD::~UDP_FILE_RECV_THREAD()
{

}


/*
工程: UDP_Server
日期: 2021-04-23
作者: DS小龙哥
环境: win10 QT5.12.6 MinGW32
功能: 开始接收UDP的数据
*/
void UDP_FILE_RECV_THREAD::run(QString ip,int port,QString save_path)
{
     recv_save_path=save_path;

     addrinfo hints;
     addrinfo* res;

     memset(&hints, 0, sizeof(struct addrinfo));

     hints.ai_flags = AI_PASSIVE;
     hints.ai_family = AF_INET;
     hints.ai_socktype = SOCK_STREAM;
     //hints.ai_socktype = SOCK_DGRAM;

     if (0 != getaddrinfo(nullptr,QString("%1").arg(port).toStdString().c_str(), &hints, &res))
     {
        qDebug() << "illegal port number or port is busy.\\n" << endl;
        return;
     }

     serv = UDT::socket(res->ai_family, res->ai_socktype, res->ai_protocol);

     if (UDT::ERROR == UDT::bind(serv, res->ai_addr, res->ai_addrlen))
        {
           qDebug() << "bind: " << UDT::getlasterror().getErrorMessage() << endl;
           return;
        }

        freeaddrinfo(res);

        if (UDT::ERROR == UDT::listen(serv, 10))
        {
           qDebug() << "listen: " << UDT::getlasterror().getErrorMessage() << endl;
           return;
        }

        sockaddr_storage clientaddr;
        int addrlen = sizeof(clientaddr);

        UDTSOCKET recver;

        LogSend("开始等待客户端连接....\\n");
        while (true)
        {
           if (UDT::INVALID_SOCK == (recver = UDT::accept(serv, (sockaddr*)&clientaddr, &addrlen)))
           {
              LogSend("服务器已停止监听....\\n");
              return;
           }

           char clienthost[NI_MAXHOST];
           char clientservice[NI_MAXSERV];
           getnameinfo((sockaddr *)&clientaddr, addrlen, clienthost, sizeof(clienthost), clientservice, sizeof(clientservice), NI_NUMERICHOST|NI_NUMERICSERV);
           qDebug() << "new connection: " << clienthost << ":" << clientservice << endl;

           #ifndef WIN32
              pthread_t rcvthread;
              pthread_create(&rcvthread, NULL, recvdata, new UDTSOCKET(recver));
              pthread_detach(rcvthread);
           #else
              LogSend("新的文件已到达,创建新的线程开始接收....\\n");
              CreateThread(nullptr, 0, recvdata, new UDTSOCKET(recver), 0, nullptr);
           #endif
        }
        UDT::close(serv);
}

/*
工程: UDP_Server
日期: 2021-04-23
作者: DS小龙哥
环境: win10 QT5.12.6 MinGW32
功能: 退出线程
*/
void UDP_FILE_RECV_THREAD::close_file()
{
    UDT::close(serv);
}



/*
工程: UDP_Server
日期: 2021-05-19
作者: DS小龙哥
环境: win10 QT5.12.6 MinGW32
功能:
数据传输格式:  "image#<文件名称>#<大小>"
数据格式示例:  "image#D:/123.mp4#80000"
*/
#ifndef WIN32
void* recvdata(void* usocket)
#else
DWORD WINAPI recvdata(LPVOID usocket)
#endif
{
   UDTSOCKET recver = *(UDTSOCKET*)usocket;
   delete (UDTSOCKET*)usocket;

   unsigned char* data;
   int size = 1024*1024;
   int rs=0;
   data = new unsigned char[size];
   qint64 file_len=0;

   int rcv_size;
   int var_size = sizeof(int);
   UDT::getsockopt(recver, 0, UDT_RCVDATA, &rcv_size, &var_size);
   qDebug()<<"rcv_size:"<<rcv_size;

   if (UDT::ERROR == (rs = UDT::recv(recver,(char*)data,1024,0)))
   {
      qDebug() << "文件头接收失败:" << UDT::getlasterror().getErrorMessage() << endl;
      return 0;
   }

    qDebug()<<"rs:"<<rs;

   qint64 rx_byte=0;
   qint64 s_file_size=0;
   QString s_file_name;
   std::string std_str_data=(char*)data;
   QString Rx_str_data=QString("%1").fromStdString(std_str_data);
   qDebug()<<"接收的原始第一包数据:"<<Rx_str_data;

   int addr=Rx_str_data.indexOf(QByteArray("image#"));

   QFile *recv_file_p=nullptr;
   if(addr!=-1) //查找成功
   {
       QString str_size=Rx_str_data.section('#',2,2); //取出图片字节大小字符串
       s_file_name=QString("%1/%2").arg(recv_save_path).arg(Rx_str_data.section('#',1,1));
       s_file_size=str_size.toLongLong();   //得到图片字节大小

       //创建文件
       recv_file_p=new QFile(s_file_name);
       if(recv_file_p->open(QIODevice::ReadWrite)==true)
       {
           log_queue.write_queue(QString("%1 文件创建成功.\\n").arg(s_file_name));
       }
       else
       {
           recv_file_p=nullptr;
           log_queue.write_queue(QString("%1 文件创建失败.\\n").arg(s_file_name));
       }

        log_queue.write_queue(QString("收到新文件:%1:%2\\n").arg(s_file_name).arg(s_file_size));
    }

   qint64 current_ms_time=0,old_ms_time=0;
   double speed_M_SEC=0.0;

   while (true)
   {
      int rcv_size;
      int var_size = sizeof(int);
      UDT::getsockopt(recver, 0, UDT_RCVDATA, &rcv_size, &var_size);
      if (UDT::ERROR == (rs = UDT::recv(recver,(char*)data,size,0)))
      {
         qDebug() << "recv:" << UDT::getlasterror().getErrorMessage() << endl;
         log_queue.write_queue("客户端断开连接...");
         break;
      }

      file_len+=rs;

      //qDebug()<<"接收数据:"<<data;

      //获取本地时间
      current_ms_time=QDateTime::currentMSecsSinceEpoch();

     // qDebug()<<"ID:"<<recver<<",接收长度:"<<file_len<<",Time:"<<current_ms_time;

      if(current_ms_time-old_ms_time>1000)
      {
          old_ms_time=current_ms_time;

          //计算速度
          speed_M_SEC=(file_len-rx_byte)/1024/1024.0;
          //如果小于0，表示文件1秒内就传输完毕了
          if(speed_M_SEC<1.0)speed_M_SEC=s_file_size/1024.0/1024.0;
          log_queue.write_queue(QString("%1 M/s").arg(speed_M_SEC));
          rx_byte=file_len;
      }

      //存储数据到文件
      if(recv_file_p)
      {
          //存储数据到文件
          recv_file_p->write((char*)data,rs);

          //接收完毕
          if(file_len>=s_file_size)
          {
              //qDebug()<<"ID:"<<recver<<",接收长度:"<<file_len<<",Time:"<<current_ms_time;
              //qDebug()<<"接收完毕...";
              log_queue.write_queue(QString("接收完毕:%1:%2:%3\\n").arg(s_file_name).arg(s_file_size).arg(file_len));
              break;
          }
      }
      else
      {
          break;
      }
   }

   recv_file_p->close();
   delete recv_file_p;
   recv_file_p=nullptr;

   log_queue.write_queue(QString("线程退出:%1:%2:%3\\n").arg(s_file_name).arg(s_file_size).arg(file_len));

   delete [] data;

   UDT::close(recver);

   #ifndef WIN32
      return NULL;
   #else
      return 0;
   #endif
}

void UDP_FILE_RECV_THREAD::WriteLog(QString text)
{
    log_queue.write_queue(text);
}