gpt4 book ai didi

c++ - boost 异步 tcp 客户端

转载 作者:塔克拉玛干 更新时间:2023-11-03 00:40:46 30 4
gpt4 key购买 nike

我刚刚开始使用 boost。我正在使用异步套接字编写 TCP 客户端-服务器。

任务如下:

  1. 客户端向服务器发送一个数字
  2. 客户端可以在收到服务器的回答之前发送另一个数字。
  3. 服务器收到一个数字,用它做一些计算并将结果发送回客户端。
  4. 多个客户端可以连接到服务器。

现在可以执行以下操作

  • 从客户端发送一个数字到服务器
  • 服务器在当前线程中接收到一个数字并在 OnReceive 处理程序中进行计算(我知道这很糟糕......但是我应该如何启动一个新线程来并行计算)
  • 服务器返回应答但客户端已经断开连接

如何让客户端在键盘输入数字的同时等待服务器的响应?

为什么我的客户端不等待服务器的响应?

客户端代码:

using boost::asio::ip::tcp;

class TCPClient
{
public:
TCPClient(boost::asio::io_service& IO_Service, tcp::resolver::iterator EndPointIter);
void Close();

private:
boost::asio::io_service& m_IOService;
tcp::socket m_Socket;

string m_SendBuffer;
static const size_t m_BufLen = 100;
char m_RecieveBuffer[m_BufLen*2];

void OnConnect(const boost::system::error_code& ErrorCode, tcp::resolver::iterator EndPointIter);
void OnReceive(const boost::system::error_code& ErrorCode);
void OnSend(const boost::system::error_code& ErrorCode);
void DoClose();
};

TCPClient::TCPClient(boost::asio::io_service& IO_Service, tcp::resolver::iterator EndPointIter)
: m_IOService(IO_Service), m_Socket(IO_Service), m_SendBuffer("")
{
tcp::endpoint EndPoint = *EndPointIter;

m_Socket.async_connect(EndPoint,
boost::bind(&TCPClient::OnConnect, this, boost::asio::placeholders::error, ++EndPointIter));
}

void TCPClient::Close()
{
m_IOService.post(
boost::bind(&TCPClient::DoClose, this));
}
void TCPClient::OnConnect(const boost::system::error_code& ErrorCode, tcp::resolver::iterator EndPointIter)
{
cout << "OnConnect..." << endl;
if (ErrorCode == 0)
{
cin >> m_SendBuffer;
cout << "Entered: " << m_SendBuffer << endl;
m_SendBuffer += "\0";

m_Socket.async_send(boost::asio::buffer(m_SendBuffer.c_str(),m_SendBuffer.length()+1),
boost::bind(&TCPClient::OnSend, this,
boost::asio::placeholders::error));
}
else if (EndPointIter != tcp::resolver::iterator())
{
m_Socket.close();
tcp::endpoint EndPoint = *EndPointIter;

m_Socket.async_connect(EndPoint,
boost::bind(&TCPClient::OnConnect, this, boost::asio::placeholders::error, ++EndPointIter));
}
}

void TCPClient::OnReceive(const boost::system::error_code& ErrorCode)
{
cout << "receiving..." << endl;
if (ErrorCode == 0)
{
cout << m_RecieveBuffer << endl;

m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer, m_BufLen),
boost::bind(&TCPClient::OnReceive, this, boost::asio::placeholders::error));
}
else
{
cout << "ERROR! OnReceive..." << endl;
DoClose();
}
}

void TCPClient::OnSend(const boost::system::error_code& ErrorCode)
{
cout << "sending..." << endl;
if (!ErrorCode)
{
cout << "\""<< m_SendBuffer <<"\" has been sent" << endl;
m_SendBuffer = "";

m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer, m_BufLen),
boost::bind(&TCPClient::OnReceive, this, boost::asio::placeholders::error));
}
else
{
cout << "OnSend closing" << endl;
DoClose();
}

}

void TCPClient::DoClose()
{
m_Socket.close();
}

