关于php的socket初探

天空一直巨响 · 发表于 2015-09-20 11:07:46

本帖最后由天空一直巨响于 2015-09-20 11:35 编辑

做了好几年的flash的aser，技术一直停留在flash这个狭小的容器内，最近转向php，才慢慢体会到之前充实技术开发前的那段极度渴望去学习新知识的那种动力，可能在大多数的人眼里，php是专为web而生，做wep app是一流的快速好用，并极易简单和容易上手，做起web所需的各项功能可能是分分钟搞定，然后做socket了？可能就觉得不屑一顾了，好像是不务正业，用了自己的短板去做不擅长的事情。其实不然，往往越不起眼的东西却能带给你惊喜，废话不多说，直奔主题。关于php的socket的探究。关于socket，有很多不同的定义，一般意义上的说明，都是将socket称作"套接字"，也就是对底层tcp/ip的一种封装，所以要想真正理解socket的原理，就得去深入理解tcp/ip实现网络通信的机制，这属于计算机网络这块的东西，与实际的编程没有太大的关联，这里就不深入说明了，有兴趣的可以问度娘。而从程序的角度来看，这样的理解更合适，
socket起源于Unix，而Unix/Linux基本哲学之一就是“一切皆文件”，都可以用“打开open –> 读写write/read –> 关闭close”模式来操作。我的理解就是Socket就是该模式的一个实现，socket即是一种特殊的文件，一些socket函数就是对其进行的操作（读/写IO、打开、关闭）
既然Unix/Linux是将socket以一种io的形式来编程实现，那对于socket的研究必然有几个概念要理解：
1、阻塞/非阻塞：这两个概念是针对 IO 过程中进程的状态来说的，阻塞 IO 是指调用结果返回之前，当前线程会被挂起；相反，非阻塞指在不能立刻得到结果之前，该函数不会阻塞当前线程，而会立刻返回。

2、同步/异步：这两个概念是针对调用如果返回结果来说的，所谓同步，就是在发出一个功能调用时，在没有得到结果之前，该调用就不返回；相反，当一个异步过程调用发出后，调用者不能立刻得到结果，实际处理这个调用的部件在完成后，通过状态、通知和回调来通知调用者。

3、多路复用（IO/Multiplexing）：为了提高数据信息在网络通信线路中传输的效率，在一条物理通信线路上建立多条逻辑通信信道，同时传输若干路信号的技术就叫做多路复用技术。对于 Socket 来说，应该说能同时处理多个连接的模型都应该被称为多路复用，目前比较常用的有 select/poll/epoll/kqueue 这些 IO 模型（目前也有像 Apache 这种每个连接用单独的进程/线程来处理的 IO 模型，但是效率相对比较差，也很容易出问题，所以暂时不做介绍了）。在这些多路复用的模式中，异步阻塞/非阻塞模式的扩展性和性能最好。

引入阻塞/非阻塞,、同步/异步比喻很形象的一段话：
有A，B，C，D四个人在钓鱼：
A用的是最老式的鱼竿，所以呢，得一直守着，等到鱼上钩了再拉杆；
B的鱼竿有个功能，能够显示是否有鱼上钩，所以呢，B就和旁边的MM聊天，隔会再看看有没有鱼上钩，有的话就迅速拉杆；
C用的鱼竿和B差不多，但他想了一个好办法，就是同时放好几根鱼竿，然后守在旁边，一旦有显示说鱼上钩了，它就将对应的鱼竿拉起来；
D是个有钱人，干脆雇了一个人帮他钓鱼，一旦那个人把鱼钓上来了，就给D发个短信

按照我的理解，阻塞/非阻塞是对于进程来说，而同步/异步是对于函数（系统调用）来说，对于程序来说，肯定是异步非阻塞最好

接下来看以不同io模型来实现一个socket（io） server的功能
1、使用 accept 阻塞的古老模型：属于同步阻塞 IO 模型，代码如下：
socket_server.php

