Boost.Asio命名空间

Boost.Asio所有东西都放在boost::asio命名空间或者子命名空间:

  • boost::asio:核心类和函数放在这里。重要的类有io_service和streambuf。重要的自由函数,比如read,read_at,read_util,它们相应的异步版本,以及同步写和异步写函数。
  • boost::asio::ip:网络部分属于这里。重要的类有address,endpoint,tcp,udp,icmp。重要的自由函数connect和async_connect。注意boost::asio::ip::tcp::socket只是一个boost::asio::ip::tcp中的typedef。
  • boost::asio::error:包含调用I/O例程的错误码。
  • boost::asio::ssl:包含处理SSL的类。
  • boost::asio::local:包含POSIX相关的类。
  • boost::asio::windows:包含Windows相关的类。

IP地址

为了处理IP地址,Boost.Asio提供ip::address,ip::address_v4和ip::address_v6类。以下是一些最重要的函数:

  • ip::address(v4_or_v6_address):这个函数转换一个v4或v6地址为ip::address。
  • ip::address::from_string(str):从一个IPv4或IPv6地址字符串创建地址。
  • ip::address::to_string():返回一个友好可读的地址字符串。
  • ip::address_v4::broadcast([addr, mask]):创建一个广播地址。
  • ip::address_v4::any():返回一个代表任何地址的地址。
  • ip::address_v4::loopback(),ip_address_v6::loopback():返回回路地址
  • ip::host_name():返回当前主机名。

你可能最常用到ip::address::from_string:

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

端点

端点是你要连接的地址和端口。不同类型有自己的endpoint类,比如ip::tcp::endpoint,ip::udp::endpoint和ip::icmp::endpoint。

如果你想连接本机80端口,这样:

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

创建端点有3种方法:

  • endpoint():默认构造函数,有时候可以用于UDP/ICMP套接字。
  • endpoint(protocol, port):通常用于服务端套接字接受新连接。
  • endpoint(addr, port):创建一个指定地址和端口的端点。

下面是例子:

1
2
3
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);

如果想要连接到一个主机名,可以这样做:

1
2
3
4
5
6
7
// outputs "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;

如果resolve()函数成功,它将返回至少一个入口。你可以使用第一个,也可以遍历所有的。

套接字

Boost.Asio有3中socket类:ip::tcp,ip::udp和ip::icmp,并且可以扩展。你可以创建自己的socket类。

你可以认为ip::tcp,ip::udp和ip::icmp类是占位符,它们提供其他类和函数的简便访问:

  • ip::tcp::socket, ip::tcp::acceptor, ip::tcp::endpoint, ip::tcp::resolver, ip::tcp::iostream
  • ip::udp::socket, ip::udp::endpoint, ip::udp::resolver
  • ip::icmp::socket, ip::icmp::endpoint, ip::icmp::resolver

socket类创建相应的套接字,创建时需要传递io_service实例:

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

每一个socket名字是一个typedef:

  • ip::tcp::socket = basic_stream_socket<tcp>
  • ip::udp::socket = basic_datagram_socket<udp>
  • ip::icmp::socket = basic_raw_socket<icmp>

同步错误码

所有的同步函数重载了抛出异常或返回错误码的版本,像下面的例子:

1
2
3
sync_func(arg1, arg2 ... argN); // throws
boost::system::error_code ec;
sync_func(arg1 arg2, ..., argN, ec); // returns error code

Socket成员函数

不是所有的函数对每一种类型的socket都可用。注意所有的异步函数立刻返回,同步函数只有在操作完成才会返回。

连接相关的函数

  • assign(protocol, socket):这个函数将一个原始套接字赋给一个socket实例。使用它处理遗留代码。
  • open(protocol):这个函数使用指定IP协议(v4或v6)打开一个socket。这个函数主要用于UDP/ICMP套接字,或者服务器套接字。
  • bind(endpoint):绑定到指定端点。
  • connect(endpoint):同步连接到指定端点。
  • async_connect(endpoint):异步连接到指定端点。
  • is_open():如果socket已打开返回true。
  • close():关闭套接字。此套接字任何异步操作立刻取消并以error::operation_aborted错误码结束。
  • shutdown(type_of_shutdown):从现在开始禁止send,receive操作。
  • cancel():取消此套接字所有异步操作。所有异步操作立刻以error::operation_aborted错误码结束。

例子如下:

1
2
3
4
5
6
7
8
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();

读写函数

对于所有异步读写函数,其处理函数的签名为:

1
void handler(const boost::system::error_code &e, size_t bytes)

  • async_receive(buffer, [flags,] handler):从套接字开始异步receive操作。
  • async_read_some(buffer, handler):等价于async_receive(buffer, handler)。
  • async_receive_from(buffer, endpoint[, flags], handler):从指定端点开始异步receive操作。
  • async_send(buffer[, flags], handler):开始异步发送buffer中的数据。
  • async_write_some(buffer, handler):等价于async_send(buffer, handler)。
  • async_send_to(buffer, endpoint, handler):开始异步发送buffer中的数据到指定端点。
  • receive(buffer[, flags]):同步读取数据到指定buffer。这个函数阻塞知道收到数据或一个错误发生。
  • read_some(buffer):等价于receive(buffer)。
  • receive_from(buffer, endpoint[, flags]):同步地从指定端点读取数据到指定buffer。这个函数阻塞直到受到数据或错误发生。
  • send(buffer[, flags]):同步发送buffer中的数据。函数阻塞直到受到数据或发生错误。
  • write_some(buffer):等价于send(buffer)。
  • send_to(buffer, endpoint[, flags]):同步发送buffer中的数据到指定端点。函数阻塞直到受到数据或发生错误。
  • avaliable():返回可以同步读取的字节数,不用阻塞。

flags的默认值是0,但可以是下列的组合:

  • ip::socket_type::socket::message_peek:这个标志只是偷看消息。它返回这个消息,但是下一次调用会重读这个消息。
  • ip::socket_type::socket::message_out_of_band:这个标志处理out-of-band(OOB)数据。OOB数据是标记为比普通数据更重要的数据。
  • ip::socket_type::socket::message_do_not_route:这个标志指定消息应该不使用路由表发送。
  • ip::socket_type::socket::message_end_of_record:这个标志指示数据标志一个记录的结尾。在Windows下不支持。

如果你使用下面这段代码,你最可能使用message_peek:

1
2
3
4
5
char buff[1024];
sock.receive(buffer(buff), ip::tcp::socket::message_peek);
memset(buff,1024, 0);
// re-reads what was previously read
sock.receive(buffer(buff));

同步读写一个TCP套接字:

1
2
3
4
5
6
7
ip::tcp::endpoint ep( ip::address::from_string("127.0.0.1"), 80);
ip::tcp::socket sock(service);
sock.connect(ep);
sock.write_some(buffer("GET /index.html\r\n"));
std::cout << "bytes available " << sock.available() << std::endl;
char buff[512];
size_t read = sock.read_some(buffer(buff));

同步读写一个UDP套接字:

1
2
3
4
5
6
7
ip::udp::socket sock(service);
sock.open(ip::udp::v4());
ip::udp::endpoint receiver_ep("87.248.112.181", 80);
sock.send_to(buffer("testing\n"), receiver_ep);
char buff[512];
ip::udp::endpoint sender_ep;
sock.receive_from(buffer(buff), sender_ep);

注意为了使用receive_from从一个UDP套接字读取数据,你需要一个默认构造的端点。

异步读取一个UDP服务器套接字:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
using namespace boost::asio;
io_service service;
ip::udp::socket sock(service);
boost::asio::ip::udp::endpoint sender_ep;
char buff[512];
void on_read(const boost::system::error_code & err, std::size_t read_bytes) {
    std::cout << "read " << read_bytes << std::endl;
    sock.async_receive_from(buffer(buff), sender_ep, on_read);
}
int main(int argc, char* argv[]) {
    ip::udp::endpoint ep(ip::address::from_string("127.0.0.1"), 8001);
    sock.open(ep.protocol());
    sock.set_option(boost::asio::ip::udp::socket::reuse_address(true));
    sock.bind(ep);
    sock.async_receive_from(buffer(buff,512), sender_ep, on_read);
    service.run();
}

