Boost.Asio 基本原理

Boost.Asio 基本原理

这一章涵盖了使用 Boost.Asio 时必须知道的一些事情。我们也将深入研究比同步编程更复杂、更有乐趣的异步编程。

网络 API

这一部分包含了当使用 Boost.Asio 编写网络应用程序时必须知道的事情。

Boost.Asio 命名空间

Boost.Asio 的所有内容都包含在 boost::asio 命名空间或者其子命名空间内。

IP地址

对于 IP 地址的处理，Boost.Asio 提供了 ip::address、ip::address_v4 和 ip::address_v6 类。它们提供了相当多的函数。大多数情况你会选择用 ip::address::from_string：

ip::address addr = ip::address::from_string("127.0.0.1");

如果你想通过一个主机名进行连接，下面的代码片段是无用的：

// 抛出异常
ip::address addr = ip::address::from_string("www.yahoo.com");

端点

端点是一个地址加上端口，它指明了你要连接到哪里。不同类型的 socket 有它自己的 endpoint 类，比如 ip::tcp::endpoint、ip::udp::endpoint 和 ip::icmp::endpoint。

如果想连接到本机的80端口，你可以这样做：

ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 80);

有三种方式来让你建立一个端点。例子如下：

ip::tcp::endpoint ep1;
ip::tcp::endpoint ep2(ip::tcp::v4(), 80);
ip::tcp::endpoint ep3(ip::address::from_string("127.0.0.1"), 80);

如果你想连接到一个主机（不是IP地址），你需要这样做：

// 输出 "87.248.122.122"
io_service service;
ip::tcp::resolver resolver(service);
ip::tcp::resolver::query query("www.yahoo.com", "80");
ip::tcp::resolver::iterator iter = resolver.resolve(query);
ip::tcp::endpoint ep = *iter;
std::cout << ep.address().to_string() << std::endl;

你可以用你需要的 socket 类型来替换 tcp。首先，为你想要查询的名字创建一个查询器，然后用 resolve() 函数解析它。如果成功，它至少会返回一个条目。你可以利用返回的迭代器，使用第一个条目或者遍历整个列表来拿到全部的条目。

给定一个端点，可以获得它的地址、端口和 IP 协议（v4 或者 v6）：

std::cout << ep.address().to_string() << ":" << ep.port()
<< "/" << ep.protocol().family() << std::endl;

套接字

Boost.Asio 有三种类型的套接字类：ip::tcp、ip::udp 和 ip::icmp。当然它也是可扩展的，你可以创建自己的 socket 类，尽管这相当复杂。如果你选择这样做，参照一下 boost/asio/ip/tcp.hpp、boost/asio/ip/udp.hpp 和 boost/asio/ip/icmp.hpp。它们都是含有内部 typedef 关键字的超小类。

你可以把 ip::tcp、ip::udp、ip::icmp 类当作占位符；它们可以让你便捷地访问其他类/函数。例如，socket 类创建一个相应的 socket，并且总是在构造的时候传入 io_service 实例：

io_service service;
ip::udp::socket sock(service);
sock.set_option(ip::udp::socket::reuse_address(true));

每一个 socket 的名字都是一个 typedef 关键字，比如 ip::tcp::socket 实际上就是 basic_stream_socketip::tcp 的别名。

同步错误码

所有的同步函数都有抛出异常或者返回错误码的重载，比如下面的代码片段：

sync_func(arg1, arg2, ..., argN); // 抛出异常
boost::system::error_code ec;
sync_func(arg1, arg2, ..., argN, ec); // 返回错误码

在这一章剩下的部分，你会见到大量的同步函数。简单起见，我省略了有返回错误码的重载，但是不可否认它们确实是存在的。

socket成员方法

这些方法被分成了几组。并不是所有的方法都可以在各个类型的套接字里使用。这个部分的结尾将有一个列表来展示各个方法分别属于哪个 socket 类。

注意所有的异步方法都立刻返回，而它们相对的同步实现需要操作完成之后才能返回。

连接相关的函数

这些方法是用来连接或绑定 socket、断开 socket 连接以及查询连接是活动还是非活动的。例子如下：

io_service service;
ip::tcp::endpoint ep( ip::address::from_string("127.0.0.1"), 80);
ip::tcp::socket sock(service);
sock.open(ip::tcp::v4());
sock.connect(ep);
sock.write_some(buffer("GET /index.html\r\n"));
char buff[1024];
sock.read_some(buffer(buff, 1024));
sock.shutdown(ip::tcp::socket::shutdown_receive);
sock.close();

读写函数

这些是在套接字上执行 I/O 操作的函数。

对于异步函数来说，处理程序的格式 void handler(const boost::system::error_code& e, size_t bytes) 都是一样的。

