Boost.Asio：客户端和服务端

客户端和服务端

在这一章节，我们会深入学习怎样使用 Boost.Asio 建立非凡的客户端和服务端应用。你可以运行并测试它们，而且在理解之后，你可以把它们作为框架来构造自己的应用。本章会大量用到上一章的工具函数，如果对 read_complete 等还有印象模糊的地方，建议先回顾《Boost.Asio 基本原理》。

在接下来的例子中，客户端可以发送如下请求：

• login <username>
• ping
• ask_clients

为了更有趣一点，我们增加了一些难度：客户端发起请求的间隔是随机的，而服务端会将空闲超过 5 秒的客户端断开连接。

同步客户端/服务端

首先，我们会实现同步应用。你会发现它的代码很直接而且易读。而且因为所有的网络调用都是阻塞的，所以同步客户端不需要额外的线程（同步服务端则至少需要两个线程，稍后会看到）。

同步客户端

同步客户端会以你所期望的串行方式运行：连接到服务端，登录服务器，然后进入请求循环——休眠一下，发起一个请求，读取服务端返回，再休眠一会，一直循环下去……

因为我们是同步的，所以我们让事情变得简单一点。我们用一个 talk_to_svr 类来封装客户端的所有逻辑（片段开头的 io_service service; 是全局实例，socket 都基于它创建）：

io_service service; // 所有 socket 共用的全局 io_service

struct talk_to_svr {
    talk_to_svr(const std::string & username) : sock_(service), started_(true), username_(username) {}
    void connect(ip::tcp::endpoint ep) {
        sock_.connect(ep);
    }
    void loop() {
        write("login " + username_ + "\n");
        read_answer();
        while ( started_) {
            write_request();
            read_answer();
            boost::this_thread::sleep(millisec(rand() % 7000));
        }
    }

    std::string username() const { return username_; }
    ...
private:
    ip::tcp::socket sock_;
    enum {max_msg = 1024};
    int already_read_;
    char buff_[max_msg];
    bool started_;
    std::string username_;
};

入口函数只是创建客户端实例，先连接服务器，然后进入循环：

ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 8001);
void run_client(const std::string &client_name) {
    talk_to_svr client(client_name);
    try {
        client.connect(ep);
        client.loop();
    } catch(boost::system::system_error & err) {
        std::cout << "client terminated " << std::endl;
    }
}

int main(int argc, char* argv[]) {
    run_client("hulk");
}

在这个循环中，我们只是发起一个 ping 请求，读取服务端的返回，然后随机休眠一段时间，再重复上述过程。我们的睡眠是随机的（有时候超过 5 秒），这样服务端就有可能在某个时间点断开我们的连接：

void write_request() {
    write("ping\n");
}
void read_answer() {
    already_read_ = 0;
    read(sock_, buffer(buff_), boost::bind(&talk_to_svr::read_complete, this, _1, _2));
    process_msg();
}
void process_msg() {
    std::string msg(buff_, already_read_);
    if (msg.find("login ") == 0) on_login();
    else if (msg.find("ping") == 0) on_ping(msg);
    else if (msg.find("clients ") == 0) on_clients(msg);
    else std::cerr << "invalid msg " << msg << std::endl;
}

对于读取结果，我们使用之前章节就说到的 read_complete() 来保证每次都恰好读到一条完整的消息（读到换行符 '\n' 为止），随后由 process_msg() 把服务端的返回分发到对应的方法去处理：

void on_login() {do_ask_clients();}
void on_ping(const std::string & msg) {
    std::istringstream in(msg);
    std::string answer;
    in >> answer >> answer;
    if (answer == "client_list_changed")
        do_ask_clients();
}
void on_clients(const std::string & msg) {
    std::string clients = msg.substr(8);
    std::cout << username_ << ", new client list:" << clients;
}
void do_ask_clients() {
    write("ask_clients\n");
    read_answer();
}
void write(const std::string & msg) { sock_.write_some(buffer(msg)); }
size_t read_complete(const boost::system::error_code & err, size_t bytes) {
    // ... 与之前章节相同：以 '\n' 判断一条消息是否读完，并把已读字节数存入 already_read_
}