<?php
/**
* SocketServer Class
* By James.Huang <shagoo#gmail.com>
**/
set_time_limit(0);
class SocketServer
{
private static $socket;
function SocketServer($port)
{
global $errno, $errstr;
if ($port < 1024) {
die("Port must be a number which bigger than 1024/n");
}
$socket = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
if (!$socket) die("$errstr ($errno)");
// stream_set_timeout($socket, -1); // 保证服务端 socket 不会超时，似乎没用：）
while ($conn = stream_socket_accept($socket, -1)) { // 这样设置不超时才油用
static $id = 0;
static $ct = 0;
$ct_last = $ct;
$ct_data = '';
$buffer = '';
$id++; // increase on each accept
echo "Client $id come./n";
while (!preg_match('//r?/n/', $buffer)) { // 没有读到结束符，继续读
// if (feof($conn)) break; // 防止 popen 和 fread 的 bug 导致的死循环
$buffer = fread($conn, 1024);
echo 'R'; // 打印读的次数
$ct += strlen($buffer);
$ct_data .= preg_replace('//r?/n/', '', $buffer);
}
$ct_size = ($ct - $ct_last) * 8;
echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
fwrite($conn, "Received $ct_size byte data./r/n");
fclose($conn);
}
fclose($socket);
}
}
new SocketServer(2000);

复制代码

天空一直巨响 · 发表于 2015-09-20 11:35:33

2、使用 select/poll 的同步模型：属于同步非阻塞 IO 模型，代码如下：
<?php
/**
* SelectSocketServer Class
* By James.Huang <shagoo#gmail.com>
**/
set_time_limit(0);
class SelectSocketServer
{
private static $socket;
private static $timeout = 60;
private static $maxconns = 1024;
private static $connections = array();
function SelectSocketServer($port)
{
global $errno, $errstr;
if ($port < 1024) {
die("Port must be a number which bigger than 1024/n");
}

$socket = socket_create_listen($port);
if (!$socket) die("Listen $port failed");

socket_set_nonblock($socket); // 非阻塞

while (true)
{
$readfds = array_merge(self::$connections, array($socket));
$writefds = array();

// 选择一个连接，获取读、写连接通道
if (socket_select($readfds, $writefds, $e = null, $t = self::$timeout))
{
// 如果是当前服务端的监听连接
if (in_array($socket, $readfds)) {
// 接受客户端连接
$newconn = socket_accept($socket);
$i = (int) $newconn;
$reject = '';
if (count(self::$connections) >= self::$maxconns) {
$reject = "Server full, Try again later./n";
}
// 将当前客户端连接放入 socket_select 选择
self::$connections[$i] = $newconn;
// 输入的连接资源缓存容器
$writefds[$i] = $newconn;
// 连接不正常
if ($reject) {
socket_write($writefds[$i], $reject);
unset($writefds[$i]);
self::close($i);
} else {
echo "Client $i come./n";
}
// remove the listening socket from the clients-with-data array
$key = array_search($socket, $readfds);
unset($readfds[$key]);
}

// 轮循读通道
foreach ($readfds as $rfd) {
// 客户端连接
$i = (int) $rfd;
// 从通道读取
$line = @socket_read($rfd, 2048, PHP_NORMAL_READ);
if ($line === false) {
// 读取不到内容，结束连接
echo "Connection closed on socket $i./n";
self::close($i);
continue;
}
$tmp = substr($line, -1);
if ($tmp != "/r" && $tmp != "/n") {
// 等待更多数据
continue;
}
// 处理逻辑
$line = trim($line);
if ($line == "quit") {
echo "Client $i quit./n";
self::close($i);
break;
}
if ($line) {
echo "Client $i >>" . $line . "/n";
}
}

// 轮循写通道
foreach ($writefds as $wfd) {
$i = (int) $wfd;
$w = socket_write($wfd, "Welcome Client $i!/n");
}
}
}
}

function close ($i)
{
socket_shutdown(self::$connections[$i]);
socket_close(self::$connections[$i]);
unset(self::$connections[$i]);
}
}
new SelectSocketServer(2000);

