阅读285 返回首页    go 阿里云 go 技术社区[云栖]

基于流模式长连接的简单实现

基于流模式的长连接我们可以做很多事情，比方说在局域网内，我们建立这种模式，可以时时传输数据，而不用每次传输数据是创建socket，传输完后，关闭socket。可以减少创建销毁socket的时间。
基于流模式的下，数据一直在发送，我们需要知道，每次发送数据量，所以常见的方式时，我们在发送数据时，指定此次发送数据的长度，服务器读取流数据时，先读取数据长度，然后再按长度读取此次发送的数据。
我使用select复用IO机制实现了一个简单的client,server机制，希望对初学者有帮助
客户端代码：
  1. /*
  2.  * =====================================================================================
  3.  *
  4.  *       Filename:  client.cc
  5.  *    Description:  
  6.  *        Version:  1.0
  7.  *        Created:  2008年12月18日 09时50分36秒 CST
  8.  *       Revision:  none *
  9.  *         Author:  ugg (ugg_xchj@yahoo.com.cn)
  10.  *        Company:  
  11.  *
  12.  * =====================================================================================
  13.  */
  15. #include <string>
  16. #include <iostream>
  17. #include <netdb.h>
  18. #include <sys/socket.h>
  19. #include <sys/types.h>
  20. #include <netinet/in.h>
  21. #include <arpa/inet.h>
  23. using namespace::std;
  25. // 默认内容设置
  26. string hostname="localhost";
  27. int    hostport=7763;
  28. string sendContents="this is client";
  30. void getCMD(int argc, char *argv[])
  31. {
  32.     switch(argc)
  33.     {
  34.         case 2:
  35.             hostname = argv[1];
  36.            break;
  37.         case 3:
  38.             hostname = argv[1];
  39.             hostport = atoi(argv[2]);   
  40.             if(hostport < 1024 || hostport >65535)
  41.             {
  42.                 cerr << "Error: port=" << hostport 
  43.                     << " Error, range 1024 - 65535" << endl;
  44.                 exit(0);
  45.             }
  46.             break;
  47.         case 4:
  48.             hostname = argv[1];
  49.             hostport = atoi(argv[2]);   
  50.             if(hostport < 1024 || hostport >65535)
  51.             {
  52.                 cerr << "Error: port=" << hostport 
  53.                     << " Error, range 1024 - 65535" << endl;
  54.                 exit(0);
  55.             }
  56.             sendContents = argv[3]; 
  57.             break;
  58.         default:
  59.             break;
  60.     }
  61. }   
  63. int
  64. main ( int argc, char *argv[] )
  65. {
  66.     getCMD(argc,argv);
  67.     
  68.     int fd; 
  69.     // create socket
  70.     if((fd=socket(PF_INET,SOCK_STREAM,0)) == -1)
  71.     {
  72.         cerr << "Error: socket()" <<endl;
  73.         exit(0);
  74.     }
  75.     // 
  76.     struct hostent *he;
  77.     he = gethostbyname(hostname.c_str());
  78.     if(he == NULL ) {
  79.         cerr <<"Error: gethostbyname() error,hostname=" << hostname << endl;
  80.         exit(0);
  81.     }
  83.     struct sockaddr_in serv_addr;
  85.     serv_addr.sin_family=AF_INET;
  86.     serv_addr.sin_port=htons(hostport);
  87.     serv_addr.sin_addr=*((struct in_addr*)he->h_addr);
  88.     bzero( &(serv_addr.sin_zero),8);
  90.     // connction
  91.     if(connect(fd,(struct sockaddr*)&serv_addr,sizeof(serv_addr))==-1) {
  92.         cerr<<"Error: connect() error"<<endl;
  93.         exit(0);
  94.     }
  96.     cout << "send contents to server(" << hostname 
  97.         << ":" << hostport << ")" << endl;
  98.     cout << sendContents << endl;
  99.     cout << "................" << endl;
  101.     // 增加此次发送消息的长度
  102.     char buffer[1024];      
  103.     sprintf(buffer,"%8d",sendContents.length());
  105.     sendContents = string(buffer,8)+sendContents;
  106.     
  107.     const char* content = sendContents.c_str();
  108.     int send=0;
  109.     int length = sendContents.length();
  110.     
  111.     // send
  112.     while(1)
  113.     {
  114.         int ret = write(fd,content+send,length-send);
  115.         if(ret == 0)
  116.         {
  117.             // server close
  118.             cerr << "Error: server close" << endl;
  119.             exit(0);
  120.         }
  121.         send += ret;
  122.         if(length == send)
  123.             break;
  124.     }
  126.     string contents;
  127.     // 先接受8字节，获取服务返回长度
  128.     int ret = recv(fd,buffer,8,0);
  129.     // 服务器关闭
  130.     if(ret == 0)
  131.     {
  132.         cerr << "Error: server close" << endl;  
  133.         exit(0);
  134.     }
  135.     if(ret == 8)
  136.     {
  137.         buffer[8]='/0';
  138.         int len = atoi(buffer);
  139.         if(len < 1024){
  140.             int rets=0;
  141.             while((ret = recv(fd,buffer,len-rets,0))>0)
  142.             {
  143.                 contents.append(buffer,ret);
  144.                 rets += ret;
  145.                 if(len==rets)
  146.                     break;
  147.             }
  148.         }else{
  149.             int buflen=1024;
  150.             while( (ret = recv(fd,buffer,buflen,0)) > 0)
  151.             {
  152.                 contents.append(buffer,ret);
  153.                 len-=ret;
  154.                 if(len<2048){
  155.                     buflen=len;
  156.                 }
  157.                 if(len <= 0)
  158.                     break;
  159.             }
  160.         }
  161.     }else {
  162.         cerr << "Error: recv data Error " << endl;
  163.     }
  165.     cout << "recv: ";
  166.     cout << contents << endl;
  168.     return 0;
  169. }               /* ----------  end of function main  ---------- */
服务器端代码

  1. /*
  2.  * =====================================================================================
  3.  *
  4.  *       Filename:  server.cc
  5.  *    Description:  
  6.  *        Version:  1.0
  7.  *        Created:  2008年12月18日 09时50分50秒 CST
  8.  *       Revision:  none *
  9.  *         Author:  ugg (ugg_xchj@yahoo.com.cn)
  10.  *        Company:  
  11.  *
  12.  * =====================================================================================
  13.  */
  14. #include <string>
  15. #include <iostream>
  16. #include <netdb.h>
  17. #include <sys/socket.h>
  18. #include <sys/types.h>
  19. #include <netinet/in.h>
  20. #include <arpa/inet.h>
  21. #include <errno.h>
  22. #include <map>
  24. using namespace::std;
  26. struct clientInfo
  27. {
  28.     string host;
  29.     int port;
  30. };
  32. typedef map<int, clientInfo> mapgroups;
  33. typedef map<int, clientInfo>::iterator mapgroupsor;
  34. typedef map<int, clientInfo>::const_iterator mapgroupscor;
  36. //  记录客户端的信息
  37. mapgroups groups;
  39. // 默认内容设置
  40. string hostname="localhost";
  41. int    hostport=7763;
  42. string serversendContents="this is server";
  44. void getCMD(int argc, char *argv[])
  45. {
  46.     switch(argc)
  47.     {
  48.         case 2:
  49.             hostname = argv[1];
  50.            break;
  51.         case 3:
  52.             hostname = argv[1];
  53.             hostport = atoi(argv[2]);   
  54.             if(hostport < 1024 || hostport >65535)
  55.             {
  56.                 cerr << "Error: port=" << hostport 
  57.                     << " Error, range 1024 - 65535" << endl;
  58.                 exit(0);
  59.             }
  60.             break;
  61.         case 4:
  62.             hostname = argv[1];
  63.             hostport = atoi(argv[2]);   
  64.             if(hostport < 1024 || hostport >65535)
  65.             {
  66.                 cerr << "Error: port=" << hostport 
  67.                     << " Error, range 1024 - 65535" << endl;
  68.                 exit(0);
  69.             }
  70.             serversendContents = argv[3];   
  71.             break;
  72.         default:
  73.             break;
  74.     }
  75. }   
  77. void clearfd(int fd,fd_set& rdfds)
  78. {
  79.     FD_CLR(fd,&rdfds);  
  80.     mapgroupsor it = groups.find(fd);
  81.     if(it != groups.end())
  82.     {
  83.         cerr << "client host=" << it->second.host << ", port=" << it->second.port 
  84.             << " close" << endl;
  85.         groups.erase(fd);
  86.     }
  87. }
  89. // recv and send message
  90. void recvandsend(int fd,fd_set& rdfds,string& contents, const string& texts)
  91. {
  92.     // 接受消息
  93.     char buffer[1024];
  94.     int rets=0;
  95.     int ret=0;
  96.     ret = recv(fd,buffer,8,0);
  98.     if(ret == 0)
  99.     {
  100.         clearfd(fd,rdfds);
  101.         return;
  102.     }
  103.     if(ret == 8)
  104.     {
  105.         buffer[8]='/0';
  106.         int len = atoi(buffer);
  107.         if(len < 1024){
  108.             while((ret = recv(fd,buffer,len,0))>0)
  109.             {
  110.                 contents.append(buffer,ret);
  111.                 rets += ret;
  112.                 len-=ret;
  113.                 if(len==0)
  114.                     break;
  115.             }
  116.             if(ret == 0){
  117.                 clearfd(fd,rdfds);
  118.                 return;
  119.             }
  120.         }else{
  121.             int buflen=1024;
  122.             while( (ret = recv(fd,buffer,buflen,0)) > 0)
  123.             {
  124.                 contents.append(buffer,ret);
  125.                 rets += ret;
  126.                 len-=ret;
  127.                 if(len<2048){
  128.                     buflen=len;
  129.                 }
  130.                 if(len <= 0)
  131.                     break;
  132.             }
  133.             if(ret == 0){
  134.                 clearfd(fd,rdfds);
  135.                 return;
  136.             }
  137.         }
  138.     }else{
  139.         clearfd(fd,rdfds);
  140.         return;
  141.     }
  143.     // 发送信息
  144.     sprintf(buffer,"%8d",texts.length());   
  145.     string sendContents = string(buffer,8);
  146.     sendContents += texts;
  148.     const char* content = sendContents.c_str();
  149.     int send=0;
  150.     int length = sendContents.length();
  152.     // send
  153.     while(1)
  154.     {
  155.         int ret = write(fd,content+send,length-send);
  156.         if(ret == 0)
  157.         {
  158.             // server close
  159.             cerr << "Error: server close" << endl;
  160.             clearfd(fd,rdfds);
  161.             return;
  162.         }
  163.         send += ret;
  164.         if(length == send)
  165.             break;
  166.     }
  167. }   
  169. int
  170. main ( int argc, char *argv[] )
  171. {
  172.     getCMD(argc,argv);
  173.     
  174.     int fd; 
  175.     // create socket
  176.     if((fd=socket(PF_INET,SOCK_STREAM,0)) == -1)
  177.     {
  178.         cerr << "Error: socket()" <<endl;
  179.         exit(0);
  180.     }
  181.     // 
  182.     struct hostent *he;
  183.     he = gethostbyname(hostname.c_str());
  184.     if(he == NULL ) {
  185.         cerr <<"Error: gethostbyname() error,hostname=" << hostname << endl;
  186.         exit(0);
  187.     }
  189.     struct sockaddr_in serv_addr;
  191.     serv_addr.sin_family=AF_INET;
  192.     serv_addr.sin_port=htons(hostport);
  193.     serv_addr.sin_addr=*((struct in_addr*)he->h_addr);
  194.     bzero( &(serv_addr.sin_zero),8);
  196.     // bind
  197.     if(bind(fd,(struct sockaddr*)&serv_addr,sizeof(serv_addr))==-1) {
  198.         cerr<<"Error: bind() error"<<endl;
  199.         exit(0);
  200.     }
  202.     // listen
  203.     int ret_listen = listen(fd,5);
  204.     if(ret_listen < 0)
  205.     {
  206.         cerr << "Error: Listen port error,socket fd: " << fd << endl;
  207.         cerr << "Error: listen errno = " << ret_listen << endl;
  208.         exit(0);
  209.     }
  211.     fd_set rdfds;
  212.     int nfds;
  214.     // First poll the sockets
  215.     FD_ZERO(&rdfds);
  216.     FD_SET(fd, &rdfds);
  217.     while(1)
  218.     {
  219.         if ((nfds = select(FD_SETSIZE, &rdfds, NULL, NULL, NULL)) < 0) {
  220.             if (errno != EINTR) {
  221.                 cerr << "Error: select error" << endl;
  222.                 exit(3);
  223.             }
  224.         }
  226.         if(FD_ISSET(fd, &rdfds))
  227.         {
  228.             socklen_t len;
  229.             struct sockaddr_in client_addr;
  230.             len = sizeof(struct sockaddr);
  231.             int clientfd = accept(fd,(struct sockaddr *)&client_addr,&len);
  232.             if(clientfd == -1)
  233.             {
  234.                 cerr << "Error: accept client error" << endl;
  235.             }else{
  236.                 clientInfo info;    
  237.                 info.host = inet_ntoa(client_addr.sin_addr);
  238.                 info.port = ntohs(client_addr.sin_port);
  239.                 cerr << "server: got connection from "<< info.host 
  240.                     <<", port "<< info.port << endl;
  241.                 groups.insert(make_pair(clientfd,info));
  242.                 FD_SET(clientfd,&rdfds);
  243.             }
  244.         }else{
  245.             mapgroupscor it;
  246.             for(it = groups.begin(); it != groups.end(); ++it)
  247.             {
  248.                 if(FD_ISSET(it->first,&rdfds))
  249.                 {
  250.                     string strRev;
  251.                     recvandsend(it->first,rdfds,strRev,serversendContents);         
  252.                     break;
  253.                 }
  254.             }
  255.         }
  257.     }   
  259.     return 0;
  260. }               /* ----------  end of function main  ---------- */
上面的代码在linux下已经编译通过

另外你可以到这里下载源代码

最后更新：2017-04-02 00:06:39

  上一篇：go 在C#中获取Access数据库中的所有表名和列名
  下一篇：go Reporting Service在Web Application中的应用