稍后我们将讨论缓冲区。让我们先来了解一下标记（flag）。标记的默认值是 0，但是也可以是以下几种：

• message_peek：只查看数据，不把它从接收队列中移除
• message_out_of_band：处理带外（OOB）数据
• message_do_not_route：不进行路由，直接使用本地接口

你最常用的可能是 message_peek，使用方法请参照下面的代码片段：

char buff[1024];
sock.receive(buffer(buff), ip::tcp::socket::message_peek);
memset(buff, 0, 1024);
// 重新读取之前已经读取过的内容
sock.receive(buffer(buff) );

在同步或异步地从不同类型的套接字上读取数据时，有几点需要注意。

注意：使用 receive_from 从一个 UDP 套接字读取数据时，你需要构造一个默认的端点，它会在读取完成后被填充为发送方的端点。

ip::udp::endpoint sender_ep;
char buff[1024];
sock.receive_from(buffer(buff), sender_ep);

套接字选项

这些函数用来处理套接字的高级选项。下面是你可以获取/设置的套接字选项：

名字	描述	类型
broadcast	如果为 true，允许广播消息	bool
debug	如果为 true，启用套接字级别的调试	bool
do_not_route	如果为 true，则阻止路由选择只使用本地接口	bool
enable_connection_aborted	如果为 true，记录在 accept() 时中断的连接	bool
keep_alive	如果为 true，会发送心跳	bool
linger	如果为 true，套接字会在有未发送数据的情况下挂起 close()	bool
receive_buffer_size	套接字接收缓冲区大小	int
receive_low_watermark	规定套接字输入处理的最小字节数	int
reuse_address	如果为 true，套接字能绑定到一个已用的地址	bool
send_buffer_size	套接字发送缓冲区大小	int
send_low_watermark	规定套接字数据发送的最小字节数	int
ip::v6_only	如果为 true，则只允许 IPv6 的连接	bool

每个名字代表了一个内部套接字 typedef 或者类。下面是对它们的使用：

io_service service;
ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 80);
ip::tcp::socket sock(service);
sock.connect(ep);
// TCP套接字可以重用地址
ip::tcp::socket::reuse_address ra(true);
sock.set_option(ra);
// 获取套接字读取的数据
ip::tcp::socket::receive_buffer_size rbs;
sock.get_option(rbs);
std::cout << rbs.value() << std::endl;
// 把套接字的缓冲区大小设置为8192
ip::tcp::socket::send_buffer_size sbs(8192);
sock.set_option(sbs);

注意：要让上述选项生效，套接字必须已经被打开；否则，会抛出异常。

TCP VS UDP VS ICMP

就像我之前所说，不是所有的成员方法在所有的套接字类中都可用。我做了一个包含成员函数不同点的列表。如果一个成员函数没有出现在这里，说明它在所有的套接字类中都是可用的。

名字	TCP	UDP	ICMP
async_read_some	是	-	-
async_receive_from	-	是	是
async_write_some	是	-	-
async_send_to	-	是	是
read_some	是	-	-
receive_from	-	是	是
write_some	是	-	-
send_to	-	是	是

其他需要考虑的事情

最后要注意的一点，套接字实例不能被拷贝，因为拷贝构造方法和 = 操作符是不可访问的。

io_service service;
ip::tcp::socket s1(service), s2(service);
s1 = s2; // 编译时报错
ip::tcp::socket s3(s1); // 编译时报错

这是非常有意义的，因为每一个实例都拥有并管理着一个资源（原生套接字本身）。如果我们允许拷贝构造，结果是我们会有两个实例拥有同样的原生套接字；这样我们就需要去处理所有者的问题（让一个实例拥有所有权？或者使用引用计数？还是其他的方法）。Boost.Asio 选择不允许拷贝（如果你想要创建一个备份，请使用共享指针）：

io_service service;
typedef boost::shared_ptr<ip::tcp::socket> socket_ptr;
socket_ptr sock1(new ip::tcp::socket(service));
socket_ptr sock2(sock1); // ok
socket_ptr sock3;
sock3 = sock1; // ok

套接字缓冲区

当从一个套接字读写内容时，你需要一个缓冲区，用来保存读取和写入的数据。缓冲区内存的有效时间必须比 I/O 操作的时间要长；你需要保证它们在 I/O 操作结束之前不被释放。对于同步操作来说，这很容易保证，因为缓冲区在 receive 和 send 调用的整个过程中都一直存在。

char buff[512];
...
sock.receive(buffer(buff));
strcpy(buff, "ok\n");
sock.send(buffer(buff));

但是在异步操作时就没这么简单了，看下面的代码片段：

// 非常差劲的代码 ...
void on_read(const boost::system::error_code &err, std::size_t read_bytes)
{ ... }
void func() {
    char buff[512];
    sock.async_receive(buffer(buff), on_read);
}

