异步编程
本节深入讨论异步编程将遇到的若干问题。建议多次阅读,以便吃透这一节的内容,这一节是对整个boost.asio来说是非常重要的。
为什么需要异步
如前所述,通常同步编程要比异步编程更简单。同步编程下,我们很容易线性地对问题进行考量,函数A调用完,继续执行B,B执行完,继续执行C,以此类推,相对比较直观。而对于异步编程,假设有5个事件,我们很难知道它们具体的执行顺序,你甚至不知道,它到底会不会被执行。
虽然编写异步的程序,很难,但是依然需要使用这种方法。因为服务器程序需要同时并行的处理大量的客户端。并行的客户端越多,异步编程的优势就越明显。
假设有一个服务器程序,需要同时处理1000个并行的客户端,客户端和服务器之间的通信消息,以’\n’来结尾。
这是同步模式下的代码,使用1个线程:
using namespace boost::asio;
struct client{
ip::tcp::socket sock;
char buff[1024]; //每个消息最大1024个字节
int already_read; //已经读取了多少字节
};
std::vector<client> clients;
void handle_clients() {
while(true) {
for(int i=0; i<clients.size(); ++i) {
if(clients[i].sock.available() ) on_read(clients[i]));
}
}
}
void on_read(client& c) {
int to_read = std::min(1024 - c.already_read, c.sock.available());
c.sock.read_some(buffer(c.buff + c.already_read, to_read);
c.already_read += to_read;
if(std::find(c.buff, c.buff + c.already_read, '\n') < c.buff + c.already_read) {
int pos = std::find(c.buff, c.buff + c.alread_read, '\n') - c.buff;
std::string msg(c.buff, c.buff + pos);
std::copy(c.buff + pos, c.buff + 1024, c.buff);
c.already_read -= pos;
on_read_msg(c, msg);
}
}
void on_read_msg(client & c, const std::string& msg) {
if(msg == "request_login")
c.sock.write("request_ok\n");
else if ...
}
在任务服务器程序中,你都需要避免代码被阻塞。看上面的代码,我们希望handle_clients()这个函数尽可能的不被阻塞。任何时候函数阻塞,从客户端发来的消息就会等待,直到函数不阻塞的时候才能来处理它们。为了避免程序被阻塞,我们只在socket中确实有数据的时候才去读取,实现的代码是:
if(clients[i].sock.available() ) on_read(clients[i]));
在on_read函数中,我们只读取socket确实有效的数据;调用read_util(c.sock, buffer(…),’\n’);是非常不好的选择,它会阻塞程序,直到完整的读取了全部的数据才返回。我们永远不知道到底什么时候才能读取完毕。
程序的瓶颈是on_read_msg()这个函数;所有输入的数据都被卡在这里。on_read_msg()应该尽量避免出现这种情况,但有时这又是完全没办法避免的。(比如,当socket的缓冲区满了以后,操作必将被阻塞)。
下面是同步模式下的代码,使用10个线程:
using namespace boost::asio;
//...这里的定义和前面一样
bool set_reading() {
boost::mutex::scope_lock lk(cs_);
if(is_reading_) return false;
else { if_reading_ = true; return true; }
}
void unset_reading() {
boost::mutex::scoped_lock lk(cs_);
is_reading_ = false;
}
private:
boost::mutex cs_;
bool is_reading_;
};
std::vector<client> clients;
void handle_clients() {
for(int i=0; i<10; ++i) {
boost::thread(handle_clients_thread);
}
}
void handle_clients_thread() {
while(true) {
for(int i=0; i<clients.size(); ++i) {
if(clients[i].sock.available()) {
if(clients[i].set_reading()) {
on_read(clients[i]);
clients[i].unset_reading();
}
}
}
}
}
void on_read(client & c) {
...
}
void on_read_msg(client & c, const std::string& msg) {
...
}
为了使用多线程,我们需要同步机制,set_reading()和unset_reading()就是在做这个用的。set_reading(),非常重要,它测试当前是否有线程在做读取操作。
可以看出来代码已经比较复杂了。
还有一种同步编程的方法,就是每个client分配一个线程。但是随着并行的客户端数量的增长,这显然是不可行的。
现在我们来看看异步方法。当client请求数据的时候,on_read被调用,发送返回数据,之后又等待下一个请求(执行另一个异步的读操作)。
异步代码,10个线程:
using namespace boost::asio;
io_service service;
struct client {
ip::tcp::socket sock;
streambuf buff;
};
std::vector<client> clients;
void handle_clients() {
for(int i=0; i<clients.size(); ++i) {
async_read_util(clients[i].sock, clients[i].buff, '\n',
boost::bind(on_read, clients[i], _1, _2));
}
for(int i=0; i<10; ++i)
boost::thread(handle_clients_thread);
}
void handle_clients_thread() {
service_run();
}
void on_read(client& c, const error_code& err, size_t read_bytes) {
std::istream in(&c.buff);
std::string msg;
std::getline(in, msg);
if(msg == "request_login")
c.sock.async_write("request_ok\n", on_write);
else if ...
...
//现在在同一个socket上等待下一个读取请求
async_read_util(c.sock, c.buff, '\n',
boost::bind(on_read, c, _1, _2));
}