3、使用 epoll/kqueue 的异步模型：属于异步阻塞/非阻塞 IO 模型，代码如下：
<?php
/**
* EpollSocketServer Class (use libevent)
* By James.Huang <shagoo#gmail.com>
*
* Defined constants:
*
* EV_TIMEOUT (integer)
* EV_READ (integer)
* EV_WRITE (integer)
* EV_SIGNAL (integer)
* EV_PERSIST (integer)
* EVLOOP_NONBLOCK (integer)
* EVLOOP_ONCE (integer)
**/
set_time_limit(0);
class EpollSocketServer
{
private static $socket;
private static $connections;
private static $buffers;

function EpollSocketServer ($port)
{
global $errno, $errstr;

if (!extension_loaded('libevent')) {
die("Please install libevent extension firstly/n");
}

if ($port < 1024) {
die("Port must be a number which bigger than 1024/n");
}

$socket_server = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
if (!$socket_server) die("$errstr ($errno)");

stream_set_blocking($socket_server, 0); // 非阻塞

$base = event_base_new();
$event = event_new();
event_set($event, $socket_server, EV_READ | EV_PERSIST, array(__CLASS__, 'ev_accept'), $base);
event_base_set($event, $base);
event_add($event);
event_base_loop($base);

self::$connections = array();
self::$buffers = array();
}

function ev_accept($socket, $flag, $base)
{
static $id = 0;

$connection = stream_socket_accept($socket);
stream_set_blocking($connection, 0);

$id++; // increase on each accept

$buffer = event_buffer_new($connection, array(__CLASS__, 'ev_read'), array(__CLASS__, 'ev_write'), array(__CLASS__, 'ev_error'), $id);
event_buffer_base_set($buffer, $base);
event_buffer_timeout_set($buffer, 30, 30);
event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
event_buffer_priority_set($buffer, 10);
event_buffer_enable($buffer, EV_READ | EV_PERSIST);

// we need to save both buffer and connection outside
self::$connections[$id] = $connection;
self::$buffers[$id] = $buffer;
}

function ev_error($buffer, $error, $id)
{
event_buffer_disable(self::$buffers[$id], EV_READ | EV_WRITE);
event_buffer_free(self::$buffers[$id]);
fclose(self::$connections[$id]);
unset(self::$buffers[$id], self::$connections[$id]);
}

function ev_read($buffer, $id)
{
static $ct = 0;
$ct_last = $ct;
$ct_data = '';
while ($read = event_buffer_read($buffer, 1024)) {
$ct += strlen($read);
$ct_data .= $read;
}
$ct_size = ($ct - $ct_last) * 8;
echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
event_buffer_write($buffer, "Received $ct_size byte data./r/n");
}

function ev_write($buffer, $id)
{
echo "[$id] " . __METHOD__ . "/n";
}
}
new EpollSocketServer(2000);
先说一下，以上的例子是基于 PHP 的 libevent 扩展实现的，需要运行的话要先安装此扩展，参考：http://pecl.php.net/package/libevent。

epoll/kqueue 的例子最好是能在linux环境下运行，因为windows下的php不支持多epoll模型，也不支持多进程。
以上就是我对php的初探，3中io模型的代码确定有效，可以再次基础上做server或者gateway服务器的开发，第一次发贴，写的匆忙，还有很多东西没有说明，不足之处望请见谅，以后持续编辑此贴！希望能和大家沟通交流！

参考资料：

http://blog.csdn.net/shagoo/article/details/6396089

http://www.cnblogs.com/skynet/archive/2010/12/12/1903949.html

清泛网 · 发表于 2015-09-21 09:14:52

总结的很好，例子经典，感谢楼主分享{:geiliv2:}