boost asio库单线程异步发送乱序 贴代码.....

ruchong

1. #include <iostream>
   
2. #include <istream>
   
3. #include <ostream>
   
4. #include <fstream>
   
5. #include <string>
   
6. #include <boost/asio.hpp>
   
7. #include <boost/bind.hpp>
   
8. #include <deque>
   
9. #include <queue>
   
10. #include <boost/asio.hpp>
    
11. #include <boost/bind.hpp>
    
12. #include <boost/noncopyable.hpp>
    
13. #include <boost/thread.hpp>
    
14. #include <boost/lexical_cast.hpp>
    
15. #include <boost/shared_ptr.hpp>
    
16. 
    
17. using boost::asio::ip::tcp;
    
18. /**
    
19. * * @brief 发送数据解构体
    
20. * */
    
21. struct SSendDataBuff
    
22. {
    
23.     SSendDataBuff(const char *s, size_t n) : pData(NULL), nSize(0)
    
24.     {
    
25.         pData = new char[n];
    
26.         memset(pData, 0, n);
    
27.         memcpy(pData, s, n);
    
28.         nSize = n;
    
29.     }
    
30. 
    
31.     ~SSendDataBuff()
    
32.     {
    
33.         if(pData)
    
34.         {
    
35.             delete [] pData;
    
36.             pData = NULL;
    
37.         }
    
38.         nSize = 0;
    
39.     }
    
40. 
    
41.     char *pData;      //数据
    
42.     size_t nSize;     //数据长度
    
43. };
    
44. class client
    
45. {
    
46.     public:
    
47.         client(boost::asio::io_service& io_service)
    
48.             :socket_(io_service), resolver(io_service)//, m_io_work(io_service)
    
49.     {
    
50.         m_bHasSend = true;
    
51.         tcp::resolver::query query("192.168.0.226", "9988");
    
52.         tcp::resolver::iterator iterator = resolver.resolve(query);
    
53. 
    
54.         tcp::endpoint endpoint = *iterator;
    
55.         socket_.async_connect(endpoint,
    
56.                 boost::bind(&client::__handle_connect, this,
    
57.                     boost::asio::placeholders::error));
    
58.     }
    
59. void __handle_connect(const boost::system::error_code& err)
    
60.         {
    
61.             if(err)         //连接失败
    
62.             {
    
63.                 printf("Connect is faild ....\n");
    
64.             }
    
65.             {
    
66.                 printf("Connect is Successful ....\n");
    
67.                 FILE *pt = fopen("./libs", "rb");
    
68.                 if(NULL == pt)
    
69.                 {
    
70.                     assert(0);
    
71.                 }
    
72. 
    
73.                 char buff[1024];
    
74.                 int nLen = 0;
    
75.                 memset(buff, 0, sizeof(buff));
    
76.                 while((nLen = fread(buff, sizeof(char), sizeof(buff), pt)) != 0)
    
77.                 {
    
78.                     sends(buff, nLen);
    
79.                     memset(buff, 0, sizeof(buff));
    
80.                 }
    
81.                 printf("===============\n");
    
82.                 fclose(pt);
    
83.             }
    
84.         }
    
85. void sends(const char *pData, size_t nSize)
    
86.         {
    
87.             boost::mutex::scoped_lock AsynSendlock(m_SendQueMutex);
    
88.             if(m_bHasSend)      //可以直接发送不需要放到缓冲区
    
89.             {
    
90.                 __asynSendData(pData, nSize);
    
91.                 m_bHasSend = false;
    
92.             }
    
93.             else
    
94.             {
    
95.                 SSendDataBuff *pSendDataBuffer = new SSendDataBuff(pData, nSize);
    
96.                 m_AsynSendQue.push(pSendDataBuffer);
    
97.             }
    
98.         }
    
99. void __asynSendData(const char *pData, size_t nSize)
    
100.         {
     
101.         #if 0
     
102.             static FILE *pt = fopen("./TEST_WRITE", "wb");
     
103.             if(NULL == pt)
     
104.             {
     
105.                 assert(0);
     
106.             }
     
107. 
     
108.             size_t nLen = fwrite(pData, sizeof(char), nSize, pt);
     
109.             if(nLen != nSize)
     
110.             {
     
111.                 assert(0);
     
112.             }
     
113.             fflush(pt);
     
114.             #endif
     
115. 
     
116.             async_write(socket_, boost::asio::buffer(pData, nSize),
     
117.                     boost::bind(&client::__handle_tcp_write, this,
     
118.                         boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
     
119.         }
     
120. void __handle_tcp_write(const boost::system::error_code& e, size_t nSize)
     
121.         {
     
122.             printf("__handle_tcp_write ... nSize = %d\n", nSize);
     
123.             if(e)         //写错误
     
124.             {
     
125.                 assert(0);
     
126.                 return;
     
127.             }
     
128. 
     
129.             boost::mutex::scoped_lock AsynSendlock(m_SendQueMutex);
     
130.             if(!m_AsynSendQue.empty())     //如果发送队列不为空发送数据
     
131.             {
     
132.                 SSendDataBuff *p = m_AsynSendQue.front();
     
133.                 __asynSendData(p->pData, p->nSize);
     
134.                 delete p;
     
135.                 p = NULL;
     
136. 
     
137.                 //清除
     
138.                 m_AsynSendQue.pop();
     
139.             }
     
140.             else
     
141.             {
     
142.                 m_bHasSend = true;
     
143.             }
     
144.         }
     
145. private:
     
146.         tcp::socket socket_;
     
147.         tcp::resolver resolver;
     
148.         //boost::asio::io_service::work m_io_work;
     
149. 
     
150.         volatile bool m_bHasSend;                   //标识数据是否直接发送还是放到缓冲区
     
151.         std::queue<SSendDataBuff *> m_AsynSendQue;       //保存异步发送数据队列
     
152.         boost::mutex m_SendQueMutex;                //异步发送数据队列锁
     
153. };
     
154. 
     
155. int main(int argc, char* argv[])
     
156. {
     
157.     system("rm -rf TEST_WRITE");
     
158.     try
     
159.     {
     
160.         boost::asio::io_service io_service;
     
161.         client c(io_service);
     
162.         io_service.run();
     
163.         printf("HELLO ........\n");
     
164.     }
     
165.     catch (std::exception& e)
     
166.     {
     
167.         std::cout << "Exception: " << e.what() << "\n";
     
168.     }
     
169. 
     
170.     return 0;
     
171. }
     

复制代码

传一个文件libs大概50多M ， 传完后比较了下md5值, 有时候一样有时候不一样, 但大小都一样.....

服务器代码如下, 非常简单的一个原始套接口

1. #include <assert.h>
   
2. #include <stdio.h>
   
3. #include <fcntl.h>
   
4. #include <string.h>
   
5. #include <stdlib.h>
   
6. #include <pthread.h>
   
7. #include <iostream>
   
8. #include <sstream>
   
9. #include <sys/types.h>
   
10. #include <sys/socket.h>
    
11. #include <netinet/in.h>
    
12. #include <sys/types.h>
    
13. #include <sys/socket.h>
    
14. #include <arpa/inet.h>
    
15. 
    
16. int Bind(const char *ip, unsigned int nPort)
    
17. {
    
18.     int nSockfd = ::socket(PF_INET, SOCK_STREAM, 0);
    
19.     if(-1 == nSockfd)
    
20.     {
    
21.         assert(0);
    
22.     }
    
23. //解构体初始化
    
24.     struct sockaddr_in servaddr;
    
25.     bzero(&servaddr, sizeof(servaddr));
    
26.     servaddr.sin_family = PF_INET;
    
27.     if(NULL == ip)
    
28.     {
    
29.         servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
30.     }
    
31.     else
    
32.     {
    
33.         int nRet = inet_pton(PF_INET, ip, &servaddr.sin_addr);
    
34.         if(-1 == nRet)
    
35.         {
    
36.             assert(0);
    
37.         }
    
38.     }
    
39.     servaddr.sin_port = htons(nPort);
    
40. 
    
41.     int nRet = ::bind(nSockfd, (struct sockaddr*)&servaddr, sizeof(servaddr));
    
42.     if(-1 == nRet)
    
43.     {
    
44.         assert(0);
    
45.     }
    
46. 
    
47.     nRet = ::listen(nSockfd, 256);
    
48.     if(-1 == nRet)
    
49.     {
    
50.         assert(0);
    
51.     }
    
52. return nSockfd;
    
53. }
    
54. 
    
55. int main()
    
56. {
    
57.     system("rm -rf TEST_WRITE");
    
58.     int nSockfd = Bind("192.168.0.226", 9988);
    
59.     int nClientSockfd = 0;
    
60.     while(1)
    
61.     {
    
62.         nClientSockfd = ::accept(nSockfd, NULL, NULL);
    
63.         if(-1 == nClientSockfd)
    
64.         {
    
65.             if(errno == EINTR)
    
66.             {
    
67.                 continue;
    
68.             }
    
69.             return -1;
    
70.         }
    
71.         else
    
72.         {
    
73.             printf("new Client .........\n");
    
74.             break;
    
75.         }
    
76.     }
    
77. while(1)
    
78.     {
    
79.         char buff[65535];
    
80.         int nLen = recv(nClientSockfd, buff, sizeof(buff), 0);
    
81.         if(-1 == nLen)
    
82.         {
    
83.             assert(0);
    
84.         }
    
85. 
    
86.         static FILE *pt = fopen("./TEST_WRITE", "wb");
    
87.         if(NULL == pt)
    
88.         {
    
89.             assert(0);
    
90.         }
    
91. 
    
92.         size_t n = fwrite(buff, sizeof(char), nLen, pt);
    
93.         if(n != nLen)
    
94.         {
    
95.             assert(0);
    
96.         }
    
97.         fflush(pt);
    
98.     }
    
99. 
    
100.     return 0;
     
101. }

复制代码

请各位朋友帮个忙， 多谢了.......

ruchong

难道大家都没有用boost吗....大家帮帮忙 ...
多谢....

ideawu

不要在sends里把消息加入队列, 直接调用异步发送就行了。

ruchong

楼上的兄弟， 直接异步发送乱序的可能性更大.....

难道大家都不用boost的吗....
非常着急多谢大家....

hiheasy

SSendDataBuff *p = m_AsynSendQue.front();

                __asynSendData(p->pData, p->nSize);

                delete p;  //发送是异步的，这里不能直接删除对象，造成发送的数据出错
                p = NULL;