在读取服务端对我们 ping 操作的返回时，如果得到的消息是 client_list_changed，我们就需要重新请求客户端列表。

同步服务端

同步服务端也是相当简单的，不过正常来说服务端都比客户端要难实现：它要管理所有已经连接的客户端。因为 accept() 是一个阻塞操作，我们不能只用一个线程——至少需要两个线程，一个负责接受新的客户端连接，另外一个负责处理已经存在的客户端请求。

为了分辨客户端发送过来的请求，我们需要保存一个客户端的列表。每个 talk_to_client 实例都拥有一个 socket，而 socket 类是不支持拷贝构造的，所以如果你想把它们保存在一个 std::vector 对象中，就需要借助智能指针。这里有两种实现方式：在 talk_to_client 内部保存一个指向 socket 的智能指针，然后创建 talk_to_client 实例的数组；或者让 talk_to_client 实例用成员变量的方式持有 socket，然后创建一个指向 talk_to_client 的智能指针数组。我选择后者，但你也可以选择前一种方式：

typedef boost::shared_ptr<talk_to_client> client_ptr;
typedef std::vector<client_ptr> array;
array clients;
boost::recursive_mutex cs; // 用线程安全的方式访问客户端数组

有了客户端列表和全局互斥量，服务端的两个线程和入口函数如下：

void accept_thread() {
    ip::tcp::acceptor acceptor(service,ip::tcp::endpoint(ip::tcp::v4(), 8001));
    while (true) {
        client_ptr new_( new talk_to_client);
        acceptor.accept(new_->sock());
        boost::recursive_mutex::scoped_lock lk(cs);
        clients.push_back(new_);
    }
}

void handle_clients_thread() {
    while (true) {
        boost::this_thread::sleep( millisec(1));
        boost::recursive_mutex::scoped_lock lk(cs);
        for(array::iterator b = clients.begin(), e = clients.end(); b!= e; ++b)
            (*b)->answer_to_client();
        // 删除已经超时的客户端
        clients.erase(std::remove_if(clients.begin(), clients.end(), boost::bind(&talk_to_client::timed_out,_1)), clients.end());
    }
}
int main(int argc, char* argv[]) {
    boost::thread_group threads;
    threads.create_thread(accept_thread);
    threads.create_thread(handle_clients_thread);
    threads.join_all();
}

talk_to_client 的主要代码如下：

struct talk_to_client : boost::enable_shared_from_this<talk_to_client>
{
    talk_to_client() { ... }
    std::string username() const { return username_; }
    void answer_to_client() {
        try {
            read_request();
            process_request();
        } catch ( boost::system::system_error&) { stop(); }
        if ( timed_out())
            stop();
    }
    void set_clients_changed() { clients_changed_ = true; }
    ip::tcp::socket& sock() { return sock_; }
    bool timed_out() const {
        ptime now = microsec_clock::local_time();
        long long ms = (now - last_ping).total_milliseconds();
        return ms > 5000 ;
    }
    void stop() {
        boost::system::error_code err; sock_.close(err);
    }
    void read_request() {
        if ( sock_.available())
            already_read_ += sock_.read_some(buffer(buff_ + already_read_, max_msg - already_read_));
    }
    ...
private:
    // max_msg、already_read_、buff_ 等成员和同步客户端中的一样
    bool clients_changed_;
    ptime last_ping;
};

上述代码拥有非常好的自释能力。其中最重要的是 read_request() 和 process_request()：read_request() 只在 socket 中已有有效数据时才读取，这样服务端永远不会阻塞；而消息的解析和分发则在 process_request() 中完成：

