• linux下的单线程reactor文件传输

    先简单了解一下reactor模型：

     

    ## 具体流程如下：

    1.注册读就绪事件和相应的事件处理器

    2.事件分离器等待事件

    3.事件到来，激活分离器，分离器调用事件对应的处理器

    4.事件处理器完成实际的读操作，处理读到的数据，注册新的事件，然后返还控制权。

    ## 优点：

    1.响应快，不必为单个同步时间所阻塞，虽然reactor本身依然是同步的

    2.编程相对简单，可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销

    3可扩展性，可以方便的通过增加Reactor实例个数来充分利用CPU资源

    4.可复用性，Reactor框架本身与具体事件处理逻辑无关，具有很高的复用性。

    这里reactor有一个特殊的机制：采用回调函数，谁发生就自己调用自己去解决自己的问题；

    相比于epoll的遍历消息列表，reactor的效率会大大增加。

    这里先例举出一个简单的reactor框架

    1. #include<stdlib.h>
    2. #include<stdio.h>
    3. #include<unistd.h>
    4. #include<sys/socket.h>
    5. #include<arpa/inet.h>
    6. #include<poll.h>
    7. #include<sys/epoll.h>
    8. #include<pthread.h>
    9.  
    10. #define MAXNUM 4096
    11. typedef struct reactor
    12. {
    13.         int epfd;//树根
    14.         struct epoll_event sockary[MAXNUM];
    15.  
    16. }reactor;
    17. reactor*eventloop=NULL;
    18.  
    19. typedef struct item
    20. {
    21.         int sockfd;
    22.         void(*callback)(int,int,void*);
    23.         char szbuf[MAXNUM];
    24.         int nbuffersize;
    25. }item;
    26.  
    27. void recv_cb(int fd,int events,void *arg)
    28. {
    29.         item*pitem=(item*)arg;
    30.         int res=recv(fd,pitem->szbuf,MAXNUM,0);
    31.         if(res>0)
    32.         {
    33.                 printf("recv is :%s\n",pitem->szbuf);
    34.         }
    35.         else
    36.         {
    37.                 struct epoll_event ee;
    38.                 ee.events=EPOLLIN;
    39.                 epoll_ctl(eventloop->epfd,EPOLL_CTL_DEL,fd,&ee);
    40.                 close(fd);
    41.                 free(pitem);
    42.         }
    43.  
    44. }
    45. void accept_cb(int fd,int events,void *arg)
    46. {
    47.  
    48.         int clientfd=accept(fd,0,0);
    49.         printf("accept success");
    50.         item*pitem=(item*)malloc(sizeof(item));
    51.         pitem->sockfd=clientfd;
    52.         pitem->callback=recv_cb;
    53.         struct epoll_event ep;
    54.         ep.events=EPOLLIN|EPOLLET;
    55.         ep.data.ptr=pitem;
    56.         epoll_ctl(eventloop->epfd,EPOLL_CTL_ADD,clientfd,&ep);
    57.  
    58. }
    59.  
    60. int main()
    61. {
    62. //创建套接子socket()
    63.         int sockfd=socket(AF_INET,SOCK_STREAM,0);
    64.         if(sockfd==-1)
    65.         {
    66.         perror("socket failed...");
    67.         exit(0);
    68.         }
    69. //绑定bind()
    70.         struct sockaddr_in addr;
    71.         addr.sin_family=AF_INET;
    72.         addr.sin_addr.s_addr=0;
    73.         addr.sin_port=htons(8899);
    74.         if(-1==bind(sockfd,(struct sockaddr*)&addr,sizeof(addr)))
    75.         {
    76.         perror("bind failed...");
    77.         exit(0);
    78.         }
    79. //监听listen()
    80.         if(-1==listen(sockfd,128))
    81.         {
    82.         perror("listen failed...");
    83.         exit(0);
    84.         }
    85. //定义struct
    86.         eventloop=(reactor*)malloc(sizeof(reactor));
    87.         eventloop->epfd=epoll_create(MAXNUM);
    88.         struct epoll_event ep;
    89.         ep.events=EPOLLIN|EPOLLET;
    90.         item*pitem=(item*)malloc(sizeof(item));
    91.         pitem->sockfd=sockfd;
    92.         pitem->callback=accept_cb;
    93.         ep.data.ptr=(void*)pitem;
    94.         epoll_ctl(eventloop->epfd,EPOLL_CTL_ADD,sockfd,&ep);
    95. //连接accept()
    96.  
    97.         int nreadnum=0;
    98.         while(1)
    99.         {
    100.                 //监控的红黑数文件描述，已经发生的消息的连表，最大管理数量,不产生阻塞
    101.                 nreadnum=epoll_wait(eventloop->epfd,eventloop->sockary,MAXNUM,-1);
    102.                 int i=0;
    103.                 while(i<nreadnum)
    104.                 {
    105.                         if(eventloop->sockary[i].events&EPOLLIN)
    106.                         {
    107.                                 item*pitem=(item*)eventloop->sockary[i].data.ptr;
    108.                                 pitem->callback(pitem->sockfd,eventloop->sockary[i].events,pitem);
    109.                         }
    110.                         i++;
    111.                         sleep(1);
    112.                 }
    113.  
    114.  
    115.         }
    116. close(sockfd);
    117.  
    118. return 0;
    119. }

    这个框架仅限于于客户端对话。

    我们在这框架的基础上改进

    简单介绍一下传输逻辑：

    这里我们采用传输文件的大小，在传文件内容

    在接受的时候，我们采用先接受文件大小，在接受文件内容的方式解决了粘包的问题。

    改进后的代码如下

    服务器：

    1. #include<stdlib.h>
    2. #include<stdio.h>
    3. #include<unistd.h>
    4. #include<sys/socket.h>
    5. #include<arpa/inet.h>
    6. #include<poll.h>
    7. #include<sys/epoll.h>
    8. #include<pthread.h>
    9. #include<string.h>
    10. #define MAXNUM 4096
    11. typedef struct fileinfo
    12. {
    13.         int filesize;
    14.         char filepath[1024];
    15. }fileinfo;
    16. typedef struct reactor
    17. {
    18.         int epfd;//树根
    19.         struct epoll_event sockary[MAXNUM];
    20.  
    21. }reactor;
    22. reactor*eventloop=NULL;
    23.  
    24. typedef struct item
    25. {
    26.         int sockfd;
    27.         void(*callback)(int,int,void*);
    28.         char *szbuf;
    29.         char filepath[MAXNUM];
    30.         int nbuffersize;
    31.         int noffest;
    32. }item;
    33.  
    34. void recv_cb(int fd,int events,void *arg)
    35. {
    36.         item*pitem=(item*)arg;
    37.  
    38.  
    39. //开文件看路径是否有效
    40.         FILE*fp=fopen("/home/lzl/learn/network/test/recv/recv.txt","a");
    41.         if(NULL==fp)
    42.         {
    43.                 printf("fopen failed...\n");
    44.                 return;
    45.         }
    46. //第一次接受包大小
    47.         if(pitem->nbuffersize==0&&pitem->noffest==0)
    48.         {
    49.                 int ner=recv(fd,&pitem->nbuffersize,sizeof(int),0);
    50.                 {
    51.                         if(ner>0)
    52.                         {
    53.                                 if(pitem->nbuffersize>0)
    54.                                 {
    55.                                         printf("filesize is %d\n",pitem->nbuffersize);
    56.                                         pitem->szbuf=(char*)malloc(pitem->nbuffersize);
    57.                                         pitem->noffest=0;
    58.                                 }
    59.                         }
    60.                         else
    61.                         {
    62.                                 printf("下线了\n");
    63.                                 struct epoll_event ee;
    64.                                 ee.events=EPOLLIN;
    65.                                 epoll_ctl(eventloop->epfd,EPOLL_CTL_DEL,fd,&ee);
    66.                                 close(fd);
    67.                                 free(pitem);
    68.  
    69.                         }
    70.                 }
    71.         }
    72.         else
    73.         {
    74.                 char filecont[10240];
    75.                 while(pitem->nbuffersize)
    76.                 {
    77.                                 int res=recv(fd,filecont,pitem->nbuffersize,0);
    78.                                 if(res>0)
    79.                                 {
    80.                                         printf("recv data is %s:\n",filecont);
    81.                                        pitem->nbuffersize-=res;
    82.                                         pitem->noffest+=res;
    83.                                 }
    84.  
    85.  
    86.                 }
    87.                 fwrite(filecont,1,pitem->noffest,fp);
    88.                 fclose(fp);
    89.                 printf("recvdata success...\n");
    90.                 free(pitem->szbuf);
    91.         }
    92.  
    93.  
    94. }
    95. void accept_cb(int fd,int events,void *arg)
    96. {
    97.  
    98.         int clientfd=accept(fd,0,0);
    99.         printf("accept success");
    100.         item*pitem=(item*)malloc(sizeof(item));
    101.         pitem->sockfd=clientfd;
    102.         pitem->callback=recv_cb;
    103.         struct epoll_event ep;
    104.         ep.events=EPOLLIN|EPOLLET;
    105.         ep.data.ptr=pitem;
    106.         epoll_ctl(eventloop->epfd,EPOLL_CTL_ADD,clientfd,&ep);
    107.  
    108. }
    109.  
    110. int main()
    111. {
    112. //创建套接子socket()
    113.         int sockfd=socket(AF_INET,SOCK_STREAM,0);
    114.         if(sockfd==-1)
    115.         {
    116.        perror("socket failed...");
    117.         exit(0);
    118.         }
    119. //绑定bind()
    120.         struct sockaddr_in addr;
    121.         addr.sin_family=AF_INET;
    122.         addr.sin_addr.s_addr=0;
    123.         addr.sin_port=htons(8899);
    124.         if(-1==bind(sockfd,(struct sockaddr*)&addr,sizeof(addr)))
    125.         {
    126.         perror("bind failed...");
    127.         exit(0);
    128.         }
    129. //监听listen()
    130.         if(-1==listen(sockfd,128))
    131.         {
    132.         perror("listen failed...");
    133.         exit(0);
    134.         }
    135. //定义struct
    136.         eventloop=(reactor*)malloc(sizeof(reactor));
    137.         eventloop->epfd=epoll_create(MAXNUM);
    138.         struct epoll_event ep;
    139.         ep.events=EPOLLIN|EPOLLET;
    140.         item*pitem=(item*)malloc(sizeof(item));
    141.         pitem->sockfd=sockfd;
    142.         pitem->callback=accept_cb;
    143.         ep.data.ptr=(void*)pitem;
    144.         epoll_ctl(eventloop->epfd,EPOLL_CTL_ADD,sockfd,&ep);
    145. //连接accept()
    146.  
    147.         int nreadnum=0;
    148.         while(1)
    149.         {
    150.                 //监控的红黑数文件描述，已经发生的消息的连表，最大管理数量,不产生阻塞
    151.                 nreadnum=epoll_wait(eventloop->epfd,eventloop->sockary,MAXNUM,-1);
    152.                 int i=0;
    153.                 while(i<nreadnum)
    154.                 {
    155.                         if(eventloop->sockary[i].events&EPOLLIN)
    156.                         {
    157.                                 item*pitem=(item*)eventloop->sockary[i].data.ptr;
    158.                                 pitem->callback(pitem->sockfd,eventloop->sockary[i].events,pitem);
    159.                         }
    160.                         i++;
    161.                         sleep(1);
    162.                 }
    163.  
    164.  
    165.         }
    166. close(sockfd);
    167.  
    168. return 0;
    169. }

    客户端

    1. #include<stdio.h>
    2. #include<stdlib.h>
    3. #include<unistd.h>
    4. #include<sys/socket.h>
    5. #include<arpa/inet.h>
    6. #include<string.h>
    7. #define MAXSIZE 1024
    8. #include <sys/types.h>
    9. #include <sys/stat.h>
    10. #include <fcntl.h>
    11.  
    12. typedef struct file
    13. {
    14.         int filesize;
    15.         char* filecont;
    16. }file;
    17.  
    18. int main()
    19. {
    20.         //socket()
    21.         int clientfd=socket(AF_INET,SOCK_STREAM,0);
    22.         //connect()
    23.         struct sockaddr_in addr;
    24.         addr.sin_family=AF_INET;
    25.         addr.sin_addr.s_addr=inet_addr("127.0.0.1");
    26.         addr.sin_port=htons(8899);
    27.         if(-1==connect(clientfd,(struct sockaddr*)&addr,sizeof(addr)))
    28.         {
    29.                 perror("connect failed");
    30.                 exit(0);
    31.         }
    32.  
    33.         size_t nerl;
    34.         FILE*fp=fopen("/home/lzl/learn/network/test/send/send.txt","r");
    35.  
    36.         file sfile;
    37.         sfile.filesize=0;
    38.         sfile.filecont=NULL;
    39.         int bigfilesize=0;
    40.         fseek(fp,0,SEEK_END);
    41.         bigfilesize=ftell(fp);
    42.         printf("bigsize is %d\n",bigfilesize);
    43.         fseek(fp,0,SEEK_SET);
    44.         sfile.filecont=(char*)malloc(sizeof(char)*bigfilesize);
    45.         while((nerl=fread(sfile.filecont,1,1024,fp))>0)
    46.         {
    47.  
    48.  
    49.                         sfile.filesize=bigfilesize;
    50.                         printf("filecont is %s\n",sfile.filecont);
    51.                         printf("filesize is %d\n",sfile.filesize);
    52.  
    53.                         if(send(clientfd,(char*)&sfile.filesize,sizeof(int),0)<0)
    54.                         {
    55.                                 printf("file send failed...\n");
    56.                                 break;
    57.                         }
    58.                         if(send(clientfd,(char*)sfile.filecont,bigfilesize,0)<0)
    59.                         {
    60.                                 printf("file send failed...\n");
    61.                                 break;
    62.                         }
    63.  
    64.         }sleep(1);
    65.         free(sfile.filecont);
    66.         fclose(fp);
    67.         close(clientfd);
    68.  
    69.         return 0;
    70. }

  • 相关阅读:
    .net framework中webapi使用swagger进行接口文档展示
    js数组介绍：创建、length的用法、冒泡排序法、选择排序法
    别用Excel、Python了，我找到了实现报表自动化最快的方法
    只依赖OPENCV的工作服安全帽检测YOLOV8S
    多层神经网络和激活函数
    jdk 8-future 异步处理-轮询isDone判断完成-springBoot示例
    网络安全攻防：ZigBee安全
    Hexo博客使用aplayer音乐播放插件
    如何用jxTMS开发一个功能（二）
    物理内存 虚拟内存 页映射模式
  • 原文地址：https://blog.csdn.net/van9527/article/details/127769530