在我们调用async_receive()之后，buff就已经超出有效范围，它的内存当然会被释放。当我们开始从套接字接收一些数据时，我们会把它们拷贝到一片已经不属于我们的内存中；它可能会被释放，或者被其他代码重新开辟来存入其他的数据，结果就是：段错误。

对于上面的问题有几个解决方案：

• 把缓冲区做成全局变量
• 让完成处理程序管理缓冲区的生命周期（自行释放）
• 使用连接对象来管理套接字和其他数据

第一个方法显然不是很好，因为我们都知道全局变量非常不好。此外，如果两个实例使用同一个缓冲区怎么办？

下面是第二种方式的实现：

void on_read(char * ptr, const boost::system::error_code &err, std::size_t read_bytes) {
    delete[] ptr;
}
...
char * buff = new char[512];
sock.async_receive(buffer(buff, 512), boost::bind(on_read, buff, _1, _2));

或者，如果你想要缓冲区在操作结束后自动超出范围，使用共享指针：

struct shared_buffer {
    boost::shared_array<char> buff;
    size_t size;
    shared_buffer(size_t sz) : buff(new char[sz]), size(sz) {
    }
    mutable_buffers_1 asio_buff() const {
        return buffer(buff.get(), size);
    }
};
...
// 当 on_read 的仿函数拷贝超出范围时，boost::bind 对象被释放了，
// 同时也会释放共享指针
void on_read(shared_buffer, const boost::system::error_code &err, std::size_t read_bytes) {}
shared_buffer sb(512);
sock.async_receive(sb.asio_buff(), boost::bind(on_read, sb, _1, _2));

shared_buffer 类内部持有一个 boost::shared_array；shared_array<> 的作用是让 shared_buffer 的每一份拷贝都共享同一块堆内存——当最后一个持有它的实例超出作用域时，这块内存就会被自动销毁，而这就是我们想要的结果。

因为 Boost.Asio 会给完成处理程序保留一个拷贝，当操作完成时就会调用这个完成处理程序，所以你的目的达到了。那个拷贝是一个 boost::bind 的仿函数，它拥有着实际的 shared_buffer 实例。这是非常优雅的！

第三个选择是使用一个连接对象来管理套接字和其他数据，比如缓冲区，通常来说这是正确的解决方案但是非常复杂。在这一章的末尾我们会对这种方法进行讨论。

缓冲区封装函数

纵观所有代码，你会发现：无论什么时候，当我们需要对一个 buffer 进行读写操作时，代码会把实际的缓冲区对象封装在一个 buffer() 方法中，然后再把它传递给方法调用：

char buff[512];
sock.async_receive(buffer(buff), on_read);

基本上我们都会把缓冲区包含在一个类中以便 Boost.Asio 的方法能遍历这个缓冲区，比方说，使用下面的代码：

sock.async_receive(some_buffer, on_read);

实例 some_buffer 需要满足一些需求，叫做 ConstBufferSequence 或者 MutableBufferSequence（你可以在 Boost.Asio 的文档中查看它们）。创建你自己的类去满足这些需求的细节是非常复杂的，但是 Boost.Asio 已经提供了一些类用来满足这些需求。所以你不用直接访问这些缓冲区，而可以使用 buffer() 方法。

可以放心地说，你可以把下面列出来的类型都包装到一个 buffer() 方法中。下面的代码都是有效的：

struct pod_sample { int i; long l; char c; };
...
char b1[512];
void * b2 = new char[512];
std::string b3; b3.resize(128);
pod_sample b4[16];
std::vector<pod_sample> b5; b5.resize(16);
boost::array<pod_sample,16> b6;
std::array<pod_sample,16> b7;
sock.async_send(buffer(b1), on_read);
sock.async_send(buffer(b2,512), on_read);
sock.async_send(buffer(b3), on_read);
sock.async_send(buffer(b4), on_read);
sock.async_send(buffer(b5), on_read);
sock.async_send(buffer(b6), on_read);
sock.async_send(buffer(b7), on_read);

总的来说就是：与其创建你自己的类来处理 ConstBufferSequence 或者 MutableBufferSequence 的需求，不如创建一个能在你需要的时候保留缓冲区、然后返回一个 mutable_buffers_1 实例的类，而我们早在 shared_buffer 类中就这样做了。

read/write/connect 自由函数

Boost.Asio 提供了处理 I/O 的自由函数，我们分四组来分析它们。

connect方法

这些方法把套接字连接到一个端点。它的例子如下：

using namespace boost::asio::ip;
tcp::resolver resolver(service);
tcp::resolver::iterator iter = resolver.resolve(tcp::resolver::query("www.yahoo.com","80"));
tcp::socket sock(service);
connect(sock, iter);

