WebSocket服务端开发(九)-WebSocketServer类事件

本文介绍WebSocketServer类中的事件。

WebSocketServer类支持以下事件:
onstarted:
服务器启动后触发,socket_listen成功执行后,服务器进入启动状态

onconnected:
与客户端建立连接完成后触发,此时连接已放入socket连接池中

onUpgradePartReceive:
当收到部分WebSocket握手时触发,仅当收到的握手包不完整时会触发该事件

onHandShakeFailure:
在握手失败后触发

onHandShakeSuccess:
握手成功后触发

ondisconnected:
关闭客户端连接后触发,此时还未将连接从socket连接池中移除,当disconnect函数的第二个参数slient为true时不触发该事件

onAfterRemoveSocket:
在移除socket连接后触发

onafterhealthcheck:
在健康检查完成后触发,默认该事件会断开不健康的连接

onerror:
遇到socket错误时触发

onshutdown:
服务器关闭时触发

  function onstarted($serverSocket) {
    if ($this->debug) {
      printf('Server start at %s', date('Y-m-d H:i:s') . "\n");
    }
  }

  function onconnected($socket) {
    if ($this->debug) {
      printf('Socket connect at %s-%s', date('Y-m-d H:i:s'), $socket . "\n");
    }
  }

  function onUpgradePartReceive($socketId) {
    if ($this->debug) {
      $buffer = $this->socketListMap[$socketId]['buffer'];
      printf('Receive Upgrade Part at %s-%s%s(%d bytes)' . "\n", date('Y-m-d H:i:s'), $socketId . "\n", $buffer, strlen($buffer));
    }
  }

  function onHandShakeFailure($socketId) {
    if ($this->debug) {
      printf('HandShake Failed at %s-%s', date('Y-m-d H:i:s'), $socketId . "\n");
    }
  }

  function onHandShakeSuccess($socketId) {
    if ($this->debug) {
      printf('HandShake Success at %s-%s', date('Y-m-d H:i:s'), $socketId . "\n");
    }
  }

  function ondisconnected($socketId) {
    if ($this->debug) {
      printf('Socket disconnect at %s-%s', date('Y-m-d H:i:s'), $socketId . "\n");
    }
  }

  function onAfterRemoveSocket($socketId) {
    if ($this->debug) {
      printf('[onAfterRemoveSocket]remove:' . $socketId . ',left:' . implode('|', array_keys($this->socketListMap)) . "\n");
    }
  }

  function onafterhealthcheck($unhealthyList) {
    foreach ($unhealthyList as $socketId) {
      $this->disconnect($socketId);
    }
  }

  function onerror($errCode, $socketId) {
    switch ($errCode) {
    case 10053:
      $this->disconnect($socketId);
      break;
    default:
      if ($this->debug) {
        echo 'Socket Error:' . $errorCode . "\n";
      }
      break;
    }
  }

  function onshutdown() {
    if ($this->debug) {
      printf('Server shutdown!');
    }
  }

WebSocket服务端开发(八)-WebSocketServer类错误处理

本文介绍WebSocketServer类错误处理相关方法。

错误信息显示

  //获取最后一次socket的错误码
  function getLastErrCode($socketId = null) {
    if (is_null($socketId)) {
      $socket = $this->serverSocket;
    } else {
      $socket = $this->socketListMap[$socketId]['socket'];
    }
    return socket_last_error($socket);
  }

  //通过错误码查找错误详情
  function getLastErrMsg($socketId = null, $errCode = null) {
    if (!is_numeric($errCode)) {
      $errCode = $this->getLastErrCode($socketId);
    }
    return '[' . $errCode . ']' . socket_strerror($errCode) . "\n";
  }

当客户端直接断开连接时(没有发送关闭帧),Windows下会出现socket错误10053,此时服务器可以直接断开此连接。在Linux服务器下,测试发现如果客户端断开连接,并不会出现10053错误(可能是服务器环境的问题),这时如果再收到连接会发现,已经断开的连接会造成死循环:
socket_select选中了断开的连接,然后从socket中读数据,Windows下调用socket_recv会返回false,而Linux下不会报错,这样会导致该连接不会从连接池中移除,然后会在下一次循环连接池时继续被选中,不停的循环。
这里解决方法是,对于有10053错误的,直接断开连接,不会出现任何问题。在接收数据后,如果数据长度大于0,则置该连接的errorCnt为0,否则errorCnt加一。判断errorCnt是否大于3次,如果大于3次,则断开该socket连接。也就是说,如果连续收到3次空内容的数据,就会断开连接。最后使用健康检查防止某些情况下没有移除连接的问题。

注意:该类只处理10053的socket错误,其他类型没有实现,可以通过覆盖onerror自定义自己的错误处理方式。

WebSocket服务端开发(七)-WebSocketServer类健康检查函数

本文介绍WebSocketServer类健康检查函数。

函数实现如下:

  function healthCheck() {
    //获取当前时间
    $now = time();

    //记录最后健康检查时间
    $this->lastHealthCheck = $now;

    //初始化不健康的连接列表
    $unhealthyList = array();

    //循环连接池
    foreach ($this->socketListMap as $socketId => $session) {
      //找出最后通信时间超过超时时间(目前超时时间与健康检查时间相同)
      if ($now - $session['lastCommuicate'] > $this->healthCheckInterval) {
        array_push($unhealthyList, $socketId);
      }
    }
    if ($this->debug) {
      echo 'Unhealthy socket:' . implode(',', $unhealthyList) . "\n";
    }

    //健康检查回调,默认的行为是直接断开连接
    //可以根据自己的需求改进,如发送ping帧探测等,如果仍无响应再断开连接
    $this->onafterhealthcheck($unhealthyList);
  }

注意:由于该类只是简单的实现了WebSocket协议,没有使用多线程处理,故健康检查只有在处理完一次连接后才有可能执行。如果设置健康检查时间间隔为10分钟,10分钟内收到任何数据,那么也不会进行健康检查,直到收到一个连接后才会进行健康检查。
收发数据都会更新最后通信时间的值,如果希望只在服务器收到信息时更新最后通信时间,可以在recv和send函数中修改。
建议定时从服务器发送心跳包以维持连接。

WebSocket服务端开发(六)-WebSocketServer类帧解析相关函数介绍

本文介绍WebSocketServer类解析帧相关的函数。

isMasked函数:

  //判断是否是经过掩码处理的帧
  function isMasked($byte) {
    return (ord($byte) & 0x80) > 0;
  }

getPayloadLen函数:

  //获取负载的长度
  //首字节如果小于126,则长度为首字节的值
  //首字节如果等于126,则长度为后面紧跟的两个字节表示的字节数
  //如果首字节是127,则长度为后面紧跟8字节表示的字节数
  function getPayloadLen($data) {
    $first = ord($data[0]) & 0x7F;
    $second = (ord($data[1]) << 8) + ord($data[2]);
    $third = (ord($data[3]) << 40) + (ord($data[4]) << 32) + (ord($data[5]) << 24) + (ord($data[6]) << 16) + (ord($data[7]) << 8) + ord($data[8]);
    if ($first < 126) {
      return $first;
    } else if ($first === 126) {
      return $second;
    } else {
      return ($second << 48) + $third;
    }
  }

getPayLoad函数:

  //获取负载的内容,根据帧的结构
  //第0字节为结束标记及帧类型
  //第1字节是否掩码标识位及负载长度的首字节
  //根据协议要求,浏览器发送的数据必须经过掩码处理,所以偏移量至少是1字节的帧首字节+4字节的掩码长度
  //根据帧长度的不同,表示长度的字节数可能为1、3或9
  //所以根据不同的情况截取不同长度的数据即为负载内容
  function getPayload($data, $len) {
    $offset = 5;
    if ($len < 126) {
      return substr($data, $offset + 1, $len);
    } else if ($len < 65536) {
      return substr($data, $offset + 3, $len);
    } else {
      return substr($data, $offset + 9, $len);
    }
  }

getMask函数:

  //获取掩码的值
  //仍然是根据不同的偏移位置截取
  function getMask($data, $len) {
    $offset = 1;
    if ($len < 126) {
      return substr($data, $offset + 1, 4);
    } else if ($len < 65536) {
      return substr($data, $offset + 3, 4);
    } else {
      return substr($data, $offset + 9, 4);
    }
  }

getFrameType函数:

  //获取帧类型
  function getFrameType($byte) {
    return ord($byte) & 0x0F;
  }

isFin函数:

  //判断是否为结束帧
  function isFin($byte) {
    return (ord($byte[0]) & 0x80) > 0;
  }

isControlFrame函数:

  //判断是否为控制帧,控制帧包含关闭帧、PING帧和PONG帧
  function isControlFrame($frameType) {
    return $frameType === self::FRAME_CLOSE || $frameType === self::FRAME_PING || $frameType === self::FRAME_PONG;
  }

parseBinaryFrame、parseTextFrame、parseRawFrame函数:

  //处理负载的掩码,将其还原
  function parseRawFrame($payload, $mask) {
    $payloadLen = strlen($payload);
    $dest = '';
    $maskArr = array();
    for ($i = 0; $i < 4; $i++) {
      $maskArr[$i] = ord($mask[$i]);
    }
    for ($i = 0; $i < $payloadLen; $i++) {
      $dest .= chr(ord($payload[$i]) ^ $maskArr[$i % 4]);
    }
    return $dest;
  }

  function parseTextFrame($payload, $mask) {
    return $this->parseRawFrame($payload, $mask);
  }

  function parseBinaryFrame($payload, $mask) {
    return $this->parseRawFrame($payload, $mask);
  }

closeFrame函数:

  //创建并发送关闭帧
  function closeFrame($socketId, $closeCode = 1000, $closeMsg = 'goodbye') {
    $closeCode = chr(intval($closeCode / 256)) . chr($closeCode % 256);
    $frame = $this->createFrame($closeCode . $closeMsg, self::FRAME_CLOSE);
    $this->socketSend($socketId, $frame);
    $this->disconnect($socketId);
  }

sendPing、sendPong函数:

  function sendPing($socketId, $data = 'ping') {
    $frame = $this->createFrame($data, self::FRAME_PING);
    $this->socketSend($socketId, $frame);
  }

  function sendPong($socketId, $data = 'pong') {
    $frame = $this->createFrame($data, self::FRAME_PONG);
    $this->socketSend($socketId, $frame);
  }

createFrame函数:

  //封装帧头的相关标识位、长度等信息
  function createFrame($data, $type, $fin = 0x01) {
    $dataLen = strlen($data);
    $frame = chr(($fin << 7) + $type);
    if ($dataLen < 126) {
      $frame .= chr($dataLen);
    } else if ($dataLen < 65536) {
      $frame .= chr(126);
      $frame .= chr(intval($dataLen / 256));
      $frame .= chr(intval($dataLen % 256));
    } else {
      $frame .= chr(127);
      $hexLen = str_pad(base_convert($dataLen, 10, 16), 16, '0', STR_PAD_LEFT);
      for ($i = 0; $i < 15; $i += 2) {
        $frame .= chr((intval($hexLen[$i], 16) << 8) + intval($hexLen[$i + 1], 16));
      }
    }
    $frame .= $data;
    return $frame;
  }

WebSocket服务端开发(五)-WebSocketServer类帧解析整体流程介绍

本文介绍WebSocketServer类的parseFrame函数流程。

parseFrame函数:

function parseFrame($data, $socketId) {
  //判断该帧是否是经过掩码处理
  $isMasked = $this->isMasked($data[1]);

  //如果未经掩码处理,则根据协议规定,需要断开连接
  if (!$isMasked) {
    //此处使用1002状态码,表示协议错误,发送关闭帧
    $this->closeFrame($socketId, 1002, 'There is no mask!');

    //断开连接
    $this->disconnect($socketId);
    return false;
  }
  //获取负载的长度字节数
  $payloadLen = $this->getPayloadLen(substr($data, 1, 9));

  //根据负载长度获取负载的全部数据
  $payload = $this->getPayload($data, $payloadLen);

  //获取掩码值
  $mask = $this->getMask($data, $payloadLen);

  //获取帧的类型
  $frameType = $this->getFrameType($data[0]);

  //处理帧
  switch ($frameType) {
  case self::FRAME_CONTINUE:
    //后续帧,需要拼接buffer
    $this->socketListMap[$socketId]['buffer'] .= $this->parseRawFrame($payload, $mask);
    break;
  case self::FRAME_TEXT:
    //文本帧,处理方式默认保持一致,均使用parseRawFrame处理,如果由特殊需求可以重写parseTextFrame函数
    $this->socketListMap[$socketId]['buffer'] = $this->parseTextFrame($payload, $mask);
    break;
  case self::FRAME_BIN:
    //二进制帧,处理方式默认保持一致,均使用parseRawFrame处理,如果由特殊需求可以重写parseBinaryFrame函数
    $this->socketListMap[$socketId]['buffer'] = $this->parseBinaryFrame($payload, $mask);
    break;
  case self::FRAME_CLOSE:
    //发送关闭帧(应答帧)
    $this->closeFrame($socketId);
    break;
  case self::FRAME_PING:
    //发送pong帧响应,浏览器目前不提供ping、pong帧的API,此处逻辑基本不会走到,只为实现协议内容
    $this->sendPong($socketId, $this->parseRawFrame($payload, $mask));
    break;
  case self::FRAME_PONG:
    //收到pong帧不进行任何处理(正常情况下不会收到,浏览器不会主动发送pong帧)
    break;
  default:
    //其他帧类型无法处理,直接断开连接,根据协议,此处使用1003状态码关闭连接更好
    $this->disconnect($socketId);
    break;
  }
  if ($this->debug) {
    //输出调试信息
    echo "isFin:" . ((ord($data[0]) & 0x80) >> 7) . "\n";
    echo "opCode:$frameType\n";
    echo "payLoad Length:$payloadLen\n";
    echo "Mask:$mask\n\n";
  }

  //如果是结束的数据帧,返回true,否则均为false
  //当返回true时,外层调用函数会继续将执行核心业务逻辑,读取缓冲区中的数据进行处理
  //如果是false,则不进行进一步的处理(控制帧及非结束帧都不会提交到业务层处理)
  return $this->isFin($data[0]) && !$this->isControlFrame($frameType);
}

