当前位置：首页 → 问答吧 → boost asio库单线程异步发送乱序贴代码.....

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

时间：2010-09-01

来源：互联网

#include <iostream>
#include <istream>
#include <ostream>
#include <fstream>
#include <string>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <deque>
#include <queue>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/noncopyable.hpp>
#include <boost/thread.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/shared_ptr.hpp>
using boost::asio::ip::tcp;
/**
* * @brief 发送数据解构体
* */
struct SSendDataBuff
{
SSendDataBuff(const char *s, size_t n) : pData(NULL), nSize(0)
{
pData = new char[n];
memset(pData, 0, n);
memcpy(pData, s, n);
nSize = n;
}
~SSendDataBuff()
{
if(pData)
{
delete [] pData;
pData = NULL;
}
nSize = 0;
}
char *pData; //数据
size_t nSize; //数据长度
};
class client
{
public:
client(boost::asio::io_service& io_service)
:socket_(io_service), resolver(io_service)//, m_io_work(io_service)
{
m_bHasSend = true;
tcp::resolver::query query("192.168.0.226", "9988");
tcp::resolver::iterator iterator = resolver.resolve(query);
tcp::endpoint endpoint = *iterator;
socket_.async_connect(endpoint,
boost::bind(&client::__handle_connect, this,
boost::asio::placeholders::error));
}
void __handle_connect(const boost::system::error_code& err)
{
if(err) //连接失败
{
printf("Connect is faild ....\n");
}
{
printf("Connect is Successful ....\n");
FILE *pt = fopen("./libs", "rb");
if(NULL == pt)
{
assert(0);
}
char buff[1024];
int nLen = 0;
memset(buff, 0, sizeof(buff));
while((nLen = fread(buff, sizeof(char), sizeof(buff), pt)) != 0)
{
sends(buff, nLen);
memset(buff, 0, sizeof(buff));
}
printf("===============\n");
fclose(pt);
}
}
void sends(const char *pData, size_t nSize)
{
boost::mutex::scoped_lock AsynSendlock(m_SendQueMutex);
if(m_bHasSend) //可以直接发送不需要放到缓冲区
{
__asynSendData(pData, nSize);
m_bHasSend = false;
}
else
{
SSendDataBuff *pSendDataBuffer = new SSendDataBuff(pData, nSize);
m_AsynSendQue.push(pSendDataBuffer);
}
}
void __asynSendData(const char *pData, size_t nSize)
{
#if 0
static FILE *pt = fopen("./TEST_WRITE", "wb");
if(NULL == pt)
{
assert(0);
}
size_t nLen = fwrite(pData, sizeof(char), nSize, pt);
if(nLen != nSize)
{
assert(0);
}
fflush(pt);
#endif
async_write(socket_, boost::asio::buffer(pData, nSize),
boost::bind(&client::__handle_tcp_write, this,
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
}
void __handle_tcp_write(const boost::system::error_code& e, size_t nSize)
{
printf("__handle_tcp_write ... nSize = %d\n", nSize);
if(e) //写错误
{
assert(0);
return;
}
boost::mutex::scoped_lock AsynSendlock(m_SendQueMutex);
if(!m_AsynSendQue.empty()) //如果发送队列不为空发送数据
{
SSendDataBuff *p = m_AsynSendQue.front();
__asynSendData(p->pData, p->nSize);
delete p;
p = NULL;
//清除
m_AsynSendQue.pop();
}
else
{
m_bHasSend = true;
}
}
private:
tcp::socket socket_;
tcp::resolver resolver;
//boost::asio::io_service::work m_io_work;
volatile bool m_bHasSend; //标识数据是否直接发送还是放到缓冲区
std::queue<SSendDataBuff *> m_AsynSendQue; //保存异步发送数据队列
boost::mutex m_SendQueMutex; //异步发送数据队列锁
};
int main(int argc, char* argv[])
{
system("rm -rf TEST_WRITE");
try
{
boost::asio::io_service io_service;
client c(io_service);
io_service.run();
printf("HELLO ........\n");
}
catch (std::exception& e)
{
std::cout << "Exception: " << e.what() << "\n";
}
return 0;
}

复制代码