一个主机名可以被解析成多个地址，而 connect 和 async_connect 能很好地把你从尝试每个地址然后找到一个可用地址的繁重工作中解放出来，因为它们已经帮你做了这些。

read/write方法

这些方法对一个流进行读写操作（可以是套接字，或者其他表现得像流的类）。

首先，要注意第一个参数变成了流，而不单是 socket。这个参数包含了 socket 但不仅仅是 socket。比如，你可以用一个 Windows 的文件句柄来替代 socket。当下面情况出现时，所有 read 和 write 操作都会结束：

• 完成条件返回 0
• 发生了错误
• 缓冲区已满（读）或已全部写完（写）

下面的代码会异步地从一个socket中读取数据直到读取到’\n’：

io_service service;
ip::tcp::socket sock(service);
char buff[512];
int offset = 0;
size_t up_to_enter(const boost::system::error_code &, size_t bytes) {
    for (size_t i = 0; i < bytes; ++i)
        if (buff[i + offset] == '\n')
            return 0;
    return 1;
}
void on_read(const boost::system::error_code &, size_t) {}
...
async_read(sock, buffer(buff), up_to_enter, on_read);

Boost.Asio 也提供了一些简单的完成处理仿函数：

• transfer_all()：传输所有数据，直到缓冲区填满或写完
• transfer_at_least(n)：至少传输 n 字节
• transfer_exactly(n)：恰好传输 n 字节

例子如下：

char buff[512];
void on_read(const boost::system::error_code &, size_t) {}
// 读取32个字节
async_read(sock, buffer(buff), transfer_exactly(32), on_read);

上述的 4 个方法，不使用普通的缓冲区，而使用 Boost.Asio 的 streambuf 类（由 std::streambuf 继承而来）。STL 流和流缓冲区非常复杂；下面是例子：

io_service service;
void on_read(streambuf& buf, const boost::system::error_code &, size_t) {
    std::istream in(&buf);
    std::string line;
    std::getline(in, line);
    std::cout << "first line: " << line << std::endl;
}
int main(int argc, char* argv[]) {
    HANDLE file = ::CreateFile("readme.txt", GENERIC_READ, 0, 0, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, 0);
    windows::stream_handle h(service, file);
    streambuf buf;
    async_read(h, buf, transfer_exactly(256), boost::bind(on_read,boost::ref(buf), _1, _2));
    service.run();
}

在这里，我向你们展示了如何在一个 Windows 文件句柄上调用 async_read：读取前 256 个字节，然后把它们保存到缓冲区中；当操作结束时，on_read 被调用，再创建 std::istream 用来传递缓冲区，读取第一行（std::getline），最后把它输出到命令行中。

read_until/async_read_until方法

这些方法在条件满足之前会一直读取。

下面这个例子在读到一个指定的标点符号之前会一直读取：

typedef buffers_iterator<streambuf::const_buffers_type> iterator;
std::pair<iterator, bool> match_punct(iterator begin, iterator end) {
    while (begin != end) {
        if (std::ispunct(*begin))
            return std::make_pair(begin,true);
        ++begin;
    }
    return std::make_pair(end,false);
}
void on_read(const boost::system::error_code &, size_t) {}
...
streambuf buf;
async_read_until(sock, buf, match_punct, on_read);

如果我们想读到一个空格时就结束，我们需要把最后一行修改为：

async_read_until(sock, buf, ' ', on_read);

*_at方法

这些方法用来在一个流上面做随机存取操作。由你来指定 read 和 write 操作从什么地方开始（offset）。

这些方法不支持套接字。它们用来处理流的随机访问；也就是说，流是可以随机访问的。套接字显然不是这样（套接字是不可回溯的）。

下面这个例子告诉你怎么从一个文件偏移为256的位置读取128个字节：

io_service service;
int main(int argc, char* argv[]) {
    HANDLE file = ::CreateFile("readme.txt", GENERIC_READ, 0, 0, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, 0);
    windows::random_access_handle h(service, file);
    streambuf buf;
    read_at(h, 256, buf, transfer_exactly(128));
    std::istream in(&buf);
    std::string line;
    std::getline(in, line);
    std::cout << "first line: " << line << std::endl;
}

异步编程

这部分对异步编程时可能碰到的一些问题进行了深入的探究，我建议你先读一遍，然后在接下来阅读本系列文章的过程中，再经常回过头来看看，从而增强你对这部分的理解。

异步的需求

就像我之前所说的，同步编程比异步编程简单很多。这是因为，线性的思考是很简单的（调用 A，A 结束，调用 B，B 结束，然后继续）；而以事件处理的方式来思考就要困难得多。后面你会碰到这种情况，比如：五件事情，你不知道它们执行的顺序，也不知道它们是否会执行！