void process_request() {
    bool found_enter = std::find(buff_, buff_ + already_read_, '\n') < buff_ + already_read_;
    if ( !found_enter)
        return; // 消息不完整，等下一次读取
    // 处理消息
    last_ping = microsec_clock::local_time();
    size_t pos = std::find(buff_, buff_ + already_read_, '\n') - buff_;
    std::string msg(buff_, pos);
    std::copy(buff_ + already_read_, buff_ + max_msg, buff_);
    already_read_ -= pos + 1;
    if ( msg.find("login ") == 0) on_login(msg);
    else if ( msg.find("ping") == 0) on_ping();
    else if ( msg.find("ask_clients") == 0) on_clients();
    else std::cerr << "invalid msg " << msg << std::endl;
}
void on_login(const std::string & msg) {
    std::istringstream in(msg);
    in >> username_ >> username_;
    write("login ok\n");
    update_clients_changed();
}
void on_ping() {
    write(clients_changed_ ? "ping client_list_changed\n" : "ping ok\n");
    clients_changed_ = false;
}
void on_clients() {
    std::string msg;
    {
        boost::recursive_mutex::scoped_lock lk(cs);
        for(array::const_iterator b = clients.begin(), e = clients.end() ; b != e; ++b)
            msg += (*b)->username() + " ";
    }
    write("clients " + msg + "\n");
}
void write(const std::string & msg) { sock_.write_some(buffer(msg)); }

观察 process_request()。当缓冲区里有数据时，我们先通过 found_enter 判断是否已经读到一条完整的消息（以 '\n' 结尾）；如果没有，就先返回，等下一次继续读取。如果读到了，就把 '\n' 之前的这条消息解析出来，并把 '\n' 之后剩余的字节移到缓冲区头部、调整 already_read_，这样即使一次读到多条消息也不会丢失内容。剩下的代码都很容易读懂。

异步客户端/服务端

现在，是比较有趣（也比较难）的异步实现！查看示意图时，你需要知道图中的 Boost.Asio 代表一个由 Boost.Asio 代为执行的异步调用：例如 do_read() 由 Boost.Asio 代理执行，do_read() 与 on_read() 之间的箭头代表从 do_read() 到 on_read() 的逻辑流程，但你永远不知道 on_read() 什么时候会被调用，只知道它最终一定会被调用。（同步与异步的完整对比可以看本站的《Boost.Asio：同步 vs 异步》）

异步客户端

到这里事情会变得有点复杂，但是仍然是可控的。当然你也会拥有一个不会阻塞的应用。

下面的代码你应该已经很熟悉：

#define MEM_FN(x)       boost::bind(&self_type::x, shared_from_this())
#define MEM_FN1(x,y)    boost::bind(&self_type::x, shared_from_this(),y)
#define MEM_FN2(x,y,z)  boost::bind(&self_type::x, shared_from_this(),y,z)

io_service service;

class talk_to_svr : public boost::enable_shared_from_this<talk_to_svr>, boost::noncopyable {
    typedef talk_to_svr self_type;
    talk_to_svr(const std::string & username) : sock_(service), started_(true), username_(username), timer_(service) {}
    void start(ip::tcp::endpoint ep) {
        sock_.async_connect(ep, MEM_FN1(on_connect,_1));
    }
public:
    typedef boost::system::error_code error_code;
    typedef boost::shared_ptr<talk_to_svr> ptr;
    static ptr start(ip::tcp::endpoint ep, const std::string & username) {
        ptr new_(new talk_to_svr(username));
        new_->start(ep);
        return new_;
    }
    void stop() {
        if ( !started_) return;
        started_ = false;
        sock_.close();
    }
    bool started() { return started_; }
    ...
private:
    size_t read_complete(const boost::system::error_code &err, size_t bytes) {
        if (err) return 0;
        bool found = std::find(read_buffer_, read_buffer_ + bytes, '\n') < read_buffer_ + bytes;
        return found ? 0 : 1;
    }
private:
    ip::tcp::socket sock_;
    enum { max_msg = 1024 };
    char read_buffer_[max_msg];
    char write_buffer_[max_msg];
    bool started_;
    std::string username_;
    deadline_timer timer_;
};

你会看到额外多了一个 deadline_timer 成员，用来安排 ping 服务端；而且 ping 的间隔同样是随机的。下面是这个类的核心逻辑：

void on_connect(const error_code &err) {
    if ( !err)      do_write("login " + username_ + "\n");
    else            stop();
}
void on_read(const error_code & err, size_t bytes) {
    if (err) stop();
    if (!started()) return;
    // 处理消息
    std::string msg(read_buffer_, bytes);
    if (msg.find("login ") == 0) on_login();
    else if (msg.find("ping") == 0) on_ping(msg);
    else if (msg.find("clients ") == 0) on_clients(msg);
}
void on_login() {
    do_ask_clients();
}
void on_ping(const std::string & msg) {
    std::istringstream in(msg);
    std::string answer;
    in >> answer >> answer;
    if ( answer == "client_list_changed") do_ask_clients();
    else postpone_ping();
}
void on_clients(const std::string & msg) {
    std::string clients = msg.substr(8);
    std::cout << username_ << ", new client list:" << clients ;
    postpone_ping();
}

在on_read()中，首先的两行代码是亮点。在第一行，如果出现错误，我们就停止。而第二行，如果我们已经停止了（之前就停止了或者刚好停止），我们就返回。反之如果所有都是OK，我们就对收到的消息进行处理。

最后是do_*方法，实现如下：

void do_ping() {do_write("ping\n");}
void postpone_ping() {
    timer_.expires_from_now(boost::posix_time::millisec(rand() % 7000));
    timer_.async_wait( MEM_FN(do_ping));
}
void do_ask_clients() {do_write("ask_clients\n");}
void on_write(const error_code & err, size_t bytes) { do_read(); }
void do_read() {
    async_read(sock_, buffer(read_buffer_), MEM_FN2(read_complete,_1,_2), MEM_FN2(on_read,_1,_2));
}
void do_write(const std::string & msg) {
    if (!started()) return;
    std::copy(msg.begin(), msg.end(), write_buffer_);
    sock_.async_write_some(buffer(write_buffer_, msg.size()), MEM_FN2(on_write,_1,_2));
}

注意，每读完一条消息（on_read 处理完之后），我们都会通过 postpone_ping() 安排一次新的 ping。

异步服务端

这个示意图相当复杂：从 Boost.Asio 出发，你可以看到 4 条箭头分别指向 on_accept、on_read、on_write 和 on_check_ping。这意味着你永远不知道下一个完成的异步调用是哪一个，但可以确定的是，它一定是这 4 个操作中的一个。

现在，我们是异步的了；我们可以继续保持单线程。接受客户端连接是最简单的部分，如下所示：

io_service service;
ip::tcp::acceptor acceptor(service, ip::tcp::endpoint(ip::tcp::v4(), 8001));
void handle_accept(talk_to_client::ptr client, const boost::system::error_code & err)
{
    client->start();
    talk_to_client::ptr new_client = talk_to_client::new_();
    acceptor.async_accept(new_client->sock(), boost::bind(handle_accept, new_client, _1));
}
int main(int argc, char* argv[]) {
    talk_to_client::ptr client = talk_to_client::new_();
    acceptor.async_accept(client->sock(),boost::bind(handle_accept,client,_1));
    service.run();
}

上述代码会一直异步地等待新的客户端连接（每个新的客户端连接又会触发另外一个异步等待操作）。我们需要监控 client list changed 事件——有新客户端连接或者有客户端断开连接——并在事件发生时通知所有客户端。因此，我们需要保存一个客户端连接的数组；除非你从头到尾都不需要知道当前有哪些客户端在线，否则这样一个数组是必不可少的。

class talk_to_client;
typedef boost::shared_ptr<talk_to_client> client_ptr;
typedef std::vector<client_ptr> array;
array clients;

talk_to_client 类的框架如下：

class talk_to_client : public boost::enable_shared_from_this<talk_to_client> , boost::noncopyable {
    talk_to_client() { ... }
public:
    typedef boost::system::error_code error_code;
    typedef boost::shared_ptr<talk_to_client> ptr;
    void start() {
        started_ = true;
        clients.push_back( shared_from_this());
        last_ping = boost::posix_time::microsec_clock::local_time();
        do_read(); // 首先，我们异步等待客户端发来的第一条消息
    }
    static ptr new_() { ptr new_(new talk_to_client); return new_; }
    void stop() {
        if (!started_) return;
        started_ = false;
        sock_.close();
        ptr self = shared_from_this();
        array::iterator it = std::find(clients.begin(), clients.end(), self);
        clients.erase(it);
        update_clients_changed();
    }
    bool started() const { return started_; }
    ip::tcp::socket & sock() { return sock_;}
    std::string username() const { return username_; }
    void set_clients_changed() { clients_changed_ = true; }
    ...
private:
    ip::tcp::socket sock_;
    enum {max_msg = 1024};
    char read_buffer_[max_msg];
    char write_buffer_[max_msg];
    bool started_;
    std::string username_;
    deadline_timer timer_;
    boost::posix_time::ptime last_ping;
    bool clients_changed_;
};