int main()
{
try
{
cout << "Client is starting..." << endl;
boost::asio::io_service IO_Service;

tcp::resolver Resolver(IO_Service);

string port = "13";
tcp::resolver::query Query("127.0.0.1", port);

tcp::resolver::iterator EndPointIterator = Resolver.resolve(Query);

TCPClient Client(IO_Service, EndPointIterator);

cout << "Client is started!" << endl;

cout << "Enter a query string " << endl;

boost::thread ClientThread(boost::bind(&boost::asio::io_service::run, &IO_Service));

Client.Close();
ClientThread.join();
}
catch (exception& e)
{
cerr << e.what() << endl;
}

cout << "\nClosing";
getch();
}

这是控制台的输出

Client is starting...
Client is started!
OnConnect...
12
Entered: 12
sending...
"12" has been sent
receiving...
ERROR! OnReceive...

Closing

服务器部分

class Session
{
public:
Session(boost::asio::io_service& io_service)
: socket_(io_service)
{
dataRx[0] = '\0';
dataTx[0] = '\0';
}

tcp::socket& socket()
{
return socket_;
}

void start()
{
socket_.async_read_some(boost::asio::buffer(dataRx, max_length),
boost::bind(&Session::handle_read, this,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}

void handle_read(const boost::system::error_code& error, size_t bytes_transferred)
{
cout << "reading..." << endl;
cout << "Data: " << dataRx << endl;

if (!error)
{
if (!isValidData())
{
cout << "Bad data!" << endl;
sprintf(dataTx, "Bad data!\0");
dataRx[0] = '\0';
}
else
{
sprintf(dataTx, getFactorization().c_str());
dataRx[0] = '\0';
}

boost::asio::async_write(socket_,
boost::asio::buffer(dataTx, max_length*2),
boost::bind(&Session::handle_write, this,
boost::asio::placeholders::error));
}
else
{
delete this;
}
}

void handle_write(const boost::system::error_code& error)
{
cout << "writing..." << endl;
if (!error)
{
cout << "dataTx sent: " << dataTx << endl;
dataTx[0] = '\0';

socket_.async_read_some(boost::asio::buffer(dataRx, max_length),
boost::bind(&Session::handle_read, this,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
else
{
delete this;
}
}

string getFactorization() const
{
//Do something
}

bool isValidData()
{
locale loc;
for (int i = 0; i < strlen(dataRx); i++)
if (!isdigit(dataRx[i],loc))
return false;

return true;
}

private:
tcp::socket socket_;
static const size_t max_length = 100;
char dataRx[max_length];
char dataTx[max_length*2];
};

class Server
{
public:
Server(boost::asio::io_service& io_service, short port)
: io_service_(io_service),
acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
{
Session* new_session = new Session(io_service_);
acceptor_.async_accept(new_session->socket(),
boost::bind(&Server::handle_accept, this, new_session,
boost::asio::placeholders::error));
}

void handle_accept(Session* new_session, const boost::system::error_code& error)
{
if (!error)
{
new_session->start();
new_session = new Session(io_service_);
acceptor_.async_accept(new_session->socket(),
boost::bind(&Server::handle_accept, this, new_session,
boost::asio::placeholders::error));
}
else
{
delete new_session;
}
}

private:
boost::asio::io_service& io_service_;
tcp::acceptor acceptor_;
};

int main(int argc, char* argv[])
{
cout << "Server is runing..." << endl;
try
{
boost::asio::io_service io_service;

int port = 13;
Server s(io_service, port);
cout << "Server is run!" << endl;
io_service.run();
}
catch (boost::system::error_code& e)
{
std::cerr << e << "\n";
}
catch (std::exception& e)
{
std::cerr << "Exception: " << e.what() << "\n";
}

return 0;
}

服务器的输出

Server is runing...
Server is run!
reading...
Data: 12
writing...
dataTx sent: 13 //just send back received ++number
reading...
Data:

非常感谢您的帮助

========

已添加

好的,我明白了。但是检查 ErrorCode == boost::asio::error::eof 不起作用...我做错了什么?

else if (ErrorCode == boost::asio::error::eof)
{
cout << "boost::asio::error::eof in OnReceive!" << endl;
}
else
{
cout << "ERROR! OnReceive..." << ErrorCode << endl;
DoClose();
}

打印出来的是ERROR! OnReceive...system:10009 好像是我比较的不对

========

已添加

我找到了根本原因。我已经声明使用 async_receive(而不是 async_read_some)并将 main 中的行交换为

ClientThread.join();
Client.Close();

现在可以正常使用了!

现在我正在尝试同时从套接字读取数据和向套接字写入数据(因为客户端应该能够在收到服务器的响应之前发送额外的请求。

OnConnect 函数中我创建了 boost 线程:

boost::thread addMsgThread(boost::bind(&TCPClient::addMsgLoop, this));
boost::thread receivingThread(boost::bind(&TCPClient::startReceiving, this));
boost::thread sendingThread(boost::bind(&TCPClient::startSending, this));

实现

void TCPClient::startReceiving()
{
cout << "receiving..." << endl;
m_RecieveBuffer[0] = '\0';
m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer, m_BufLen),
boost::bind(&TCPClient::receivingLoop, this, boost::asio::placeholders::error)); //runtime error here
cout << "m_RecieveBuffer = " << m_RecieveBuffer << endl;
}

void TCPClient::receivingLoop(const boost::system::error_code& ErrorCode)
{
cout << "receiving..." << endl;
if (ErrorCode == 0)
{
cout << "m_RecieveBuffer = " << m_RecieveBuffer << endl;

m_Socket.async_receive(boost::asio::buffer(m_RecieveBuffer, m_BufLen),
boost::bind(&TCPClient::receivingLoop, this, boost::asio::placeholders::error));
}
else
{
cout << "ERROR! receivingLoop..." << ErrorCode << endl;
DoClose();
}
}

void TCPClient::addMsgLoop()
{
while (true)
{
string tmp;
cin >> tmp;

cout << "Entered: " << tmp << endl;
tmp += "\0";

try
{
msgQueue.push(tmp);
}
catch(exception &e)
{
cerr << "Canno add msg to send queue... " << e.what() << endl;
}
}
}

receivesend 线程的问题相同:运行时错误(在 boost 库中的某处写入访问冲突)。

void TCPClient::startReceiving()
{
...
m_Socket.async_receive(); //runtime error here
}

在后续版本中一切正常(但我不知道如何在回答前实现多次发送)。谁能告诉我如何解决这个问题或如何通过其他方式实现这个问题?可能合并会有所帮助,但我现在确信这是个好方法。

最佳答案

boost::asio::ip::tcp::socket::async_read_some顾名思义就是不保证读取完整的数据。当客户端完成写入时,它将 error 对象设置为 boost::asio::error::eof

你得到的错误是因为:

服务器部分

        if (!error)
{
...
}
else
{
delete this;
}

else block 中,您假设这是一个错误情况并关闭连接。这并非总是如此。在 else 之前,您需要检查 error == boost::asio::error::eof

除此之外,在读取处理程序中,您应该继续收集缓冲区中读取的任何内容,直到您遇到 error == boost::asio::error::eof。只有这样,您才应该验证读取的数据并写回客户端。

查看 HTTP 服务器 1 , 2 , 3examples 中实现部分。

更新:更新问题的答案

更新后的代码存在线程同步问题。

  1. msgQueue 被两个或多个线程同时访问,没有任何锁。
  2. 可以同时调用同一个套接字上的读写。

如果我正确理解你的问题,你想:

  1. 获取用户输入并将其发送到服务器。
  2. 保持同时接收服务器的响应。

您可以使用两个 boost::asio::io_service::strands对于这两个任务。使用 Asio 时,链是同步任务的方式。 Asio 确保发布在链中的任务同步执行。

  1. strand1 中发布一个 send 任务,如下所示:read_user_input -> send_to_server -> handle_send -> read_user_input

  2. strand2 中发布一个 read 任务,如下所示:read_some -> handle_read -> read_some

    <

这将确保 msgQueue 不会被两个线程同时访问。使用两个套接字对服务器进行读取和写入,以确保不会在同一个套接字上同时调用读取和写入。

关于c++ - boost 异步 tcp 客户端,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/12990840/

30 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com