套接字控制

这些函数处理高级套接字选项:

  • get_io_service():返回构造时传递的io_service实例。
  • get_option(option):返回一个套接字选项。
  • set_option(option):设置套接字选项
  • io_control(cmd):在套接字上执行一个I/O命令。

以下是你可以读取或设置的套接字选项:

名字 描述 类型
broadcast 如果真,允许广播消息 bool
debug 如果真,使socket-level调试生效 bool
enable_connection_aborted 如果真,报告在accept()时连接被中止 bool
receive_buffer_size 套接字的接收缓冲区大小 int
receive_low_watermark 提供处理套接字输入的最小字节数 int
reuse_address 如果真,套接字可以绑定一个已经在使用的地址 bool
send_buffer_size 套接字发送缓冲区大小 int
send_low_watermark 提供处理套接字输出的最小字节数 int
ip::v6_only 如果真,只允许IPv6通信 bool

每一个名字代表内部socket的一个typedef或一个类。以下是例子:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), 80);
ip::tcp::socket sock(service);
sock.connect(ep);
// TCP socket can reuse address
ip::tcp::socket::reuse_address ra(true);
sock.set_option(ra);
// get sock receive buffer size
ip::tcp::socket::receive_buffer_size rbs;
sock.get_option(rbs);
std::cout << rbs.value() << std::endl;
// set sock's buffer size to 8192
ip::tcp::socket::send_buffer_size sbs(8192);
sock.set_option(sbs);

v#### TCP vs UDP vs ICMP ####

如果一个成员函数不在下表,说明它对所有socket类可用:

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

其它函数

  • local_endpoint():返回本地地址。
  • remote_endpoint():返回远程地址。
  • native_handle():返回原始套接字的句柄。你只有在想要调用原始套接字API的时候需要这个函数。
  • non_blocking():如果套接字非阻塞返回true,否则false。
  • native_non_blocking():和non_blocking一样,但是它在原始套接字上调用原始的套接字API。
  • at_mark():如果套接字即将读取OOB数据,返回true。很少使用。

其它考虑

最后需要注意的是,一个套接字实例不能被复制,因为复制构造函数和赋值操作符无法访问。如果你想要创建拷贝,使用共享指针:

1
2
3
4
5
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

Socket缓冲区

当读取或写入一个套接字,需要一个缓冲区,它将保存进来和流出的数据。缓冲区的内存必须比I/O操作更长久。你必须保证I/O操作持续期间它不被释放或超出范围。

这个问题有下面几种解决方法:

  • 使用全局缓冲区。
  • 创建一个缓冲区,当操作完成时销毁它。
  • 使用一个连接对象维护套接字和额外的数据,比如缓冲区。

缓存包装器函数

任何时候我们需要一个缓冲区用于读写操作时,用buffer()函数包装真实的缓冲区对象。它包装任何缓冲区为一个类,允许Boost.Asio函数迭代访问缓冲区。比如你使用下面的代码:

1
sock.async_receive(some_buffer, on_read);

some_buffer实例需要满足一些条件,即ConstBufferSequence或MutableBufferSequence。使用buffer()函数就可以满足这些条件。

简单地说,你可以用buffer()函数包装下面的对象:

  • char[] const数组
  • void*指针和大小
  • std::string
  • POD[] const数组(POD代表plain old data,即构造函数和析构函数不用做任何事)
  • std::vector<POD>
  • boost::array<POD>
  • std::array<POD>

读/写/连接自由函数

Boost.Asio提供自由函数处理I/O。

连接函数

连接socket到一个端点:

  • connect(socket, begin[, end][, condition]):同步连接,尝试begin和end之间每一个端点。begin迭代器是socket_type::resolver::query调用返回的结果。你可以提供一个条件函数在每次连接尝试前调用。它的签名是Iterator connect_condition(const boost::system::error_code &ec, Iterator next)。
  • async_connect(socket, begin[, end][, condition], handler):执行异步连接,最后调用完成处理函数。处理函数的签名是void handler(const boost::system::error_code &ec, Iterator iterator)。

示例如下:

1
2
3
4
5
6
7
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将你从尝试每一个地址的负担中释放出来。

读/写函数

  • async_read(stream, buffer[, completion], handler):从stream异步读取数据。完成后调用处理函数,其签名为void handler(const boost::system::error_code &ec, size_t bytes)。你可以指定一个完成函数。每一次成功读后都调用一次完成函数,并告诉Boost.Asio异步读操作是否完成,如果没有继续读。完成函数的签名为size_t completion(const boost::system::error_code &ec, size_t bytes_transfered)。当完成函数返回0时,我们认为读操作完成。如果返回非0值,它指示下一次async_read_some操作读取的最大字节数。
  • async_write(stream, buffer[, completion], handler):异步写数据到stream。参数的意义和async_read类似。
  • read(stream, buffer[, completion]):从stream同步读取数据。参数的意义和async_read类似。
  • write(stream, buffer[, completion]):同步写数据到stream。参数的意义和async_read类似。

每一个读或写操作在下面这些条件发生时结束:

  • 提供的缓冲区已满(对于读)或在缓冲区的数据都被写完(对于写)
  • 完成函数返回0,如果你提供了这样一个函数
  • 一个错误发生

下面的代码会异步读知道发现’\n’:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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_at_least(n)
  • transfer_exactly(n)
  • transfer_all()

例子如下:

1
2
3
4
char buff[512];
void on_read(const boost::system::error_code &, size_t) {}
// read exactly 32 bytes
async_read(sock, buffer(buff), transfer_exactly(32), on_read);

最后4个函数,使用继承std::streambuf的stream_buffer函数而不是通常的buffer。STL stream和stream_buffer非常灵活。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
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();
}

read_until/async_read_util函数

这些函数一直读直到一个条件满足:

  • async_read_until(stream, stream_buffer, delim, handler):开始一个异步读操作。当遇到一个delim时,读操作停止。delim可以是任意一个字符,std::string或boost::regex。处理函数的签名为void handler(const boost::system::error_code &ec, size_t bytes)。
  • async_read_until(stream, stream_buffer, completion, handler):这个函数和前一个一样,只是将delim替换为完成函数,其签名是pair<iterator, bool> completion(iterator begin, iterator end)。迭代器参数是buffers_iterator<streambuf::const_buffers_type>。你需要记住的是迭代器类型是随机访问迭代器。返回true说明读操作应该停止。
  • read_until(stream, stream_buffer, delim):同步读,参数意义与async_read_until一样。
  • read_until(stream, stream_buffer, completion):同步读,参数意义与async_read_until一样。

下面的代码将读到一个标点符号为止:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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);
    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);

如果想读到一个空格,修改最后一行:

1
async_read_until(sock, buff, ' ', on_read);

*_at函数

这些函数在一个stream上做随机读写操作。你指定读写操作从哪开始(offset):

  • async_read_at(stream, offset, buffer[, completion], handler):在指定流上从offset开始异步读操作。处理函数的签名是void handler(const boost::system::error_code &ec, size_t bytes)。缓冲区可以是通常的buffer()包装器或一个streambuf函数。如果指定了一个完成函数,它在每次成功读操作后被调用,并告诉Boost.Asio操作async_read_at是否完成,其签名为size_t completion(const boost::system::error_code &ec, size_t bytes)。当完成函数返回0时,我们认为读操作完成;如果返回非0值,它指示下一次async_read_some_at调用要读取的最大字节数。
  • async_write_at(stream, offset, buffer[, completion], handler):开始一个异步写操作。参数的意思和async_read_at一样。
  • read_at(stream, offset, buffer[, completion]):同步读,参数的意思和async_read_at一样。
  • write_at(stream, offset, buffer[, completion]):同步读,参数的意思和async_read_at一样。

这些函数不处理socket,它们处理随机访问流。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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;
}

异步编程

这部分将深入钻研进行异步编程会碰到的一些问题。

异步的需求

尽管异步编程更难,你可能还是会选择它。比如说编写需要处理非常多并行客户端的服务器。并行客户端越多,异步编程比同步编程越容易。

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

为了实现监听循环,io_service类提供4个函数,run(),run_one(),poll()和poll_one()。