我会根据它运行在哪一端，把这个连接类叫作 talk_to_client 或者 talk_to_server，这样更容易说清楚它在做什么。

其余的代码（MEM_FN 宏、read_complete 等）和异步客户端中的一样。不同的是 stop()：它除了关闭 socket，还会把自己从客户端数组中移除，并通过 update_clients_changed() 通知其他客户端。

服务端持续不断地等待异步的read操作：

void on_read(const error_code & err, size_t bytes) {
    if (err) stop();
    if (!started()) return;
    std::string msg(read_buffer_, bytes);
    if (msg.find("login ") == 0) on_login(msg);
    else if (msg.find("ping") == 0) on_ping();
    else if (msg.find("ask_clients") == 0) on_clients();
}
void on_login(const std::string & msg) {
    std::istringstream in(msg);
    in >> username_ >> username_;
    do_write("login ok\n");
    update_clients_changed();
}
void on_ping() {
    do_write(clients_changed_ ? "ping client_list_changed\n" : "ping ok\n");
    clients_changed_ = false;
}
void on_clients() {
    std::string msg;
    for(array::const_iterator b =clients.begin(),e =clients.end(); b != e; ++b)
        msg += (*b)->username() + " ";
    do_write("clients " + msg + "\n");
}

这段代码简单易懂；需要注意的一点是：当一个新客户端登录时，我们调用 update_clients_changed()，这个方法把所有客户端的 clients_changed_ 标志置为 true。

服务端每收到一个请求就用相应的方式进行回复，如下所示：

void do_ping() {do_write("ping\n"); }
void do_ask_clients() {do_write("ask_clients\n"); }
void on_write(const error_code & err, size_t bytes) { do_read(); }
void do_read() {
    async_read(sock_, buffer(read_buffer_), MEM_FN2(read_complete,_1,_2), MEM_FN2(on_read,_1,_2));
    post_check_ping();
}
void do_write(const std::string & msg) {
    if (!started()) return;
    std::copy(msg.begin(), msg.end(), write_buffer_);
    sock_.async_write_some( buffer(write_buffer_, msg.size()), MEM_FN2(on_write,_1,_2));
}
size_t read_complete(const boost::system::error_code & err, size_t bytes) {
    // ... 就像之前
}

在每个write操作的末尾，on_write()方法被调用，这个方法会触发另外一个异步读操作，这样的话”等待请求－回复请求”这个循环就会一直执行，直到客户端断开连接或者超时。

在每次读操作开始时，我们还会异步地等待 5 秒钟，观察客户端是否超时。如果超时，我们就关闭它的连接：

void on_check_ping() {
    ptime now = microsec_clock::local_time();
    if ((now - last_ping).total_milliseconds() > 5000)
        stop();
    last_ping = boost::posix_time::microsec_clock::local_time();
}
void post_check_ping() {
    timer_.expires_from_now(boost::posix_time::millisec(5000));
    timer_.async_wait( MEM_FN(on_check_ping));
}

这就是整个服务端的实现。你可以运行并让它工作起来！

在代码中，我向你们展示了这一章学到的东西。为了更容易理解，我把代码稍微精简了一些——比如，大部分的控制台输出我都省略了。我建议你亲手把例子补全并运行一遍，因为从头到尾读一次代码，能大大加深你对本章展示应用的理解。

总结

我们已经学到了怎么写一些基础的客户端/服务端应用。我们已经避免了一些诸如内存泄漏和死锁的低级错误。所有的编码都是框架式的，这样你就可以根据你自己的需求对它们进行扩展。

在接下来的章节中，我们会更加深入地了解使用 Boost.Asio 进行同步编程和异步编程的不同点，同时你也会学会如何嵌入你自己的异步操作，请看本站的《Boost.Asio：同步 vs 异步》。