想了个不是问题的问题

昨天被一个问题困扰到，最后发现这个问题就不是个问题。

怎么说呢？就是写服务器和客户端程序，两端想互相发送文件，对于小文件来说的话，是没啥，感觉不到发送过程的缓慢，但是换成一个几百兆的文件，就有点累了，得等上个几分钟才能发完，我就想为什么不能实现秒传。所以就在网上学习了多线程发送技术。实现了近百兆文件秒发的目标。具体功能就是服务器传文件到客户端客户端再反传给服务器，运行截图如下：
客户端接收到的文件：

服务器端发送的文件（framework.7z）和接收到客户端反传回的文件（_ramwork.7z）：
200MB的文件不到两秒传完了。结果显示内容也没啥差别。

下面是基本思想,其实在上一篇文章都说了在这里用流程图来表示吧！

这方法快是快，但致命的缺点就是不能用到IO多路复用的服务器中。只是用一个客户端和一个服务器之间的情况。要是多个客户端的话，服务器就懵了，不知道那个连接发来的内容该写到同一文件了，另外发一个文件创建太多连接对性能影响也比较大，所以这种传文件的方法用在服务器和客户端之间发大文件不现实！

不过话说回来，现在我们在网上下载个大文件的话也得等上一段时间才行。限制就在于一个TCP连接数据缓冲区本来就有限制，最多也就一次发65535bytes的长度。但跨越这个限制建立多连接用起来却不灵活。所以还是老老实实用单连接发吧，服务器那边也容易处理。
下面是我的代码实现：

• writen和readn

#pragma once
#include<unistd.h>
#include<stdlib.h>
#include<fcntl.h>
#include<stdio.h>
#include<errno.h>
int readn(int fd, void *buf, int n) ;
//写数据
int writen(int fd, void *buf, int n) ;

int readn(int fd, void *buf, int n) {
    int nleft = n; //还需要读取的字节数
    char *bufptr =(char*)buf; //指向read函数当前存放数据的位置
    int  nread = 0 ;
    while(nleft > 0){
        if((nread = read(fd, bufptr, nleft)) < 0){
            if(errno == EINTR){ //遇到中断
                continue;
            }
            else            // 其他错误
                return -1;
        }
        else if(nread == 0){ // 遇到EOF
            break;
        }
        nleft -= nread;
        bufptr += nread;
    }
    return (n - nleft);
}

int writen(int fd, void *buf, int n){
    int nleft = n;
    char *bufptr = (char*)buf;
    int nwrite;

    while(nleft > 0){
        if((nwrite = write(fd, bufptr, nleft)) < 0){
            if(errno == EINTR)
                continue;
            else
                return -1;
        }
        if(nwrite == 0)
            continue;
        nleft -= nwrite;
        bufptr += nwrite;
    }
    return (n-nleft); //  注意这里必须是 n 因为这里保证了 n 字节都被写入
}

• 两端协议

struct proctol{
    
    int flag = 12; 
    int fd ;
    int sockfd ;
    int threadNum ;
    long cur ;
    long  moved ;
    //文件长度
    long size ;
    //缓冲区大小
    char fileName[1024] ;
    char buf[4096] ;
} ;

• 服务器

#include"testServer.h"
#include"ReadWrite.h"

int main(int argc, char** argv)
{
    if(argc< 0){
        cout<< "use:./a.out filename" <<endl ;
        exit(1);
    }

    int fd = socket(AF_INET, SOCK_STREAM, 0) ;
    if(fd < 0) {
        return 0 ;
    }
    struct sockaddr_in sock_addr ;
    sock_addr.sin_family = AF_INET ;
    sock_addr.sin_port = htons(PORT) ;
    sock_addr.sin_addr.s_addr = INADDR_ANY ;
    int use = 1 ;
    if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &use, sizeof(use))< 0){
        cout << "socketoopt" <<endl ;  
        return 1 ;
    }   
    if(bind(fd, (struct sockaddr*)&sock_addr, sizeof(sock_addr))< 0) {
        cout << "bind" <<endl ;
        return 1 ;
    }

    if(listen(fd, BACKLOG) < 0){
        cout<< "listen" <<endl ;
        return 1 ;
    }
    
    SendFile(argv[1], fd) ;
    RecvData(fd)  ;
    return 0;
}