WebSocket服务端开发(四)-WebSocketServer类socket系列封装函数介绍

本文介绍WebSocketServer类中处理socket的封装函数实现。

getSocketId函数:

function getSocketId($socket) {
  //socketId由socket中的地址和端口号拼接,这样可以保证socket的唯一性,又可以通过id快速读取保存socket的信息以及其附加的其他相关信息
  if (socket_getpeername($socket, $address, $port) === FALSE) {
    return false;
  }
  return $address . '_' . $port;
}

addSocket函数:

function addSocket($socket) {
  //将socket添加到已接受的socket连接池中
  array_push($this->socketList, $socket);
  $socketId = $this->getSocketId($socket);

  //以socketId为索引建立socket映射表,便于后续快速处理socket相关信息
  $this->socketListMap[$socketId] = array(
    //读取缓冲区,由于可能存在分帧的情况,此处统一先保存到缓冲区中,带收到结束帧后统一处理缓冲区
    'buffer' => '',
    //握手成功标识,addSocket在接受连接后调用,故此时并未进行握手,初始化为false
    'handshake' => false,
    //最后通信时间,用于判断超时断开操作
    'lastCommuicate' => time(),
    //socket实例
    'socket' => $socket,
    //错误计数
    'errorCnt' => 0
  );
}

removeSocket函数:

function removeSocket($socketId) {
  $socket = $this->socketListMap[$socketId]['socket'];

  //找出socket在socketList中的索引
  $socketIndex = array_search($socket, $this->socketList);
  if ($this->debug) {
    echo "RemoveSocket at $socketIndex\n";
  }

  //移除socketList中的socket
  array_splice($this->socketList, $socketIndex, 1);

  //移除socketListMap中的相关信息
  unset($this->socketListMap[$socketId]);

  //回调事件
  $this->onAfterRemoveSocket($socketId);
}

socketAccept函数:

function socketAccept() {
  //接受socket
  $socket = socket_accept($this->serverSocket);
  if ($socket !== FALSE) {
    return $socket;
  } else if ($this->debug) {
    echo $this->getLastErrMsg();
  }
}

socketRecv函数:

//从指定socket中读取数据
function socketRecv($socketId) {
  $socket = $this->socketListMap[$socketId]['socket'];
  $bufferLen = socket_get_option($socket, SOL_SOCKET, SO_RCVBUF);
  $recv = socket_recv($socket, $buffer, $bufferLen, 0);
  if ($recv === FALSE) {
    $errCode = $this->getLastErrCode($socketId);
    $this->onerror($errCode, $socketId);
    if ($this->debug) {
      echo $this->getLastErrMsg(null, $errCode);
    }
    return NULL;
  } else if ($recv > 0) {
    if ($this->debug) {
      echo "Recv:\n";
      $this->showData($buffer);
    }
    $this->socketListMap[$socketId]['lastCommuicate'] = time();
  }
  return $buffer;
}

socketSend函数:

function socketSend($socketId, $data) {
  $socket = $this->socketListMap[$socketId]['socket'];
  if ($this->debug) {
    echo "Send:\n";
    $this->showData($data);
  }
  if (socket_write($socket, $data, strlen($data)) > 0) {
    $this->socketListMap[$socketId]['lastCommuicate'] = time();
  }
}

socketClose函数:

function socketClose($socketId) {
  $socket = $this->socketListMap[$socketId]['socket'];
  socket_close($socket);
}

connect函数:

function connect($socket) {
  $this->addSocket($socket);
  $this->onconnected($socket);
}

disconnectBySocket函数:

//此函数通过遍历查找socket信息,性能较低,不建议使用,通常使用disconnect直接根据socketId断开连接
//此函数通常用于一些异常socket的处理
function disconnectBySocket($socket) {
  $socketIndex = array_search($socket, $this->socketList);
  foreach ($this->socketListMap as $socketId => $session) {
    if ($session['socket'] == $socket) {
      $this->disconnect($socketId);
      return;
    }
  }
}

disconnect函数:

function disconnect($socketId, $silent = false) {
  $this->socketClose($socketId);
  if (!$silent) {
    $this->ondisconnected($socketId);
  }
  $this->removeSocket($socketId);
}

WebSocket服务端开发(三)-WebSocketServer类握手相关函数介绍

本文介绍WebSocketServer类握手相关函数的实现。

握手函数doHandShake:

function doHandShake($socketId){
  //一旦进入了doHandshake函数,说明已收到完整的请求头,故将此socketId从handshakingList中移除
  array_splice($this->handshakingList, array_search($socketId, $this->handshakingList), 1);

  //获取socket的相关信息
  $session = $this->socketListMap[$socketId];

  //获取http请求头
  $headers = $this->getHeaders($session['buffer']);

  //请求的数据内容会清空,因为已经读取过了,这里buffer是一个读取缓冲区
  $this->socketListMap[$socketId]['buffer'] = '';
  $this->socketListMap[$socketId]['headers'] = $headers;

  //checkBaseHeader用于检查基本头信息,如果有任何一个头信息不符合WebSocket协议,则检查失败
  //checkCustomHeader为用户自定义头部检查,需要继承类覆盖实现,一般检查cookie、origin等与业务相关的头部信息
  if (!$this->checkBaseHeader($headers) || !$this->checkCustomHeader($headers)) {
    //生成握手失败响应
    $this->badRequest($socketId);

    //关闭连接
    $this->disconnect($socketId);

    //握手失败回调
    $this->onHandShakeFailure($socketId);
    return false;
  } else {
    //获取握手返回头部数据
    $responseHeader = $this->getHandShakeHeader($headers);
  }
  //发送响应头
  $this->socketSend($socketId, $responseHeader);

  //已握手标记置为true,之后在收到该socket数据将进入数据处理逻辑
  $this->socketListMap[$socketId]['handshake'] = true;

  //握手成功回调
  $this->onHandShakeSuccess($socketId);
}

checkBaseHeader函数:

function checkBaseHeader($header) {
  //检查Upgrade字段是否为websocket
  return strcasecmp($header['Upgrade'], 'websocket') === 0 &&
  //检查Connection字段是否为Upgrade
  strcasecmp($header['Connection'], 'Upgrade') === 0 &&
  //检查Sec-WebSocket-Key字段Base64解码后长度是否为16字节
  strlen(base64_decode($header['Sec-WebSocket-Key'])) === 16 &&
  //检查WebSocket协议版本是否为13,该类仅处理版本为13的WebSocket协议
    $header['Sec-WebSocket-Version'] === '13';
}

badRequest函数:

function badRequest($socketId) {
  //该函数仅拼装握手错误的响应信息,并发送
  $message = 'This is a websocket server!';
  $out = "HTTP/1.1 400 Bad request\n";
  $out .= "Server: WebSocket Server/lyz810\n";
  $out .= "Content-Length: " . strlen($message) . "\n";
  $out .= "Connection: close\n\n";
  $out .= $message;
  $this->socketSend($socketId, $out);
}

getHandShakeHeader函数:

function getHandShakeHeader($headers) {
  //拼装响应头的相关字段
  $responseHeader = array(
    'HTTP/1.1 101 Switching Protocols',
    'Upgrade: WebSocket',
    'Connection: Upgrade',
    'Sec-WebSocket-Accept: ' . $this->getWebSocketAccept($headers['Sec-WebSocket-Key']),
  );
  if (isset($headers['Sec-WebSocket-Protocol'])) {
    //子协议选择,应由继承类覆盖实现,否则默认使用最先出现的子协议
    $protocol = $this->selectProtocol(explode(',', $headers['Sec-WebSocket-Protocol']));
    array_push($responseHeader, 'Sec-WebSocket-Protocol: ' . $protocol);
  }
  return implode("\r\n", $responseHeader) . "\r\n\r\n";
}

getWebSocketAccept函数:

function getWebSocketAccept($websocketKey) {
  //根据协议要求,计算WebSocket-accept-key
  return base64_encode(sha1($websocketKey . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
}

WebSocket服务端开发(二)-WebSocketServer类主流程介绍

本文介绍WebSocketServer主函数run的实现,从整体上理解协议工作流程。

run方法代码如下:

function run() {
  //将服务器的socket添加到初始化socket列表中
  array_push($this->socketList, $this->serverSocket);
  //工作流程开始
  while (true) {
    //read为所有存在的socket列表
    $read = $this->socketList;
    //如果shutdown变量设置为true,服务器关闭,退出循环
    if ($this->shutdown) {
      $this->onshutdown();
      return;
    }

    if ($this->debug) {
      echo "Waiting for socket_select\n";
    }
    //该函数会从所有可读写的socket中选取一个socket进行处理,该方法会阻塞流程,只有在收到连接时该方法才会返回
    if (socket_select($read, $write, $except, NULL) === false) {
      if ($this->debug) {
        echo $this->getLastErrMsg();
      }
      continue;
    }

    foreach ($read as $socketItem) {
      //如果选取的socket是服务器监听的socket,则此时是新连接接入
      if ($socketItem === $this->serverSocket) {
        //接受socket连接
        $socket = $this->socketAccept();
        if ($socket) {
          //执行连接方法
          $this->connect($socket);
        }
      } else {
        //此时是连接过的socket,获取socketId
        $socketId = $this->getSocketId($socketItem);
        if ($socketId === FALSE) {
          //获取socketId失败,则将该socket断开连接
          $this->disconnectBySocket($socketItem);
          continue;
        }
        //接收传来的数据
        $data = $this->socketRecv($socketId);
        if (strlen($data) > 0) {
          //收到的数据长度不为空时,需要重置连接错误计数
          $this->socketListMap[$socketId]['errorCnt'] = 0;
          if (!isset($this->socketListMap[$socketId])) {
            $this->disconnect($socketId);
            continue;
          } else if (!$this->socketListMap[$socketId]['handshake']) {
            //尚未进行WebSocket协议握手,尝试读取连接缓冲区,如果缓冲区中没有数据,则将socketId记录到握手中列表
            //这是为了防止握手包被分成多个包进行传递(正常情况下不会出现此问题)
            //但根据HTTP协议,并未规定HTTP请求头不能被分割,故应该根据协议中的\r\n\r\n来判断请求头已发送完毕
            if (strlen($this->socketListMap[$socketId]['buffer']) === 0) {
              $this->handshakingList[$socketId] = time();
            }
            //将数据写入缓冲区
            $this->socketListMap[$socketId]['buffer'] .= $data;
            //比较后4个字节是否为\r\n\r\n
            if (substr_compare($this->socketListMap[$socketId]['buffer'], str_repeat(chr(0x0D) . chr(0x0A), 2), -4) === 0) {
              //进行握手处理
              $this->doHandShake($socketId);
            } else {
              //数据没有传送完毕,需要缓冲数据直到全部接收请求头(这个可以通过Telnet命令直接连接,每输入一个字节都会立即传给服务器,这时服务器应该缓存内容。但同时也应该设置超时时间,防止恶意占用服务器资源。)
              $this->onUpgradePartReceive($socketId);
            }
          } else if ($this->parseFrame($data, $socketId)) {
            //parseFrame会解析数据帧,如果该帧FIN标识为1则函数会返回true,交给businessHandler进行业务逻辑处理,数据在socketListMap的buffer中,所以只需要提供socketId即可找到该socket的所有信息。
            $this->businessHandler($socketId);
          }
        } else {
          $this->socketListMap[$socketId]['errorCnt'] += 1;
          if ($this->debug){
            echo "Receive empty data![$errorCnt]\n";
          }
          if ($errorCnt >= 3) {
            $this->disconnect($socketId);
          }
        }
      }
    }
    //每次处理完连接后,判断是否需要健康检查,检查之后会移除不健康的socket
    if (time() - $this->lastHealthCheck > $this->healthCheckInterval) {
      $this->healthCheck();
    }
    $this->removeUnhandshakeConnect();
  }
}

WebSocket服务端开发(一)-WebSocketServer类简介

本文介绍使用PHP语言编写的Server类WebSocketServer的整体设计思路及包含的方法。WebSocketServer类只是一个简单的单线程的WebSocket基类,可以通过继承该类实现自己的Server类。

一、成员变量
包含以下成员变量:

//各种帧类型的常量
const FRAME_CONTINUE = 0x00;
const FRAME_TEXT = 0x01;
const FRAME_BIN = 0x02;
const FRAME_CLOSE = 0x08;
const FRAME_PING = 0x09;
const FRAME_PONG = 0x0A;

protected $serverSocket = null;//服务器监听的socket
protected $shutdown = false;//关闭状态,如果是true表示服务器准备关闭
protected $socketList = array();//保存所有socket的数组
protected $socketListMap = array();//根据唯一id对socket进行索引,并保存socket的其他自定义属性,相当于session
private $handshakingList = array();//正在进行握手的socket,用于处理握手超时的socket
private $lastHealthCheck = null;//最后一次进行健康检查的时间,这里根据最后一次通信时间判断健康状态,检查时默认不会发送pong帧
private $healthCheckInterval = 300;//健康检查间隔,单位秒,每次处理完一个连接后会判断是否进行健康检查。
private $handshakeTimeout = 10;//握手超时时间,单位秒,为了防止过多的未完成握手占用系统资源,会对超时的握手连接进行关闭处理。

二、构造函数与析构函数

//port监听的端口号,address监听的IP地址,0.0.0.0表示监听本机上任何地址,debug为调试开关,为true是会打印错误信息及其他信息。
function __construct($port, $address = '0.0.0.0', $debug = false) {
  $this->serverSocket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
  socket_set_option($this->serverSocket, SOL_SOCKET, SO_REUSEADDR, 1);
  socket_set_option($this->serverSocket, SOL_SOCKET, TCP_NODELAY, 1);
  //socket绑定
  if (socket_bind($this->serverSocket, $address, $port) === false) {
    if ($debug) {
      echo $this->getLastErrMsg();
    }
    return;
  }

  //监听开始
  if (socket_listen($this->serverSocket) === false) {
    if ($debug) {
      echo $this->getLastErrMsg();
    }
    return;
  }

  $this->onstarted($this->serverSocket);
  $this->lastHealthCheck = time();
  $this->run();
}

function __destruct() {
  socket_close($this->serverSocket);
}

三、WebSocket类方法

//主工作函数,控制整体流程
function run(){}
//服务启动后回调
function onstarted($serverSocket){}
//socket连接后回调
function onconnected($socket){}
//收到不完整的握手包时回调
function onUpgradePartReceive($socketId){}
//握手失败后回调
function onHandShakeFailure($socketId){}
//握手成功后回调
function onHandShakeSuccess($socketId){}
//断开连接时回调
function ondisconnected($socketId){}
//从列表中移除socket时回调
function onAfterRemoveSocket($socketId){}
//健康检查后回调
function onafterhealthcheck($unhealthyList){}
//遇到socket错误时回调
function onerror($errCode, $socketId){}
//服务器关闭时回调
function onshutdown(){}

//健康检查函数
function healthCheck(){}
//业务逻辑处理函数(应由继承类覆盖实现)
function businessHandler($socketId){}
//帧是否结束
function isFin($byte){}
//帧是否进行掩码处理
function isMasked($byte){}
//获取帧类型
function getFrameType($byte){}
//处理任意类型的帧
function parseRawFrame($payload, $mask){}
//处理文本帧
function parseTextFrame($payload, $mask){}
//处理二进制帧
function parseBinaryFrame($payload, $mask){}
//创建关闭帧,支持关闭码及关闭原因
function closeFrame($socketId, $closeCode = 1000, $closeMsg = 'goodbye'){}
//发送ping帧
function sendPing($socketId, $data = 'ping'){}
//发送pong帧
function sendPong($socketId, $data = 'pong'){}
//获取掩码
function getMask($data, $len){}
//获取数据负载
function getPayload($data, $len){}
//获取负载长度
function getPayloadLen($data){}
//判断是否为控制帧
function isControlFrame($frameType){}
//解析帧
function parseFrame($data, $socketId){}
//创建指定类型的帧,支持分帧
function createFrame($data, $type, $fin = 0x01){}
//计算WebSocket-Accept值
function getWebSocketAccept($websocketKey){}
//协议选择(协商),建议继承后通过覆盖自定义
function selectProtocol($protocols){}
//获取握手响应头
function getHandShakeHeader($headers){}
//基础协议头检查,只检查WebSocket协议要求的请求头字段
function checkBaseHeader($header){}
//用户自定义头检查,如cookie等信息检查,需要覆盖此方法实现,该处返回恒为true
function checkCustomHeader($header){}
//创建握手失败的响应
function badRequest($socketId){}
//获取响应头的各个字段
function getHeaders($header){}
//移除握手超时的连接
function removeUnhandshakeConnect(){}
//WebSocket握手
function doHandShake($socketId){}
//显示数据信息,仅调试使用
function showData($buffer){}
//发送数据
function sendData($socketId, $data, $type = self::FRAME_TEXT, $isFin = true){}
//根据socket获取socketId
function getSocketId($socket){}
//添加socket到列表中
function addSocket($socket){}
//从列表中移除socket
function removeSocket($socketId){}
//关闭服务器
function shutdown(){}
//接受Socket
function socketAccept(){}
//从socket中读数据
function socketRecv($socketId){}
//通过socket写数据
function socketSend($socketId, $data){}
//关闭底层socket
function socketClose($socketId){}
//连接socket操作
function connect($socket){}
//通过socket断开连接
function disconnectBySocket($socket){}
//通过socketId断开连接
function disconnect($socketId, $silent = false){}
//获取错误码
function getLastErrCode($socketId = null){}
//获取错误详情
function getLastErrMsg($socketId = null, $errCode = null){}

后面的文章将具体介绍每个函数的实现。

WebSocket协议详解及应用(七)-WebSocket协议关闭帧

本文介绍WebSocket协议的关闭帧,包括客户端及服务器如何发送并处理关闭帧、关闭帧错误码及错误处理方法。

一、关闭WebSocket连接

要断开WebSocket连接,需要一个端点断开底层的TCP连接。端点需要通过某种方式来完全关闭TCP连接,例如TLS会话,并适当的丢弃未接收完毕的数据。端点也在必要时可以通过一些有效的方式断开连接,如在受到攻击时。
在一般情况下,底层TCP连接应先被服务端断开,以便保持TIME_WAIT状态。这是为了防止其在2个最大分节生命期(1~4分钟,Windows操作系统为4分钟)之内重新打开,否则可能会由于接到一个高序列的SYN包而重新打开连接。在一些异常的情况下(如在一段时间内未收到服务器端TCP关闭帧),客户端可以关闭TCP连接。如果服务器发出关闭指令,则它需要立即关闭连接。而客户端发出关闭指令需要等待服务器发送的TCP关闭帧。

二、关闭握手阶段

关闭握手阶段需要一个状态码和一个可选的关闭原因,端点必须发送一个关闭控制帧,并设置状态码和关闭原因。一旦端点发送并接收了关闭帧,就需要按上节中的方法关闭WebSocket连接。

三、关闭握手阶段开始

当接收或发送关闭帧后,代表关闭握手阶段开始,此时WebSocket连接进入到CLOSING状态。

四、关闭WebSocket连接结束

当底层TCP连接关闭时,代表WebSocket连接已关闭,此时WebSocket连接状态改为CLOSED。如果TCP连接在WebSocket关闭握手结束后断开,则此次WebSocket为一次完整的(cleanly)关闭。
如果WebSocket连接未能建立,它仍叫做连接关闭,但不是完整的。

五、关闭码

关闭帧可以包含一个关闭码和一个关闭原因。关闭帧可以由任何一方发起,也可以双方同时发起。若关闭帧没有指明关闭码,则认为关闭码为1005,如果WebSocket连接断开,而没有任何关闭帧(如底层传输时丢帧),则认为关闭码为1006。
注意:双方发送的关闭码可能不一致。例如,对方发送了一个关闭帧,但本地程序还没有将数据及关闭帧从socket接收缓存中读取出来,然后本地程序决定发送一个关闭帧,双方都会发送并接收到一个关闭帧并不会再次发送关闭帧(即只进行一次收发,即使不是发过关闭帧后收到的帧)。

六、关闭原因

关闭原因是可选的,跟在关闭码后面,为UTF-8编码的数据,并未对其内容做详细的定义。如果没有设置关闭原因,则关闭原因是一个空的字符串。
注意:同关闭码一样,双方发送的关闭原因可能不一致。

七、强制关闭连接

一些情况会引起强制关闭连接,当情况发生时,客户端需要关闭连接并将错误返回给用户(如控制台中报错等),同样,服务器需要关闭连接并将问题记录在日志中。
如果WebSocket连接建立在端点需要强制关闭连接之前,端点需要在处理关闭帧之前发送关闭帧并发送正确的关闭码。当强制关闭连接后,端点不能再次尝试向对方发送任何数据(包括关闭帧)。
除了上述情况或指定的应用层协议(如WebSocket API)外,客户端不应该断开连接。

八、关闭码

1000 正常关闭
1001 端点丢失,如服务器宕机或浏览器切换其他页面
1002 协议错误
1003 数据类型错误(例如端点只能处理文本,但传来了二进制消息)
1004 保留
1005 保留,禁止由端点发送此类型关闭帧,它是用来当端点没有表明关闭码时的默认关闭码。
1006 保留,禁止由端点发送此类型关闭帧,它是用来当端点未发送关闭帧,连接异常断开时使用。
1007 数据内容错误(如在text帧中非utf-8编码的数据)
1008 端点已接收消息,但违反其策略。当没有更好的关闭码(1003或1009)的时候用此关闭码或者不希望显示错误细节。
1009 内容过长
1010 客户端期望服务器协商一个或多个扩展,但这些扩展并未在WebSocket握手响应中返回。
1011 遇到未知情况无法执行请求
1015 保留,禁止由端点发送此类型关闭帧,它会在TLS握手失败(如证书验证失败)时返回。
保留关闭码
0-999 尚未使用
1000-2999 协议保留,用于未来版本、扩展等
3000-3999 为库、框架、应用程序保留,这些状态码可在IANA中注册,这些状态码并未在此协议中实现。
4000-4999 私有保留,不可被注册。用于开发者自定义关闭码。