尽管异步编程更难，但是你会更倾向于选择使用它，比如：写一个需要处理很多并发访问的服务端。并发访问越多，异步编程就比同步编程越简单。

假设：你有一个需要处理1000个并发访问的应用，从客户端发给服务端的每个信息都会再返回给客户端，以’\n’结尾。

同步方式的代码，1个线程：

using namespace boost::asio;
struct client {
    ip::tcp::socket sock;
    char buff[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.already_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_some(buffer("request_ok\n"));
    // ... 其他消息类型的处理 ...
}

有一种情况是在任何服务端（和任何基于网络的应用）都需要避免的，就是代码无响应的情况。在我们的例子里，我们需要 handle_clients() 方法尽可能少地阻塞。如果方法在某个点上阻塞，任何进来的信息都需要等待方法解除阻塞才能被处理。

为了保持响应，只在一个套接字有数据的时候我们才读，也就是说，if ( clients[i].sock.available() ) on_read(clients[i])。在 on_read 时，我们只读当前可用的；调用 read_until(c.sock, buffer(...), '\n') 会是一个非常糟糕的选择，因为直到我们从一个指定的客户端读取了完整的消息之前，它都是阻塞的（我们永远不知道它什么时候会读取到完整的消息）。

这里的瓶颈就是 on_read_msg() 方法；当它执行时，所有进来的消息都在等待。一个良好的 on_read_msg() 方法实现会保证这种情况基本不会发生，但是它还是会发生（有时候向一个套接字写入数据，缓冲区满了时，它会被阻塞）。

同步方式的代码，10个线程：

using namespace boost::asio;
struct client {
    // ... 和之前一样
    bool set_reading() {
        boost::mutex::scoped_lock lk(cs_);
        if (is_reading_) return false; // 已经在读取
        else {is_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() 方法非常重要：它把”判断是否在读取”和”标记为读取中”合并成了一步（原子操作）。如果分成两步来做，你可能会有两个线程同时为一个客户端判断”是否在读取”，然后两个线程同时为一个客户端调用 on_read，结果就是数据冲突甚至导致应用崩溃。

你会发现代码变得极其复杂。

同步编程有第三个选择，就是为每个连接开辟一个线程。但是当并发的线程增加时，这就成了一种灾难性的情况。

然后，让我们来看异步编程。我们不断地异步读取。当一个客户端请求某些东西时，on_read被调用，然后回应，然后等待下一个请求（然后开始另外一个异步的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_until(clients[i].sock, clients[i].buff, '\n', boost::bind(on_read, boost::ref(clients[i]), _1, _2));
    for ( int i = 0; i < 10; ++i)
        boost::thread(handle_clients_thread);
}
void handle_clients_thread() {
    service.run();
}
void on_write(const error_code & err, size_t) {}
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")
        async_write(c.sock, buffer("request_ok\n"), on_write);
    // ... 其他消息类型的处理 ...
    // 等待同一个客户端下一个读取操作
    async_read_until(c.sock, c.buff, '\n', boost::bind(on_read, boost::ref(c), _1, _2));
}

发现代码变得有多简单了吧？client 结构里面只有两个成员，handle_clients() 仅仅调用了 async_read_until，然后它创建了 10 个线程，每个线程都调用 service.run()。这些线程会处理所有来自客户端的异步 read 操作，然后分发所有向客户端的异步 write 操作。另外需要注意的一件事情是：on_read() 一直在为下一次异步 read 操作做准备（看最后一行代码）。

异步 run(), run_one(), poll(), poll_one()

为了实现监听循环，io_service 类提供了 4 个方法：run()、run_one()、poll() 和 poll_one()。虽然大多数时候使用 service.run() 就可以，但是你还是需要在这里学习其他方法实现的功能。

持续运行

再一次说明，如果有等待执行的操作，run() 会一直执行，直到你手动调用 io_service::stop()。为了保证 io_service 一直执行，通常你添加一个或者多个异步操作，然后在它们被执行时，你继续一直不停地添加异步操作，比如下面代码：

using namespace boost::asio;
io_service service;
ip::tcp::socket sock(service);
char buff_read[1024];
char buff_write[1024] = "ok";
void on_read(const boost::system::error_code &err, std::size_t bytes);
void on_write(const boost::system::error_code &err, std::size_t bytes)
{
    sock.async_read_some(buffer(buff_read), on_read);
}
void on_read(const boost::system::error_code &err, std::size_t bytes)
{
    // ... 处理读取操作 ...
    sock.async_write_some(buffer(buff_write,3), on_write);
}
void on_connect(const boost::system::error_code &err) {
    sock.async_read_some(buffer(buff_read), on_read);
}
int main(int argc, char* argv[]) {
    ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 2001);
    sock.async_connect(ep, on_connect);
    service.run();
}

run_one(), poll(), poll_one() 方法

我在之前说过异步方法的 handler 是在调用了 io_service::run 的线程里被调用的。因为在至少 90%～95% 的时候，这是你唯一要用到的方法，所以我就把它说得简单了。对于调用了 run_one()、poll() 或者 poll_one() 的线程这一点也是适用的。

run_one() 方法最多执行和分发一个异步操作。

你可以认为下面两段代码是等效的：

io_service service;
service.run(); // 或者
while (!service.stopped()) service.run_one();

你可以使用 run_one() 启动一个异步操作，然后等待它执行完成。

io_service service;
bool write_complete = false;
void on_write(const boost::system::error_code & err, size_t bytes)
{ write_complete = true; }
...
std::string data = "login ok";
write_complete = false;
async_write(sock, buffer(data), on_write);
do service.run_one(); while (!write_complete);

还有一些使用 run_one() 方法的例子，包含在 Boost.Asio 诸如 blocking_tcp_client.cpp 和 blocking_udp_client.cpp 的文件中。

poll_one() 方法以非阻塞的方式最多运行一个准备就绪的等待操作。你可以使用 poll_one() 去保证所有已完成 I/O 操作的 handler 都得到执行，同时做一些其他的事情：

io_service service;
while (true) {
    // 运行所有完成了IO操作的handler
    while (service.poll_one()) ;
    // ... 在这里做其他的事情 ...
}

poll() 方法会以非阻塞的方式运行所有等待的操作。下面两段代码是等效的：

io_service service;
service.poll(); // 或者
while (service.poll_one()) ;

所有上述方法都会在失败的时候抛出 boost::system::system_error 异常。这是我们所不希望发生的事情；这里抛出的异常通常都是致命的，也许是资源耗尽，或者是你 handler 的其中一个抛出了异常。另外，每个方法都有一个不抛出异常，而是返回一个 boost::system::error_code 的重载：

io_service service;
boost::system::error_code err;
service.run(err);
if (err) std::cout << "Error " << err << std::endl;

异步工作

异步工作不仅仅指用异步的方式接受客户端到服务端的连接、异步地从一个socket 读取或者写入到socket。它包含了所有可以异步执行的操作。

默认情况下，你是不知道每个异步 handler 的调用顺序的。除了通常的异步调用（来自异步 socket 的读取/写入/接收），你可以使用 service.post() 来使你的自定义方法被异步地调用。例如：

#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <iostream>
using namespace boost::asio;
io_service service;
void func(int i) {
    std::cout << "func called, i= " << i << std::endl;
}

void worker_thread() {
    service.run();
}

int main(int argc, char* argv[]) {
    for ( int i = 0; i < 10; ++i)
        service.post(boost::bind(func, i));
    boost::thread_group threads;
    for (int i = 0; i < 3; ++i)
        threads.create_thread(worker_thread);
    // 等待所有线程被创建完
    boost::this_thread::sleep( boost::posix_time::millisec(500));
    threads.join_all();
}

在上面的例子中，service.post(some_function) 添加了一个异步方法调用。这个方法会让某个调用了 service.run() 的线程去调用给定的 some_function，然后立即返回。在我们的例子中，这个线程是我们之前创建的三个线程中的一个。你不能确定异步方法调用的顺序。你不要期待它们会以我们调用 post() 方法的顺序来调用。下面是运行之前代码可能得到的结果：

func called, i= 0
func called, i= 2
func called, i= 1
func called, i= 4
func called, i= 3
func called, i= 6
func called, i= 7
func called, i= 8
func called, i= 5
func called, i= 9

有时候你会想让一些异步处理方法顺序执行。比如，你去一个餐馆（go_to_restaurant），下单（order），然后吃（eat）。你需要先去餐馆，然后下单，最后吃。这样的话，你需要用到 io_service::strand，它会让你的异步方法被顺序调用。看下面的例子：

using namespace boost::asio;
io_service service;
void func(int i) {
    std::cout << "func called, i= " << i << "/" << boost::this_thread::get_id() << std::endl;
}
void worker_thread() {
    service.run();
}
int main(int argc, char* argv[])
{
    io_service::strand strand_one(service), strand_two(service);
    for ( int i = 0; i < 5; ++i)
        service.post(strand_one.wrap(boost::bind(func, i)));
    for ( int i = 5; i < 10; ++i)
        service.post(strand_two.wrap(boost::bind(func, i)));
    boost::thread_group threads;
    for ( int i = 0; i < 3; ++i)
        threads.create_thread(worker_thread);
    // 等待所有线程被创建完
    boost::this_thread::sleep( boost::posix_time::millisec(500));
    threads.join_all();
}

在上述代码中，我们保证前面的 5 个 func 调用和后面的 5 个 func 调用分别是顺序执行的。func called, i = 0 在 func called, i = 1 之前被调用，然后调用 func called, i = 2……同样 func called, i = 5 在 func called, i = 6 之前，func called, i = 6 在 func called, i = 7 被调用……你需要注意的是尽管方法是顺序调用的，但是不意味着它们都在同一个线程执行。运行这个程序可能得到的一个结果如下：

func called, i= 0/002A60C8
func called, i= 5/002A6138
func called, i= 6/002A6530
func called, i= 1/002A6138
func called, i= 7/002A6530
func called, i= 2/002A6138
func called, i= 8/002A6530
func called, i= 3/002A6138
func called, i= 9/002A6530
func called, i= 4/002A6138

异步 post() VS dispatch() VS wrap()

Boost.Asio 提供了三种让你把处理方法添加为异步调用的方式。在之前的章节中你会看到关于 service.post() 的一个例子，以及运行这个例子可能得到的一种结果。我们对它做一些修改，然后看看 service.dispatch() 是怎么影响输出的结果的：

using namespace boost::asio;
io_service service;
void func(int i) {
    std::cout << "func called, i= " << i << std::endl;
}
void run_dispatch_and_post() {
    for ( int i = 0; i < 10; i += 2) {
        service.dispatch(boost::bind(func, i));
        service.post(boost::bind(func, i + 1));
    }
}
int main(int argc, char* argv[]) {
    service.post(run_dispatch_and_post);
    service.run();
}

在解释发生了什么之前，我们先运行程序，观察结果：

func called, i= 0
func called, i= 2
func called, i= 4
func called, i= 6
func called, i= 8
func called, i= 1
func called, i= 3
func called, i= 5
func called, i= 7
func called, i= 9

偶数先输出，然后是奇数。这是因为我用 dispatch() 输出偶数，然后用 post() 输出奇数。dispatch() 会先在当前线程直接调用 handler 再返回——因为当前线程正运行在 service.run() 里；而 post() 每次都只是把 handler 加入队列，然后立即返回。现在，让我们讲讲 service.wrap(handler)。wrap() 返回了一个仿函数，它可以用来做另外一个方法的参数：

using namespace boost::asio;
io_service service;
void dispatched_func_1() {
    std::cout << "dispatched 1" << std::endl;
}
void dispatched_func_2() {
    std::cout << "dispatched 2" << std::endl;
}
void test(boost::function<void()> func) {
    std::cout << "test" << std::endl;
    service.dispatch(dispatched_func_1);
    func();
}
void service_run() {
    service.run();
}
int main(int argc, char* argv[]) {
    test( service.wrap(dispatched_func_2));
    boost::thread th(service_run);
    boost::this_thread::sleep( boost::posix_time::millisec(500));
    th.join();
}

test(service.wrap(dispatched_func_2)); 会把 dispatched_func_2 包装起来创建一个仿函数，然后传递给 test 当作一个参数。当 test() 被调用时，它会先分发（立即调用）dispatched_func_1，然后再调用 func()。这时，你会发现调用 func() 和 service.dispatch(dispatched_func_2) 是等价的，因为它们是连续调用的。程序的输出证明了这一点：

test
dispatched 1
dispatched 2

io_service::strand 类（用来序列化异步调用）也包含了 dispatch()、post() 和 wrap() 等成员函数，它们的作用和 io_service 的 dispatch()、post() 和 wrap() 是一样的。然而，大多数情况下你只需要把 io_service::strand::wrap() 方法做为 io_service::post() 或者 io_service::dispatch() 方法的参数即可。

保持活动

假设你需要做下面的操作：

io_service service;
ip::tcp::socket sock(service);
char buff[512];
...
read(sock, buffer(buff));

在这个例子中，sock和buff的存在时间都必须比read()调用的时间要长。也就是说，在调用read()返回之前，它们都必须有效。这就是你所期望的；你传给一个方法的所有参数在方法内部都必须有效。当我们采用异步方式时，事情会变得比较复杂。

io_service service;
ip::tcp::socket sock(service);
char buff[512];
void on_read(const boost::system::error_code &, size_t) {}
...
async_read(sock, buffer(buff), on_read);

在这个例子中，sock 和 buff 的存在时间都必须比这次 async_read 操作持续的时间要长，但是 async_read 操作持续的时间我们是不知道的，因为它是异步的。

当使用socket缓冲区的时候，你会有一个buffer实例在异步调用时一直存在（使用boost::shared_array<>）。在这里，我们可以使用同样的方式，通过创建一个类并在其内部管理socket和它的读写缓冲区。然后，对于所有的异步操作，传递一个包含智能指针的boost::bind仿函数给它：

using namespace boost::asio;
io_service service;
struct connection : boost::enable_shared_from_this<connection> {
    typedef boost::system::error_code error_code;
    typedef boost::shared_ptr<connection> ptr;
    connection() : sock_(service), started_(true) {}
    void start(ip::tcp::endpoint ep) {
        sock_.async_connect(ep, boost::bind(&connection::on_connect, shared_from_this(), _1));
    }
    void stop() {
        if (!started_) return;
        started_ = false;
        sock_.close();
    }
    bool started() { return started_; }
private:
    void on_connect(const error_code & err) {
        // 这里你决定用这个连接做什么: 读取或者写入
        if (!err) do_read();
        else stop();
    }
    void on_read(const error_code &err, size_t bytes) {
        if (!started()) return;
        std::string msg(read_buffer_, bytes);
        if (msg == "can_login") do_write("access_data");
        else if (msg.find("data") == 0) process_data(msg);
        else if (msg == "login_fail") stop();
    }
    void on_write(const error_code &err, size_t bytes) {
        do_read();
    }
    void do_read() {
        sock_.async_read_some(buffer(read_buffer_), boost::bind(&connection::on_read, shared_from_this(), _1, _2));
    }
    void do_write(const std::string & msg) {
        if ( !started() ) return;
        // 注意: 因为在做另外一个async_read操作之前你想要发送多个消息,
        // 所以你需要多个写入buffer
        std::copy(msg.begin(), msg.end(), write_buffer_);
        sock_.async_write_some(buffer(write_buffer_, msg.size()), boost::bind(&connection::on_write, shared_from_this(), _1, _2));
    }

