PHP网络编程之深入Libevent（十五节）
PHP网络编程之深入Libevent（十五节）
老李秀
发布于 2020-02-19 12:22:26
发布于 2020-02-19 12:22:26
大家周末好，这里有趣有用广告少的公众号高性能API社区，我是老李，本文属于《PHP网络编程》系列中的一个章节。
前两天老孟跟我说：
毫不要脸地说，我写的这些文章都不属于快餐消耗品，你不动手亲自实践是压根搞不定的，哪儿有那么容易就能得到的认知啊！况且我讲的并不全，有很多资料知识是需要你自己搜索补充的。而且老李自认为很少在公众号里瞎TM发没用的文章，几乎篇篇都是干货、水很少、很紧致，老铁们啊，听我一句劝：
春宵一刻值千金，绝知此事要躬行
我看了一下《PHP网络编程》整本书的整体进度，由于最近我周六日火力超频全开的缘故，已经将近完成三分之二了。作为作者，这本书是我对PHP语言的一份贡献和热情，是多年从业的一个厚积薄发的总结，是对《UNIX网络编程》的致敬；作为读者，如果你能紧紧跟随着这本书的脚本，你将能掀开高性能服务器基石的面纱，以后无论你是使用Swoole还是Workerman甚至NodeJS，只要是基于事件的高性能服务器，无论是什么编程语言，你都能很快入手学习掌握。我再次强调一遍：这种xue微偏底层一丢丢的基础知识，绝非类似于《XXX框架实战小程序》、《YYY框架实战电商》，这种基础知识看了后不会有立竿见影的效果，甚至你在工作里CURD都用不到。
经过了我精心设计铺垫的好几个章节后，让我们继续尝试研究PHP-Libevent（epoll），在这里我再次提醒：PHP中对Libevent的实现是一个叫做event的扩展，我们文章将依该扩展为准进行演示，请自行安装（我假装你们都知道如何安装PHP扩展）。
在前面的章节里，老李给大伙儿表演了一波儿Select IO复用实现的聊天室还有HTTP服务器，在HTTP服务器那个章节里还免费送了一波儿HTTP协议学习方法。真是隔着胸前两坨弹跳的白肉球子都无法阻挡老李迸发出来的良心光芒，在知识付费市场镰刀碰镰刀的现如今，这种行为就犹如韩红老师直接送口罩到医院门口、犹如陈光标现场送现金、犹如你瑞幸账号躺着的0折拉新优惠券...
之前我用两个章节来铺垫PHP中如何搞epoll操作：
今天继续搞一波儿epoll，先来使用event扩展来实现最基础的网络IO，下面的demo代码非常简单，大概就是把客户端飞过来的数据echo出来：
<?php
$s_host = '0.0.0.0';
$i_port = 6666;
$r_listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEADDR, 1 );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEPORT, 1 );
socket_bind( $r_listen_socket, $s_host, $i_port );
socket_listen( $r_listen_socket );
// 将$listen_socket设置为非阻塞IO
socket_set_nonblock( $r_listen_socket );
// 这两个数组级别的变量非常有意思
// 一个用于保存event对象
// 一个用于保存client的连接socket
$a_event_array = array();
$a_client_array = array();
// 创建event-base
$o_event_base = new EventBase();
$s_method_name = $o_event_base->getMethod();
// 确保自己用的是epoll
if ( 'epoll' != $s_method_name ) {
exit( "not epoll" );
}
// 在$listen_socket上添加一个 持久的读事件
// 为啥是读事件？
// 因为$listen_socket上发生事件就是：客户端建立连接
// 所以，应该是读事件
// 而且，我们应该用上 PERSIST 将事件设置为持久事件
$o_event = new Event( $o_event_base, $r_listen_socket, Event::READ | Event::PERSIST, function( $r_listen_socket, $i_event_flag, $o_event_base ) {
global $a_event_array;
global $a_client_array;
// socket_accept接受连接，生成一个新的socket，一个客户端连接socket
$r_connection_socket = socket_accept( $r_listen_socket );
// 注意这个操作：将客户端连接保存到数组...
// 如果没有这行，客户端连接上后自动断开...
$a_client_array[] = $r_connection_socket;
// 在这个客户端连接socket上添加 持久的读事件
// 也就说 要从客户端连接上读取消息
$o_event = new Event( $o_event_base, $r_connection_socket, Event::READ | Event::PERSIST, function( $r_connection_socket ) {
$s_content = socket_read( $r_connection_socket, 1024 );
echo $s_content;
} );
$o_event->add();
// 注意这个操作：将事件保存到事件数组...
// 如果没有这行，那么事件会丢失，客户端发送的消息毛都收不到
$a_event_array[] = $o_event;
}, $o_event_base );
$o_event->add();
// 注意这个操作：将事件保存到事件数组...
$a_event_array[] = $o_event;
// loop起来...
$o_event_base->loop();首先关于EventBase、Event不再赘述，前面入门epoll那篇里我几乎说过了所有Event扩展提供的类的简要功能说明。上述代码的主要流程非常简单：
- 一创建好一个非阻塞的$listen_socket
- 二在这个$listen_socket上创建一个持久的读事件
- 三是当在$listen_socket发现可读事件后就执行socket_accept()操作
- 四是socket_accept()会生成新的客户端连接socket然后给这个socket添加持久读取事件
- 五是当在客户端连接socket上发现可读事件后就从上面读取内容并使用echo显示出来
值得特殊注意的是$a_event_array和$a_client_array()这两个数组。其中$a_event_array用于保存所有创建的event对象，如果不保存，事件将会丢失，什么都读不到；其中$a_client_array()用于保存客户端连接socket，如果不保存，客户端的反应就是一连接上来就会被关闭，你可以用telnet试一下。
一波儿操作，放眼望去尽是读事件，是时候表演一波儿写事件了，说白了就是服务器向客户端写内容。其实这事儿看起来应该挺简单的，好像大概似乎按葫芦画瓢就能搞定，但，是么？来来，琢磨一下，什么时候向客户端写入数据，是在Event::READ事件的回调函数中吗？这么做听起来是顺理成章的，在回调中读取了客户端飞过来的数据后，马上使用socket_write()等把数据再飞回去给客户端...
MD，那还要Event::WRITE有何用？
所以这个事儿，可不是泥想象中辣么简单。到了这个关键的节骨眼上，请让老李手把手给你小刀剌屁股---开开眼。让我们再次重提一下可读／可写的条件是什么，当然了你们重点体会一下可写的条件：
- Event::READ：只要网络缓冲中还有数据，回调函数就会被触发
- Event::WRITE：只要塞给网络缓冲的数据被写完，回调函数就会被触发
你们好好感受一下「只要塞给网络缓冲的数据被写完」，仔细琢磨...我整个demo你们试一下看看啥结果：
<?php
$s_host = '0.0.0.0';
$i_port = 6666;
$r_listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEADDR, 1 );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEPORT, 1 );
socket_bind( $r_listen_socket, $s_host, $i_port );
socket_listen( $r_listen_socket );
// 将$listen_socket设置为非阻塞IO
socket_set_nonblock( $r_listen_socket );
$a_event_array = array();
$a_client_array = array();
// 创建event-base
$o_event_base = new EventBase();
$s_method_name = $o_event_base->getMethod();
if ( 'epoll' != $s_method_name ) {
exit( "not epoll" );
}
// 在$listen_socket上添加一个 读事件
// 为啥是读事件？
// 因为$listen_socket上发生事件就是：客户端建立连接
// 所以，应该是读事件
$o_event = new Event( $o_event_base, $r_listen_socket, Event::READ | Event::PERSIST, function( $r_listen_socket, $i_event_flag, $o_event_base ) {
global $a_event_array;
global $a_client_array;
// socket_accept接受连接，生成一个新的socket，一个客户端连接socket
$r_connection_socket = socket_accept( $r_listen_socket );
$a_client_array[] = $r_connection_socket;
// 在这个客户端连接socket上添加 读事件
// 当这个客户端连接socket一旦满足可写条件，我们就可以向socket中写数据了
$o_write_event = new Event( $o_event_base, $r_connection_socket, Event::WRITE | Event::PERSIST, function( $r_connection_socket, $i_event_flag ) {
echo "Event::write回调".PHP_EOL;
// 注意，这个sleep是为了保护你...
sleep( 1 );
} );
$o_write_event->add();
$a_event_array[ intval( $r_connection_socket ) ]['write'] = $o_write_event;
}, $o_event_base );
$o_event->add();
$o_event_base->loop();上面的demo代码，我们只在客户端socket上做了一个写事件，然后用telnet连接上来，但仅仅是连接上来别的什么都别做，你注意到了服务端了么：在不断地打印「Event::write回调」这句话，加sleep真的是为了照顾配置低电脑...然后我们再结合前面说的「只要塞给网络缓冲的数据被写完」琢磨一下，实际上客户端一连上来但别的什么没做，这会儿可不就是满足socket可写这个条件么？因为缓冲区中内容已经发送完了，来吧，请发送下一波儿！
一般说来一个完整常规的写事件的使用方法是：当Event::READ事件发生后，在回调函数中首先读取数据，然后准备一个发送数据的自定义缓冲区，当这个发送数据的自定义缓冲区（注意不是socket缓冲区）中没有数据后，在客户端socket上搞一发写事件并挂起（执行add()方法），然后当Event::WRITE事件发生后开始执行写回调，在写回调里完成逻辑后，将该写事件del掉即可，一般来说都是这么用的。我们基于这种用法逻辑来实现一个聊天室：
<?php
$s_host = '0.0.0.0';
$i_port = 6666;
$r_listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEADDR, 1 );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEPORT, 1 );
socket_bind( $r_listen_socket, $s_host, $i_port );
socket_listen( $r_listen_socket );
// 将$listen_socket设置为非阻塞IO
socket_set_nonblock( $r_listen_socket );
$a_event_array = array();
$a_client_array = array();
// 创建event-base
$o_event_base = new EventBase();
$s_method_name = $o_event_base->getMethod();
if ( 'epoll' != $s_method_name ) {
exit( "not epoll" );
}
// 在$listen_socket上添加一个 读事件
// 为啥是读事件？
// 因为$listen_socket上发生事件就是：客户端建立连接
// 所以，应该是读事件
$o_event = new Event( $o_event_base, $r_listen_socket, Event::READ | Event::PERSIST, function( $r_listen_socket, $i_event_flag, $o_event_base ) {
global $a_event_array;
global $a_client_array;
// socket_accept接受连接，生成一个新的socket，一个客户端连接socket
$r_connection_socket = socket_accept( $r_listen_socket );
$a_client_array[] = $r_connection_socket;
// 在这个客户端连接socket上添加 读事件
// 也就说 要从客户端连接上读取消息
$o_read_event = new Event( $o_event_base, $r_connection_socket, Event::READ | Event::PERSIST, function( $r_connection_socket, $i_event_flag, $o_event_base ) {
$s_content = socket_read( $r_connection_socket, 1024 );
echo "接受到：".$s_content;
// 理论上这里应有一个自定义的发送数据缓冲区，其实就是PHP字符串...
// 这里我就不演示了
// 原则就是：当这个自定义的数据缓冲区没数据后，执行下面逻辑..
// 实际上，Workerman里的$connection->send()方法中
// 就有一个buffer，其实就是所谓的自定义缓冲区
// 在这个客户端连接socket上添加 读事件
// 当这个客户端连接socket一旦满足可写条件，我们就可以向socket中写数据了
global $a_event_array;
global $a_client_array;
$o_write_event = new Event( $o_event_base, $r_connection_socket, Event::WRITE | Event::PERSIST, function( $r_connection_socket, $i_event_flag ) use( &$a_event_array, &$a_client_array, $s_content ) {
foreach( $a_client_array as $r_target_socket ) {
if ( intval( $r_target_socket ) != intval( $r_connection_socket ) ) {
socket_write( $r_target_socket, $s_content, strlen( $s_content ) );
}
}
// 在写回调中逻辑执行完毕后，将该写事件删除掉...
$o_event = $a_event_array[ intval( $r_connection_socket ) ]['write'];
$o_event->del();
unset( $a_event_array[ intval( $r_connection_socket ) ]['write'] );
} );
$o_write_event->add();
$a_event_array[ intval( $r_connection_socket ) ]['write'] = $o_write_event;
}, $o_event_base );
$o_read_event->add();
$a_event_array[ intval( $r_connection_socket ) ]['read'] = $o_read_event;
}, $o_event_base );
$o_event->add();
//$a_event_array[] = $o_event;
$o_event_base->loop();上面代码你们用多个telnet连接上来，可以自娱自乐一下，但是上面代码实在是太TM丑了，我们简单封装一下，让TA变得xue微优雅一下：
<?php
$s_host = '0.0.0.0';
$i_port = 6666;
$r_listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEADDR, 1 );
socket_set_option( $r_listen_socket, SOL_SOCKET, SO_REUSEPORT, 1 );
socket_bind( $r_listen_socket, $s_host, $i_port );
socket_listen( $r_listen_socket );
// 将$listen_socket设置为非阻塞IO
socket_set_nonblock( $r_listen_socket );
$a_event_array = array();
$a_client_array = array();
// 创建event-base
$o_event_base = new EventBase();
$s_method_name = $o_event_base->getMethod();
if ( 'epoll' != $s_method_name ) {
exit( "not epoll" );
}
function read_callback( $r_connection_socket, $i_event_flag, $o_event_base ) {
$s_content = socket_read( $r_connection_socket, 1024 );
echo "接受到：".$s_content;
// 在这个客户端连接socket上添加 读事件
// 当这个客户端连接socket一旦满足可写条件，我们就可以向socket中写数据了
global $a_event_array;
global $a_client_array;
$o_write_event = new Event( $o_event_base, $r_connection_socket, Event::WRITE | Event::PERSIST, 'write_callback', array(
'content' => $s_content,
) );
$o_write_event->add();
$a_event_array[ intval( $r_connection_socket ) ]['write'] = $o_write_event;
}
function write_callback( $r_connection_socket, $i_event_flag, $a_data ) {
global $a_event_array;
global $a_client_array;
$s_content = $a_data['content'];
foreach( $a_client_array as $r_target_socket ) {
if ( intval( $r_target_socket ) != intval( $r_connection_socket ) ) {
socket_write( $r_target_socket, $s_content, strlen( $s_content ) );
}
}
$o_event = $a_event_array[ intval( $r_connection_socket ) ]['write'];
$o_event->del();
unset( $a_event_array[ intval( $r_connection_socket ) ]['write'] );
}
function accept_callback( $r_listen_socket, $i_event_flag, $o_event_base ) {
global $a_event_array;
global $a_client_array;
// socket_accept接受连接，生成一个新的socket，一个客户端连接socket
$r_connection_socket = socket_accept( $r_listen_socket );
$a_client_array[] = $r_connection_socket;
// 在这个客户端连接socket上添加 读事件
// 也就说 要从客户端连接上读取消息
$o_read_event = new Event( $o_event_base, $r_connection_socket, Event::READ | Event::PERSIST, 'read_callback', $o_event_base );
$o_read_event->add();
$a_event_array[ intval( $r_connection_socket ) ]['read'] = $o_read_event;
}
// 在$listen_socket上添加一个 读事件
// 为啥是读事件？
// 因为$listen_socket上发生事件就是：客户端建立连接
// 所以，应该是读事件
$o_event = new Event( $o_event_base, $r_listen_socket, Event::READ | Event::PERSIST, 'accept_callback', $o_event_base );
$o_event->add();
//$a_event_array[] = $o_event;
$o_event_base->loop();是不是比原来好看了些许？这样，我想了想，今天先到这里，我得赶紧拉屎去了，真是憋不住了，马上就要到口了，但是我提两个问题你们先琢磨下：
- 维护那个发送数据自定义缓冲区太麻烦了，心智负担有点儿大
- 上述demo都是单进程的，你们有尝试过多进程与Libevent结合吗？
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2020-02-13，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读