Socket(套接字)一直是网络层的底层核心内容,也是 TCP/IP 以及 UDP 底层协议的实现通道。随着互联网信息时代的爆炸式发展,当代服务器的性能问题面临越来越大的挑战,著名的 C10K 问题(http://www.kegel.com/c10k.html)也随之出现。幸亏通过大牛们的不懈努力,区别于传统的 select/poll 的 epoll/kqueue 方式出现了,目前 linux2.6 以上的内核都普遍支持,这是 Socket 领域一项巨大的进步,不仅解决了 C10K 问题,也渐渐成为了当代互联网的底层核心技术。libevent 库就是其中一个比较出彩的项目(现在非常多的开源项目都有用到,包括 Memcached),感兴趣的朋友可以研究一下。

由于网络上系统介绍这个部分的文章并不多,而涉及 PHP 的就更少了,所以石头君在这里希望通过《Socket深度探究4PHP》这个系列给对这个领域感兴趣的读者们一定的帮助,也希望大家能和我一起对这个问题进行更深入的探讨。首先,解释一下目前 Socket 领域比较易于混淆的概念有:阻塞/非阻塞、同步/异步、多路复用等。

1、阻塞/非阻塞:这两个概念是针对 IO 过程中进程的状态来说的,阻塞 IO 是指调用结果返回之前,当前线程会被挂起;相反,非阻塞指在不能立刻得到结果之前,该函数不会阻塞当前线程,而会立刻返回。

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

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

感觉概念很抽象对吧,“一切答案在于现场”,下面让我们从三种经典的 PHP Socket IO 模型实例来对以上的概念再做一次分析:

1、使用 accept 阻塞的古老模型:属于同步阻塞 IO 模型,代码如下:

socket_server.php

  1. <?php
  2. /**
  3.  * SocketServer Class
  4.  * By James.Huang <shagoo#gmail.com>
  5. **/
  6. set_time_limit(0);
  7. class SocketServer
  8. {
  9.     private static $socket;
  10.     function SocketServer($port)
  11.     {
  12.         global $errno, $errstr;
  13.         if ($port < 1024) {
  14.             die("Port must be a number which bigger than 1024/n");
  15.         }
  16.         $socket = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
  17.         if (!$socket) die("$errstr ($errno)");
  18. //      stream_set_timeout($socket, -1); // 保证服务端 socket 不会超时,似乎没用:)
  19.         while ($conn = stream_socket_accept($socket, -1)) { // 这样设置不超时才油用
  20.             static $id = 0;
  21.             static $ct = 0;
  22.             $ct_last = $ct;
  23.             $ct_data = '';
  24.             $buffer = '';
  25.             $id++; // increase on each accept
  26.             echo "Client $id come./n";
  27.             while (!preg_match('//r?/n/', $buffer)) { // 没有读到结束符,继续读
  28. //              if (feof($conn)) break; // 防止 popen 和 fread 的 bug 导致的死循环
  29.                 $buffer = fread($conn, 1024);
  30.                 echo 'R'; // 打印读的次数
  31.                 $ct += strlen($buffer);
  32.                 $ct_data .= preg_replace('//r?/n/', '', $buffer);
  33.             }
  34.             $ct_size = ($ct - $ct_last) * 8;
  35.             echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
  36.             fwrite($conn, "Received $ct_size byte data./r/n");
  37.             fclose($conn);
  38.         }
  39.         fclose($socket);
  40.     }
  41. }
  42. new SocketServer(2000);

socket_client.php

  1. <?php
  2. /**
  3.  * Socket Test Client
  4.  * By James.Huang <shagoo#gmail.com>
  5. **/
  6. function debug ($msg)
  7. {
  8. //  echo $msg;
  9.     error_log($msg, 3, '/tmp/socket.log');
  10. }
  11. if ($argv[1]) {
  12.     $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
  13. //  stream_set_blocking($socket_client, 0);
  14. //  stream_set_timeout($socket_client, 0, 100000);
  15.     if (!$socket_client) {
  16.         die("$errstr ($errno)");
  17.     } else {
  18.         $msg = trim($argv[1]);
  19.         for ($i = 0; $i < 10; $i++) {
  20.             $res = fwrite($socket_client, "$msg($i)");
  21.             usleep(100000);
  22.             echo 'W'; // 打印写的次数
  23. //          debug(fread($socket_client, 1024)); // 将产生死锁,因为 fread 在阻塞模式下未读到数据时将等待
  24.         }
  25.         fwrite($socket_client, "/r/n"); // 传输结束符
  26.         debug(fread($socket_client, 1024));
  27.         fclose($socket_client);
  28.     }
  29. }
  30. else {
  31. //  $phArr = array();
  32. //  for ($i = 0; $i < 10; $i++) {
  33. //      $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
  34. //  }
  35. //  foreach ($phArr as $ph) {
  36. //      pclose($ph);
  37. //  }
  38.     for ($i = 0; $i < 10; $i++) {
  39.         system("php ".__FILE__." '{$i}:test'");
  40.     }
  41. }

首先,解释一下以上的代码逻辑:客户端 socket_client.php 循环发送数据,最后发送结束符;服务端 socket_server.php 使用 accept 阻塞方式接收 socket 连接,然后循环接收数据,直到收到结束符,返回结果数据(接收到的字节数)。虽然逻辑很简单,但是其中有几种情况很值得分析一下:

A> 默认情况下,运行 php socket_client.php test,客户端打出 10 个 W,服务端打出若干个 R 后面是接收到的数据,/tmp/socket.log 记录下服务端返回的接收结果数据。这种情况很容易理解,不再赘述。然后,使用 telnet 命令同时打开多个客户端,你会发现服务器一个时间只处理一个客户端,其他需要在后面“排队”;这就是阻塞 IO 的特点,这种模式的弱点很明显,效率极低。

B> 只打开 socket_client.php 第 26 行的注释代码,再次运行 php socket_client.php test 客户端打出一个 W,服务端也打出一个 R,之后两个程序都卡住了。这是为什么呢,分析逻辑后你会发现,这是由于客户端在未发送结束符之前就向服务端要返回数据;而服务端由于未收到结束符,也在向客户端要结束符,造成死锁。而之所以只打出一个 W 和 R,是因为 fread 默认是阻塞的。要解决这个死锁,必须打开 socket_client.php 第 16 行的注释代码,给 socket 设置一个 0.1 秒的超时,再次运行你会发现隔 0.1 秒出现一个 W 和 R 之后正常结束,服务端返回的接收结果数据也正常记录了。可见 fread 缺省是阻塞的,我们在编程的时候要特别注意,如果没有设置超时,就很容易会出现死锁。

C> 只打开 15 行注释,运行 php socket_client.php test,结果基本和情况 A 相同,唯一不同的是 /tmp/socket.log 没有记录下返回数据。这里可以看出客户端运行在阻塞和非阻塞模式的区别,当然在客户端不在乎接受结果的情况下,可以使用非阻塞模式来获得最大效率。

D> 运行 php socket_client.php 是连续运行 10 次上面的逻辑,这个没什么问题;但是很奇怪的是如果你使用 35 - 41 行的代码,用 popen 同时开启 10 个进程来运行,就会造成服务器端的死循环,十分怪异!后来经调查发现只要是用 popen 打开的进程创建的连接会导致 fread 或者 socket_read 出错直接返回空字串,从而导致死循环,查阅 PHP 源代码后发现 PHP 的 popen 和 fread 函数已经完全不是 C 原生的了,里面都插入了大量的 php_stream_* 实现逻辑,初步估计是其中的某个 bug 导致的 Socket 连接中断所导致的,解决方法就是打开 socket_server.php 中 31 行的代码,如果连接中断则跳出循环,但是这样一来就会有很多数据丢失了,这个问题需要特别注意!

2、使用 select/poll 的同步模型:属于同步非阻塞 IO 模型,代码如下:

select_server.php

  1. <?php
  2. /**
  3.  * SelectSocketServer Class
  4.  * By James.Huang <shagoo#gmail.com>
  5. **/
  6. set_time_limit(0);
  7. class SelectSocketServer
  8. {
  9.     private static $socket;
  10.     private static $timeout = 60;
  11.     private static $maxconns = 1024;
  12.     private static $connections = array();
  13.     function SelectSocketServer($port)
  14.     {
  15.         global $errno, $errstr;
  16.         if ($port < 1024) {
  17.             die("Port must be a number which bigger than 1024/n");
  18.         }
  19.         $socket = socket_create_listen($port);
  20.         if (!$socket) die("Listen $port failed");
  21.         socket_set_nonblock($socket); // 非阻塞
  22.         while (true)
  23.         {
  24.             $readfds = array_merge(self::$connections, array($socket));
  25.             $writefds = array();
  26.             // 选择一个连接,获取读、写连接通道
  27.             if (socket_select($readfds, $writefds, $e = null, $t = self::$timeout))
  28.             {
  29.                 // 如果是当前服务端的监听连接
  30.                 if (in_array($socket, $readfds)) {
  31.                     // 接受客户端连接
  32.                     $newconn = socket_accept($socket);
  33.                     $i = (int) $newconn;
  34.                     $reject = '';
  35.                     if (count(self::$connections) >= self::$maxconns) {
  36.                         $reject = "Server full, Try again later./n";
  37.                     }
  38.                     // 将当前客户端连接放入 socket_select 选择
  39.                     self::$connections[$i] = $newconn;
  40.                     // 输入的连接资源缓存容器
  41.                     $writefds[$i] = $newconn;
  42.                     // 连接不正常
  43.                     if ($reject) {
  44.                         socket_write($writefds[$i], $reject);
  45.                         unset($writefds[$i]);
  46.                         self::close($i);
  47.                     } else {
  48.                         echo "Client $i come./n";
  49.                     }
  50.                     // remove the listening socket from the clients-with-data array
  51.                     $key = array_search($socket, $readfds);
  52.                     unset($readfds[$key]);
  53.                 }
  54.                 // 轮循读通道
  55.                 foreach ($readfds as $rfd) {
  56.                     // 客户端连接
  57.                     $i = (int) $rfd;
  58.                     // 从通道读取
  59.                     $line = @socket_read($rfd, 2048, PHP_NORMAL_READ);
  60.                     if ($line === false) {
  61.                         // 读取不到内容,结束连接
  62.                         echo "Connection closed on socket $i./n";
  63.                         self::close($i);
  64.                         continue;
  65.                     }
  66.                     $tmp = substr($line, -1);
  67.                     if ($tmp != "/r" && $tmp != "/n") {
  68.                         // 等待更多数据
  69.                         continue;
  70.                     }
  71.                     // 处理逻辑
  72.                     $line = trim($line);
  73.                     if ($line == "quit") {
  74.                         echo "Client $i quit./n";
  75.                         self::close($i);
  76.                         break;
  77.                     }
  78.                     if ($line) {
  79.                         echo "Client $i >>" . $line . "/n";
  80.                     }
  81.                 }
  82.                 // 轮循写通道
  83.                 foreach ($writefds as $wfd) {
  84.                     $i = (int) $wfd;
  85.                     $w = socket_write($wfd, "Welcome Client $i!/n");
  86.                 }
  87.             }
  88.         }
  89.     }
  90.     function close ($i)
  91.     {
  92.         socket_shutdown(self::$connections[$i]);
  93.         socket_close(self::$connections[$i]);
  94.         unset(self::$connections[$i]);
  95.     }
  96. }
  97. new SelectSocketServer(2000);

select_client.php

  1. <?php
  2. /**
  3.  * SelectSocket Test Client
  4.  * By James.Huang <shagoo#gmail.com>
  5. **/
  6. function debug ($msg)
  7. {
  8. //  echo $msg;
  9.     error_log($msg, 3, '/tmp/socket.log');
  10. }
  11. if ($argv[1]) {
  12.     $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
  13. //  stream_set_timeout($socket_client, 0, 100000);
  14.     if (!$socket_client) {
  15.         die("$errstr ($errno)");
  16.     } else {
  17.         $msg = trim($argv[1]);
  18.         for ($i = 0; $i < 10; $i++) {
  19.             $res = fwrite($socket_client, "$msg($i)/n");
  20.             usleep(100000);
  21. //          debug(fread($socket_client, 1024)); // 将产生死锁,因为 fread 在阻塞模式下未读到数据时将等待
  22.         }
  23.         fwrite($socket_client, "quit/n"); // add end token
  24.         debug(fread($socket_client, 1024));
  25.         fclose($socket_client);
  26.     }
  27. }
  28. else {
  29.     $phArr = array();
  30.     for ($i = 0; $i < 10; $i++) {
  31.         $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
  32.     }
  33.     foreach ($phArr as $ph) {
  34.         pclose($ph);
  35.     }
  36. //  for ($i = 0; $i < 10; $i++) {
  37. //      system("php ".__FILE__." '{$i}:test'");
  38. //  }
  39. }

以上代码的逻辑也很简单,select_server.php 实现了一个类似聊天室的功能,你可以使用 telnet 工具登录上去,和其他用户文字聊天,也可以键入“quit”命令离开;而 select_client.php 则模拟了一个登录用户连续发 10 条信息,然后退出。这里也分析两个问题:

A> 这里如果我们执行 php select_client.php 程序将会同时打开 10 个连接,同时进行模拟登录用户操作;观察服务端打印的数据你会发现服务端确实是在同时处理这些连接,这就是多路复用实现的非阻塞 IO 模型,当然这个模型并没有真正的实现异步,因为最终服务端程序还是要去通道里面读取数据,得到结果后同步返回给客户端。如果这次你也使用 telnet 命令同时打开多个客户端,你会发现服务端可以同时处理这些连接,这就是非阻塞 IO,当然比古老的阻塞 IO 效率要高多了,但是这种模式还是有局限的,继续看下去你就会发现了~

B> 我在 select_server.php 中设置了几个参数,大家可以调整试试:
$timeout :表示的是 select 的超时时间,这个一般来说不要太短,否则会导致 CPU 负载过高。
$maxconns :表示的是最大连接数,客户端超过这个数的话,服务器会拒绝接收。这里要提到的一点是,由于 select 是通过句柄来读写的,所以会受到系统默认参数 __FD_SETSIZE 的限制,一般默认值为 1024,修改的话需要重新编译内核;另外通过测试发现 select 模式的性能会随着连接数的增大而线性便差(详情见《Socket深度探究4PHP(二)》),这也就是 select 模式最大的问题所在,所以如果是超高并发服务器建议使用下一种模式。

3、使用 epoll/kqueue 的异步模型:属于异步阻塞/非阻塞 IO 模型,代码如下:

epoll_server.php

  1. <?php
  2. /**
  3.  * EpollSocketServer Class (use libevent)
  4.  * By James.Huang <shagoo#gmail.com>
  5.  *
  6.  * Defined constants:
  7.  *
  8.  * EV_TIMEOUT (integer)
  9.  * EV_READ (integer)
  10.  * EV_WRITE (integer)
  11.  * EV_SIGNAL (integer)
  12.  * EV_PERSIST (integer)
  13.  * EVLOOP_NONBLOCK (integer)
  14.  * EVLOOP_ONCE (integer)
  15. **/
  16. set_time_limit(0);
  17. class EpollSocketServer
  18. {
  19.     private static $socket;
  20.     private static $connections;
  21.     private static $buffers;
  22.     function EpollSocketServer ($port)
  23.     {
  24.         global $errno, $errstr;
  25.         if (!extension_loaded('libevent')) {
  26.             die("Please install libevent extension firstly/n");
  27.         }
  28.         if ($port < 1024) {
  29.             die("Port must be a number which bigger than 1024/n");
  30.         }
  31.         $socket_server = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
  32.         if (!$socket_server) die("$errstr ($errno)");
  33.         stream_set_blocking($socket_server, 0); // 非阻塞
  34.         $base = event_base_new();
  35.         $event = event_new();
  36.         event_set($event, $socket_server, EV_READ | EV_PERSIST, array(__CLASS__, 'ev_accept'), $base);
  37.         event_base_set($event, $base);
  38.         event_add($event);
  39.         event_base_loop($base);
  40.         self::$connections = array();
  41.         self::$buffers = array();
  42.     }
  43.     function ev_accept($socket, $flag, $base)
  44.     {
  45.         static $id = 0;
  46.         $connection = stream_socket_accept($socket);
  47.         stream_set_blocking($connection, 0);
  48.         $id++; // increase on each accept
  49.         $buffer = event_buffer_new($connection, array(__CLASS__, 'ev_read'), array(__CLASS__, 'ev_write'), array(__CLASS__, 'ev_error'), $id);
  50.         event_buffer_base_set($buffer, $base);
  51.         event_buffer_timeout_set($buffer, 30, 30);
  52.         event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
  53.         event_buffer_priority_set($buffer, 10);
  54.         event_buffer_enable($buffer, EV_READ | EV_PERSIST);
  55.         // we need to save both buffer and connection outside
  56.         self::$connections[$id] = $connection;
  57.         self::$buffers[$id] = $buffer;
  58.     }
  59.     function ev_error($buffer, $error, $id)
  60.     {
  61.         event_buffer_disable(self::$buffers[$id], EV_READ | EV_WRITE);
  62.         event_buffer_free(self::$buffers[$id]);
  63.         fclose(self::$connections[$id]);
  64.         unset(self::$buffers[$id], self::$connections[$id]);
  65.     }
  66.     function ev_read($buffer, $id)
  67.     {
  68.         static $ct = 0;
  69.         $ct_last = $ct;
  70.         $ct_data = '';
  71.         while ($read = event_buffer_read($buffer, 1024)) {
  72.             $ct += strlen($read);
  73.             $ct_data .= $read;
  74.         }
  75.         $ct_size = ($ct - $ct_last) * 8;
  76.         echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
  77.         event_buffer_write($buffer, "Received $ct_size byte data./r/n");
  78.     }
  79.     function ev_write($buffer, $id)
  80.     {
  81.         echo "[$id] " . __METHOD__ . "/n";
  82.     }
  83. }
  84. new EpollSocketServer(2000);

epoll_client.php

  1. <?php
  2. /**
  3.  * EpollSocket Test Client
  4.  * By James.Huang <shagoo#gmail.com>
  5. **/
  6. function debug ($msg)
  7. {
  8. //  echo $msg;
  9.     error_log($msg, 3, '/tmp/socket.log');
  10. }
  11. if ($argv[1]) {
  12.     $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
  13. //  stream_set_blocking($socket_client, 0);
  14.     if (!$socket_client) {
  15.         die("$errstr ($errno)");
  16.     } else {
  17.         $msg = trim($argv[1]);
  18.         for ($i = 0; $i < 10; $i++) {
  19.             $res = fwrite($socket_client, "$msg($i)");
  20.             usleep(100000);
  21.             debug(fread($socket_client, 1024));
  22.         }
  23.         fclose($socket_client);
  24.     }
  25. }
  26. else {
  27.     $phArr = array();
  28.     for ($i = 0; $i < 10; $i++) {
  29.         $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
  30.     }
  31.     foreach ($phArr as $ph) {
  32.         pclose($ph);
  33.     }
  34. //  for ($i = 0; $i < 10; $i++) {
  35. //      system("php ".__FILE__." '{$i}:test'");
  36. //  }
  37. }

先说一下,以上的例子是基于 PHP 的 libevent 扩展实现的,需要运行的话要先安装此扩展,参考:http://pecl.php.net/package/libevent。

这个例子做的事情和前面介绍的第一个模型一样,epoll_server.php 实现的服务端也是接受客户端数据,然后返回结果(接收到的字节数)。但是,当你运行 php epoll_client.php 的时候你会发现服务端打印出来的结果和 accept 阻塞模型就大不一样了,当然运行效率也有极大的提升,这是为什么呢?接下来就介绍一下 epoll/kqueue 模型:在介绍 select 模式的时候我们提到了这种模式的局限,而 epoll 就是为了解决 poll 的这两个缺陷而生的。首先,epoll 模式基本没有限制(参考 cat /proc/sys/fs/file-max 默认就达到 300K,很令人兴奋吧,其实这也就是所谓基于 epoll 的 Erlang 服务端可以同时处理这么多并发连接的根本原因,不过现在 PHP 理论上也可以做到了,呵呵);另外,epoll 模式的性能也不会像 select 模式那样随着连接数的增大而变差,测试发现性能还是很稳定的(下篇会有详细介绍)。

epoll 工作有两种模式 LT(level triggered) 和 ET(edge-triggered),前者是缺省模式,同时支持阻塞和非阻塞 IO 模式,虽然性能比后者差点,但是比较稳定,一般来说在实际运用中,我们都是用这种模式(ET 模式和 WinSock 都是纯异步非阻塞模型)。而另外一点要说的是 libevent 是在编译阶段选择系统的 I/O demultiplex 机制的,不支持在运行阶段根据配置再次选择,所以我们在这里也就不细讨论 libevent 的实现的细节了,如果朋友有兴趣进一步了解的话,请参考:http://monkey.org/~provos/libevent/。

到这里,第一部分的内容结束了,相信大家已经了解了 Socket 编程的几个重点概念和一些实战技巧,在下一篇《Socket深度探究4PHP(二) 》我将会对 select/poll/epoll/kqueue 几种模式做一下深入的介绍和对比,另外也会涉及到两种重要的 I/O 多路复用模式:Reactor 和 Proactor 模式。

Socket 深度探索 4 PHP (一)
标签: