一个简单的PHP的进程加socket加WS的例子

Posted 2020-07-07 木子牛

此代码基于TP框架，

服务器端命令行运行 php index.php Home/Ws/start即可。

注意更改代码

socket代码

<?php

namespace Home\Controller;
use Think\Controller;
use Home\Controller\RedisController;

class WsSocketController {

//存储连接资源
private $clients;
//主线端口
private $master;
//客户资源
private $user;
private $ip;
private $port;
private $redis;

public function __construct($socketHandle) {

$this->master = $socketHandle;
$this->clients = array(‘s‘=>$this->master);
$this->redis = RedisController::connect();
}


//监控socket
function run() {
while (true) {
$read = $this->clients;
socket_select($read, $write = NULL, $except = NULL, null);

foreach($read as $r) {

if($r == $this->master) {
//建立四步握手协议
//有客户端新进连接进来
//判断服务器socket是否被激活
//接受客户端,并将他添加到客户数组
$this->clients[] = $newCli = socket_accept($this->master);
$this->user[] = array(‘handle‘=>false);
$pid = posix_getpid().‘_‘.count($this->user);
$this->redis->sadd(‘PID‘,$pid);

} else {
//已有的连接发来的消息所有的客户数据读取
// 读到换行符或1024字节
// 虽然客户端断开连接时显示错误,所以沉默错误消息
$buffer = ‘‘;
$data = socket_recv($r,$buffer ,2048,0);

// 检查客户端是否断开连接
if ($data < 7) {
// 删除客户端为客户数组
$this->close($r);
continue;
}

$k = $this->search($r);

if($this->user[$k][‘handle‘] == false) { //只是进行了四步握手协议，此时浏览器发来WS协议请求
$this->handshake($k,$buffer); //返回WS协议应答

} else { //已经建立WS协议，获取通过WS发来的消息
$pid = posix_getpid().‘_‘.count($this->user);
$str = $this->uncode($buffer).‘__‘.$pid;
$this->sendAll($str,$r); //发送消息广播

}


}


}
}



}



/*
*发送WS协议，建立WS协议链接
*@param int socket索引
*@param string WS发送的请求协议内容
*/
private function handshake($k,$buffer){
//获取KEY及生成新的KEY
$buf = substr($buffer,strpos($buffer,‘Sec-WebSocket-Key:‘)+18);
$key = trim(substr($buf,0,strpos($buf,"\r\n")));
$new_key = base64_encode(sha1($key."258EAFA5-E914-47DA-95CA-C5AB0DC85B11",true));

//返回HTTP协议
$new_message = "HTTP/1.1 101 Switching Protocols\r\n";
$new_message .= "Upgrade: websocket\r\n";
$new_message .= "Sec-WebSocket-Version: 13\r\n";
$new_message .= "Connection: Upgrade\r\n";
$new_message .= "Sec-WebSocket-Accept: " . $new_key . "\r\n\r\n";
socket_write($this->clients[$k],$new_message,strlen($new_message));
$this->user[$k][‘handle‘] = true;

return true;

}


/*
*群发消息
*@param string 发送的内容
*@param resource 发送消息的资源连接
*/
private function sendAll($str,$socket) {
// 把这所有的客户在客户数组(除了第一个,这是一个监听套接字)
foreach ($this->clients as $send_sock) {
// 如果这个资源等于 服务器或是等于它自己则跳过，
if ($send_sock == $this->master )//|| $send_sock == $socket
continue;
//发送给其他用户一个信息
$data = $this->code($str);
socket_write($send_sock, $data,strlen($data));
}

}




/*
*查找socket资源索引
*@param resource socket资源索引
*
*/
private function search($sign) {
// foreach ($this->user as $k=>$v) {
// if($sign == $v[‘socket‘])
// return $k;
// }
$k = array_search($sign,$this->clients);
return $k;
}


/*
*断开sockt连接
*@param resource socket资源连接
*/

private function close($sign){//通过标示断开连接
$k = array_search($sign, $this->clients);
socket_close($sign); //断开的连接需要在服务器端手动关闭
unset($this->user[$k]);
unset($this->clients[$k]);
}


/*
*
*WS 信息解码
*@param string 字符串
*/
private function uncode($str){
$mask = array();
$data = ‘‘;
$msg = unpack(‘H*‘,$str);
$head = substr($msg[1],0,2);
if (hexdec($head{1}) === 8) {
$data = false;
}else if (hexdec($head{1}) === 1){
$mask[] = hexdec(substr($msg[1],4,2));
$mask[] = hexdec(substr($msg[1],6,2));
$mask[] = hexdec(substr($msg[1],8,2));
$mask[] = hexdec(substr($msg[1],10,2));
$s = 12;
$e = strlen($msg[1])-2;
$n = 0;
for ($i=$s; $i<= $e; $i+= 2) {
$data .= chr($mask[$n%4]^hexdec(substr($msg[1],$i,2)));
$n++;
}
}
return $data;
}


//信息加码
function code($msg){
$msg = preg_replace(array(‘/\r$/‘,‘/\n$/‘,‘/\r\n$/‘,), ‘‘, $msg);
$frame = array();
$frame[0] = ‘81‘;
$len = strlen($msg);
$frame[1] = $len<16?‘0‘.dechex($len):dechex($len);
$frame[2] = $this->ord_hex($msg);
$data = implode(‘‘,$frame);
return pack("H*", $data);
}


function ord_hex($data) {
$msg = ‘‘;
$l = strlen($data);
for ($i= 0; $i<$l; $i++) {
$msg .= dechex(ord($data{$i}));
}
return $msg;
}


function log($t){//控制台输出

$t=$t."\r\n";
// fwrite(STDOUT, iconv(‘utf-8‘,‘gbk//IGNORE‘,$t));

}


}

 