    void process_data(const std::string &msg) {
        // 处理服务端来的内容，然后启动另外一个写入操作
    }
private:
    ip::tcp::socket sock_;
    enum {max_msg = 1024};
    char read_buffer_[max_msg];
    char write_buffer_[max_msg];
    bool started_;
};

int main(int argc, char* argv[]) {
    ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 8001);
    connection::ptr(new connection)->start(ep);
    service.run();
}

在所有异步调用中，我们传递一个 boost::bind 仿函数当作参数。这个仿函数内部包含了一个智能指针，指向 connection 实例。只要有一个异步操作等待时，Boost.Asio 就会保存 boost::bind 仿函数的拷贝，这个拷贝保存了指向连接实例的一个智能指针，从而保证 connection 实例保持活动。问题解决！

当然，connection 类仅仅是一个框架类；你需要根据你的需求对它进行调整（它看起来会和你实际的应用有相当大的不同）。

你需要注意的是创建一个新的连接是相当简单的：connection::ptr(new connection)->start(ep)。这个方法启动了到服务端的（异步）连接。当你需要关闭这个连接时，调用 stop()。

当实例被启动时（start()），它会异步地连接到服务端。当连接完成时，on_connect() 被调用。如果没有错误发生，它启动一个 read 操作（do_read()）。当 read 操作结束时，你就可以解析这个消息；当然你应用的 on_read() 看起来会各种各样。而当你写回一个消息时，你需要把它拷贝到缓冲区，然后像我在 do_write() 方法中所做的一样将其发送出去，因为这个缓冲区同样需要在这个异步写操作中一直存活。最后需要注意的一点——当写回时，你需要指定写入的数量，否则，整个缓冲区都会被发送出去。

总结

网络 API 实际上要繁杂得多，这个章节只是做为一个参考，当你在实现自己的网络应用时可以回过头来看看。

Boost.Asio 实现了端点的概念，你可以认为是 IP 和端口。如果你不知道准确的 IP，你可以使用 resolver 对象将主机名，例如 www.yahoo.com 转换为一个或多个 IP 地址。

我们也可以看到 API 的核心——socket 类。Boost.Asio 提供了 TCP、UDP 和 ICMP 的实现。而且你还可以用你自己的协议来对它进行扩展；当然，这个工作不适合缺乏勇气的人。

异步编程是刚需。你应该已经明白为什么有时候需要用到它，尤其在写服务端的时候。使用 service.run() 来实现异步循环就可以满足大部分需求；但有时候你需要更进一步，尝试使用 run_one()、poll() 或者 poll_one()。

当实现异步时，你可以异步执行你自己的方法；使用 service.post() 或者 service.dispatch()。

最后，为了使 socket 和缓冲区（read 或者 write）在整个异步操作的生命周期中一直活动，我们需要采取特殊的防护措施。你的连接类需要继承自 boost::enable_shared_from_this，然后在内部保存它需要的缓冲区，而且每次异步调用都要传递一个智能指针给 this 操作。

下一章会进行实战操作；在实现回显客户端/服务端应用时会有大量的编程实践，请看本站的《Boost.Asio：客户端和服务端》。