void RecvData(int sockfd) {
    int connfd ;
    //接受连接
    if((connfd = accept(sockfd, NULL, NULL)) < 0) {
        return ;
    }
    proctol data ;
    bzero(&data, sizeof(data)) ;
    //获取客户端向服务发送的第一个包
    int ret = readn(connfd, &data, sizeof(data)) ;
    if(ret < 0) {
        printf("readn失败！\n") ;
        exit(1) ;
    }    
    int threadNum = data.threadNum ;
    //第一个包里面含有文件名称
    close(connfd) ;
    int fd = open(data.fileName, O_CREAT|O_WRONLY, 0666) ;
    printf("线程数量%d\n",threadNum) ;
    for(int i = 0 ; i< threadNum; i++) {
        proctol data ;
        bzero(&data, sizeof(data)) ;
        printf("第%d个连接\n", i) ;
        if((connfd = accept(sockfd, NULL, NULL)) < 0) {
            printf("接收文件accept出错！\n") ;
            return ;
        }

        data.fd = fd ;
        data.sockfd = connfd ;
        std::thread t(threadRecv, std::ref(data)) ;
        t.join() ;
    }

    close(fd) ;
}

//接收每个客户端线程发送来的数据
void threadRecv(proctol& data) {
    
    int connfd = data.sockfd ;
    int fileFd = data.fd ;
    proctol tmp ;
    bzero(&tmp, sizeof(tmp)) ;
    int ret = readn(connfd, &tmp, sizeof(tmp)) ;
    if(ret < 0) {
        printf("接收功能readn读取数据错误！\n") ;
        return ;
    }
    
    long size = tmp.size ;
    long cur = tmp.cur ;
    ret = pwrite(fileFd, tmp.buf, tmp.moved, cur) ;
    if(ret < 0) {
        printf("pwrite写文件失败！\n") ;
        return ;
    }
    size -= tmp.moved ;
    cur  += tmp.moved ;

    while(size) {
        bzero(&tmp, sizeof(tmp)) ;
        ret = readn(connfd, &tmp, sizeof(tmp)) ;  
        if(ret < 0&& errno ==EINTR) {
            printf("读取文件失败！\n");
            continue ;
        }
        else if(ret == 0) {
            break ;
        }
        if(ret < 0) {
            printf("readn接收文件失败！\n") ;
            return ;
        }
        ret = pwrite(fileFd, tmp.buf, tmp.moved, cur) ;
        if(ret < 0) {
            printf("pwrite写文件失败！\n") ;
            return ;
        }
        size -= tmp.moved ;
        cur += tmp.moved ;
    }
    close(connfd) ;
}

int SendFile(const char*fileName, int sockFd) {

    struct stat st_file ;
    if(lstat(fileName, &st_file)< 0){
        cout << "发送的文件不存在!"<<endl ;
        return 1;
    }
    //为线程分配任务
    long file_block ;
    proctol pp ;
    pp.size = st_file.st_size ;
    //获取文件的大小,20mb以上为大文件
    if(pp.size>=1024*1024*50) {
        file_block = 1024*1024*10 ;     
    }
    else {
        file_block = 1024*1024*3 ;            
    }
    printf("每个线程分的块大小：%ld\n", file_block) ;
    int threadNum = pp.size/file_block ;
    //获取最后一个包的长度
    int lastpacksize = (int)pp.size-threadNum*file_block ;
    //如果最后一个包的长度不问0的话,则多创建一个线程
    if(lastpacksize) {
        threadNum++ ;
    }
    if(threadNum > 100) {
        printf("线程数量太多！\n") ;
        return 0;
    }
    int connfd = accept(sockFd, NULL, NULL) ;
    if(connfd < 0) {
        printf("接收连接accept出错!") ;
        exit(1) ;
    }
    proctol data ;
    data.size = file_block ;
    //第一个包是文件名称和线程的数量
    strcpy(data.fileName, fileName) ;
    data.sockfd = sockFd ;
    data.threadNum = threadNum ;
    int ret = writen(connfd, &data, sizeof(data)) ;
    if(ret < 0) {
        printf("发送消息失败!") ;
        exit(1) ;
    }
    close(connfd) ;
    //打开文件
    printf("文件名：%s\n", fileName) ;
    int fd = open(fileName, O_RDONLY) ;
    if(fd < 0) {
        cout << "open error" <<endl ;
        exit(1) ;
    }
    for(int i= 0; i< threadNum; i++) {
        connfd = accept(sockFd, NULL, NULL) ;
        if(connfd < 0) {
            exit(1) ;
        }
        //发送文件
        proctol data ;
        memset(data.buf, 0, sizeof(data.buf)) ;
        strcpy(data.fileName, fileName) ;
        //当线程数量不为0且最后一个包的长度不为0
        if(i == threadNum-1&&lastpacksize) {
            data.cur = i*file_block ; 
            data.size = lastpacksize ;
        }
        else {
            //计算偏移
            data.cur = i*file_block ;
            data.size = file_block ;
        }
            
        data.fd = fd ;
        data.sockfd = connfd ;
        thread t(sender, std::ref(data)) ;
        t.join() ;
    }
    close(fd) ;
    return 1 ;
}   
//发送文
void sender(proctol& data) {
    
    long cur = data.cur ;
    int connfd = data.sockfd ;
    int read_count ;
    long size = data.size ;
    long read_size = 0 ;
    int filefd = data.fd ;
    printf("传送文件的长度%ld，名称：%s\n", data.size, data.fileName) ;
    while(size){
        //原子操作
        data.moved = sizeof(data.buf) ;
        if(sizeof(data.buf) > (size_t)size) {
            data.moved = size ;
        }
        read_count = pread(filefd, data.buf, data.moved, cur+read_size) ;
        if(read_count < 0&& errno == EINTR) {
            printf("被信号打断") ;
            continue ;
        } 
        else if(read_count == 0) {
            break ;
        }   

        else if(read_count < 0) {
            printf("pread 错误!") ;
            exit(1) ;
        }
        writen(connfd, &data, sizeof(data)) ;
        size-=read_count ;
        read_size += read_count ;
    }
    close(connfd) ;
}

• 客户端

#include"testClient.h"
#include"ReadWrite.h"
struct sockaddr_in addr ;

int main(int argc, char** argv)
{

    if(argc != 3) {
        printf("usage:./a.out ip port\n") ;
        exit(1) ;
    }
    int sockfd = socket(AF_INET, SOCK_STREAM, 0) ;
    if(sockfd < 0) {
        printf("创建套接字失败!");
        exit(1) ;
    }
    const char* ip = argv[1] ;
    int port = atoi(argv[2]) ;
    
    addr.sin_family = AF_INET ;
    addr.sin_port = htons(port) ;
    inet_pton(AF_INET, ip, &addr.sin_addr) ;
    
    if(connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        printf("connect连接失败!") ;
        exit(1) ;
    }
    int threadNum = 0 ;
    proctol data ;
    int ret = readn(sockfd, &data, sizeof(data)) ;
    if(ret < 0) {
        printf("recv 接收消息失败!") ;
        exit(1) ;
    }
    threadNum = data.threadNum ;    
    if(threadNum > 100) {
        printf("线程数量太多!\n") ;
        exit(1) ;
    }

    int fd = open(data.fileName, O_CREAT|O_WRONLY, 0777) ;
    if(fd < 0) {
        printf("open创建文件出错!") ;   
        exit(1) ;
    }
    //创建线程
    for(int i = 0; i< threadNum; i++) {
        proctol data ;
        memset(&data, 0, sizeof(data)) ;
        data.fd = fd ;
        std::thread t(receive ,std::ref(data)) ;
        t.join() ;
    }
    close(fd) ;
    printf("回车开始服务器回传数据....\n");
    getchar() ;
    //客户端给服务器发送大文件
    //发给服务器后进行标识为客户端回射的
    const char* filename= data.fileName ;
    sendToServer(filename) ;
    close(sockfd) ;
    printf("接收完成!") ;
    return 0;
}

//发送文件给服务器
void sendToServer(const char*fileName) {
    //创建套接字
      int sockfd = socket(AF_INET,  SOCK_STREAM, 0) ;                 
      if(sockfd < 0) {                                                
          printf("创建套接字失败!") ;                                 
          exit(1) ;
      }                                                               
  
      if(connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
          printf("连接服务器失败!") ;                                 
          exit(1) ;
      }   
    struct stat st ;
    lstat(fileName, &st) ;
    int threadNum = 0 ;
    long block_size = 0 ;
    int lastpack = 0 ;
    long size = st.st_size ;
    //最大文件为20M
    if(size > 1024*1024*50) {
        block_size = 1024*1024*10 ;
    }
    //不超过20M的每个线程传送2M
    else {
        block_size = 1024*1024*3 ;
    }

    threadNum = size/block_size ;
    lastpack = size -threadNum*block_size ;
    //最后一个包长度不为0,多分一个线程处理   
    if(lastpack) {
        threadNum++ ;
    }
    if(threadNum > 100) {
        printf("不适合发所指定的文件！文件过大！\n");
        exit(1) ;
    }
    proctol data ;
    data.threadNum = threadNum ;

    strcpy(data.fileName, fileName) ;
    data.fileName[0] = '_' ;
    int ret = writen(sockfd, &data, sizeof(data)) ;
    if(ret < 0) {
        printf("文件信息发送失败！\n") ;
        return ;
    }
    close(sockfd) ;

    int fd = open(fileName, O_RDONLY) ;
    for(int i= 0; i< threadNum; i++) {
        sockfd = socket(AF_INET,  SOCK_STREAM, 0) ;                 
        //和服务器建立连接
        if(connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
            printf("线程创建连接失败！\n") ;
            return ;
        }         
        if(i == threadNum-1&&lastpack){
            data.cur = i*block_size ;
            data.size = lastpack ;
        }
        //判断是否到达文件长度
        else {
            data.cur = i*block_size ;
            data.size = block_size ;
        }

        data.fd = fd ;
        data.sockfd = sockfd ;
        std::thread t1(threadSend, std::ref(data)) ;
        t1.join() ;
    }
    printf("发送完成，size=%ld", size) ;
}

//开线程向服务器端发送数据
void threadSend(proctol& data) {
    
    //线程负责的范围和负责的长度
    long cur = data.cur ;
    long size = data.size ;
    int sockfd = data.sockfd ;
    int filefd = data.fd ;
    
    proctol tmp ;
    bzero(&tmp, sizeof(tmp)) ;
    tmp.size = size ;
    tmp.cur = cur ;
    int ret = 0;
    int read_count = 0 ;
    while(size) {
        int len = sizeof(data.buf) ;
        if(size<len) {
            len = size ;
        }
        ///通知对方这次在接受到数据的基础上，
        //在文件位置为cur的基础上移动moved长度
        tmp.moved = len ;
        ret = pread(filefd, tmp.buf, len, cur+read_count) ;
        if(ret < 0) {
            printf("发送数据出错！") ;
            return  ;
        }
        writen(sockfd, &tmp, sizeof(tmp)) ;

        size -= ret ;
        read_count += ret ;
    }
    close(sockfd) ;
}

void receive(proctol& data) {
    
    int sockfd = socket(AF_INET,  SOCK_STREAM, 0) ;
    if(sockfd < 0) {
        printf("创建套接字失败!") ;
        exit(1) ;
    }

    if(connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        printf("连接服务器失败!") ;
        exit(1) ;
    }
    //接收到数据
    proctol datas ;
    int ret ;
    if((ret = readn(sockfd, &datas, sizeof(datas))) < 0) {
        printf("接收数据错误！");
        exit(1) ;
    }
    long size = datas.size ;
    long cur = datas.cur ;
    if(pwrite(data.fd, datas.buf, datas.moved, cur)<0) {
        printf("写文件出错！") ;
        exit(1) ;
    }
    size -= datas.moved;
    cur += datas.moved ;
    //接收消息
    while(size) {
        ret = readn(sockfd, &datas, sizeof(datas)) ;
        if(ret <  0 && errno == EINTR) {
            printf("读取文件失败！\n") ;
            continue ;
        }
        else if (ret == 0) {
            break ;
        }
        else if(ret < 0) {
            printf("读取文件出错！") ;
            exit(1) ;
        }
        //将数据写入到文件中
        if(pwrite(data.fd, datas.buf, datas.moved, cur) < 0) {
            printf("写文件出错！")  ;  
            exit(1) ;
        }
        cur += datas.moved ;
        size -= datas.moved ;
    }
}