永久运行

run()会一直运行,只要有追加的操作执行或者你手动调用io_service::stop()。为了保持io_service实例运行,通常添加一个或多个异步操作,并当它们执行时,保持继续添加异步操作:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
using namespace boost::asio;
io_service service;
ip::tcp::socket sock(service);
char buff_read[1024], 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)
{
    // ... process the read ...
    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()函数

run_one()函数会执行和分配最多一个异步操作:

  • 如果没有追加操作,函数立刻返回0
  • 如果有追加操作,函数阻塞知道第一个操作执行,并返回1

考虑下述等价代码:

1
2
3
io_service service;
service.run(); // OR
while (!service.stopped()) service.run_once();

你可以使用run_once()开始一个异步操作,然后等待它完成:

1
2
3
4
5
6
7
8
9
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_once() while (!write_complete);

poll_one函数运行最多一次准备运行的追加操作,非阻塞:

  • 如果有至少一个准备运行的非阻塞追加操作,poll_one运行它并返回1
  • 否则,立刻返回0

准备运行的非阻塞追加操作,通常意味着任何:

  • 一个定时器超时了,且它的async_wait回调函数需要被调用。
  • 一个I/O操作完成了(比如async_read),且其回调函数需要被调用。
  • 一个之前添加到io_services队列的自定义的回调函数。

poll()函数运行所有添加的操作,不用阻塞。下面代码等价:

1
2
3
io_service service;
service.poll(); // OR
while (service.poll_one());

所有之前的函数失败时抛出boost::system::system_error异常。这不应该发生,这里抛出的错误通常是致命的。每一个函数也有不抛异常而返回错误码的重载版本。

1
2
3
4
io_service service;
boost::system::error_code err = 0;
service.run(err);
if (err) std::cout << "Error " << err << std::endl;

异步工作

异步工作不仅仅是异步接受客户端连接,异步读取或写到套接字。它包含任何可以异步执行的操作。

默认情况下,你不知道异步处理函数的调用顺序。你可以使用service.post()抛出自定义的函数使其可以异步调用:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
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);
    // wait for all threads to be created
    boost::this_thread::sleep(boost::posix_time::millisec(500));
    threads.join_all();
}

有的时候你想要某些异步处理函数按顺序调用。你可以使用io_service::strand,它将顺序调用你的异步处理函数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
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);
    // wait for all threads to be created
    boost::this_thread::sleep( boost::posix_time::millisec(500));
    threads.join_all();
}

异步post() vs dispatch() vs wrap()

Boost.Asio提供3种方法添加你的函数异步调用:

  • service.post(handler):这个函数保证在请求io_service添加处理函数后立刻返回。处理函数会在之后被某个调用了service.run()的线程调用。
  • service.dispatch(handler):请求io_service调用指定的处理函数,同时如果当前线程调用了service.run(),它可以在函数内执行处理函数。
  • service.wrap(handler):这个函数创建一个包装函数。当包装函数被调用时,会调用service.dispatch(handler)。

看看dispatch如何影响输出:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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();
}

wrap()函数返回一个函数对象,以供将来使用:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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();
}

io_service::strand类也包含成员函数poll(),dispatch()和wrap()。其意义和io_service的一样。然而大多数时间你只会使用io_service::strand::wrap()作为io_service::poll()或io_service::dispatch()的参数。

保持活着

当使用套接字缓冲区时,你可以用一个buffer实例度过一个异步调用。我们可以使用同样的原理创建一个类,内部保存套接字和读/写缓冲区。然后对于所有异步调用,传递一个共享指针给boost::bind函数:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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) {
        // here you decide what to do with the connection: read or write
        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;
        // note: in case you want to send several messages before
        // doing another async_read, you'll need several write buffers!
        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) {
        // process what comes from server, and then perform another write
    }
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);
}

在所有的异步调用中,我们传递一个boost::bind函数对象作为参数。这个函数对象内部保存一个共享指针指向connection实例。只要还有追加的异步操作,Boost.Asio将保存一个boost::bind函数对象,它又保存一个共享指针指向connection实例,因此保持连接活着。