后台进程代码：

<?php
//运行域cli模式下
namespace Home\Controller;
use Think\Controller;

class PcntlController
{
//最多开启的进程数
private $size;
//
private $currSize=0;//当前进程数
//运行的程序对象
private $pro;
//运行的方法
private $run;



public function __construct() {
$this->daemonize();
}


/*
*构造函数
*@param string 对象
*@param string 方法名
*@param int 要开启的子进程数量
*
*/
public function set($controller,$active,$size) {
$this->size = $size;
$this->pro = $controller;
$this->run = $active;
}

/*
*进程守护化函数，使进程脱离当前终端控制，以便后台独立运行。
*
*/

function daemonize() {
$pid = pcntl_fork();//创建子进程
if($pid == -1) {
exit(‘创建进程失败‘);
} else if($pid > 0) {
//让父进程退出。以便开启新的会话
exit(0);
}

//建立一个有别于终端的新的回话易脱离终端，防止退出终端的时候，进程被kill
posix_setsid();
$pid = pcntl_fork();
if($pid == -1) {
die(‘创建进程失败‘);

} else if($pid > 0) {
//父进程推出剩下的子进程为独立进程，归为系统管理此进程
exit(0);
}
}

//运行程序
public function start() {
//循环创建子进程
// while(true) {
// $clientPid = pcntl_fork();//创建子进程
// if($clientPid == -1) {
// file_put_contents(‘/usr/local/nginx/html/youquwei/log.txt‘,‘创建进程失败‘);
// } else if($clientPid > 0) {//父进程程序
// $this->currSize++;
// if($this->currSize >= $this->size) {
// $sunpid = pcntl_wait($status);
// $this->currSize--;
// }
// }
//子进程程序执行
//work 进程 ，子进程工作
// $run = $this->run;
// $this->pro->$run();
// }
$pid = pcntl_fork();
if($pid==0){
for($i=0;$i<$this->size;$i++) {
$pid = pcntl_fork();
if($pid == 0) {
$run = $this->run;
$this->pro->$run();
}else if ($pid == -1) {
file_put_contents(‘/usr/local/nginx/html/youquwei/log.txt‘,‘创建进程失败‘);
}
}
if($pid > 0) {
$p = pcntl_wait($status);
}
}

}



}

 

 

运行代码：

      

<?php
//运行域cli模式下
namespace Home\Controller;
use Think\Controller;
use Home\Controller\PcntlController;
use Home\Controller\WsSocketController;
class WsController
{
//Socket监听句柄
private $socketHandle;

private $ip;
private $port;
//进程
private $pcntl;
//socket
private $socket;

public function __construct($ip="192.168.0.130",$port = 8080) {
$this->pcntl= new PcntlController;
$this->ip = $ip;
$this->port = $port;
ob_implicit_flush();

error_reporting(1);
//set_time_limit(0);

$this->socketHandle = $this->WebSocket();
$this->socket = new WsSocketController($this->socketHandle);

}

//建立服务器端连接
public function WebSocket(){
$server = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_set_option($server, SOL_SOCKET, SO_REUSEADDR, 1);
socket_bind($server, $this->ip, $this->port);
socket_listen($server);
return $server;
}


public function start() {

$this->pcntl->set($this->socket,‘run‘,3);
$this->pcntl->start();
//$this->socket->run();
}

}

 

 

 

前端页面代码：

    

<!doctype html>
<html>
<head>
<meta charset="utf-8">
<title>websocket_TEST</title>
</head>
<body>
<!--
SOCKET流程：
1：服务器端开启端口等待客户端建立TCP连接
2：客户端建立TCP四次握手协议TCP连接 （TCP的四次握手连接全部被封装好的代码处理）

3: 客户端发起WS协议
4：服务器判断 客户端发送消息内容即是WS协议请求
5：服务器端返回WS连接协议

6：双方建立WS长连接
7：此时客户端可以接受服务器推送过来的内容（接受WS加密内容），也可以向服务器端发送内容(服务端解码WS加密内容)。

双方关闭SOCKET

-->

 

<textarea class="log" style="width: 100%; height: 500px;">
=======websocket======
</textarea>
<input type="button" value="连接" onClick="link()">
<input type="button" value="断开" onClick="dis()">
<input type="text" id="text">
<input type="button" value="发送" onClick="send()">
<script type="text/javascript" src="http://cdn.bootcss.com/jquery/1.11.2/jquery.min.js"></script>
<script>
function link(){
var url=‘ws://127.0.0.1:8080‘;
socket=new WebSocket(url);
socket.onopen=function(){
log(‘连接成功‘);
}

socket.onmessage=function(msg) {
log(‘获得消息:‘+msg.data);
console.log(msg);
}

socket.onclose=function(){log(‘断开连接‘)}
}

function dis(){
socket.close();
socket=null;
}

function log(var1){
$(‘.log‘).append(var1+"\r\n");
}

function send(){
socket.send($(‘#text‘).val());
}

</script>
</body>
</html>