使用event+pnctl 模拟workerman写了一个demo,但是发现一个问题
就是多次tcp连接之后,开4个线程,第四次连接tcp就不返回数据了
开10个线程,连接断开之后,大概7-8次就会出现数据不返回
<?php
class Worker{
//监听socket
protected $socket = NULL;
//连接事件回调
public $onConnect = NULL;
public $reusePort=1;
//接收消息事件回调
public $onMessage = NULL;
public $workerNum=10; //子进程个数
public $allSocket; //存放所有socket
public $addr;
protected $worker_pid; //子进程pid
protected $master_pid;//主进程id
public function __construct($socket_address) {
//监听地址+端口
$this->addr=$socket_address;
$this->master_pid=posix_getpid();
}
public function start() {
//获取配置文件
$this->fork($this->workerNum);
$this->monitorWorkers(); //监视程序,捕获信号,监视worker进程
}
public function signalHandler($sigo){
switch ($sigo){
case SIGUSR1:
$this->reload();
echo "收到重启信号";
break;
}
}
public function fork($worker_num){
for ($i=1;$i<$worker_num;$i++){
$pid=pcntl_fork(); //创建成功会返回子进程id
if($pid<0){
exit('创建失败');
}else if($pid>0){
//父进程空间,返回子进程id
$this->worker_pid[]=$pid;
}else{ //返回为0子进程空间
$this->accept();//子进程负责接收客户端请求
exit;
}
}
//放在父进程空间,结束的子进程信息,阻塞状态
}
/**
* 捕获信号
* 监视worker进程.拉起进程
*/
public function monitorWorkers(){
//注册信号事件回调,是不会自动执行的
// reload
pcntl_signal(SIGUSR1, array($this, 'signalHandler'),false); //重启woker进程信号
while (1){
// 当发现信号队列,一旦发现有信号就会触发进程绑定事件回调
\pcntl_signal_dispatch();
$status=0;
$pid = pcntl_wait($status,\WUNTRACED); //当信号到达之后就会被中断
\pcntl_signal_dispatch();
//进程重启的过程当中会有新的信号过来,如果没有调用pcntl_signal_dispatch,信号不会被处理
}
}
public function accept(){
$opts = array(
'socket' => array(
'backlog' =>10240, //成功建立socket连接的等待个数
),
);
$context = stream_context_create($opts);
//开启多端口监听,并且实现负载均衡
stream_context_set_option($context,'socket','so_reuseport',1);
stream_context_set_option($context,'socket','so_reuseaddr',1);
$this->socket=stream_socket_server($this->addr,$errno,$errstr,STREAM_SERVER_BIND|STREAM_SERVER_LISTEN,$context);
stream_set_blocking($this->socket,false);//非阻塞
$base=new EventBase();
$ss =$this->socket;
//监听服务端的socket
$event=new Event($base,$ss,Event::PERSIST |Event::READ | Event::WRITE,function ($ss){
$clientSocket = stream_socket_accept($ss);
//触发事件的连接的回调
if(!empty($clientSocket) && is_callable($this->onConnect)){
call_user_func($this->onConnect,$clientSocket);
}
$base=new EventBase();
//监听客户端socket
$event=new Event($base,$clientSocket,Event::PERSIST |Event::READ | Event::WRITE,function ($clientSocket){
$buffer=fread($clientSocket,65535);
if(empty($buffer)){
if(!is_resource($clientSocket) || feof($clientSocket) ){
//触发关闭事件
fclose($clientSocket);
}
}
//正常读取到数据,触发消息接收事件,响应内容
if(!empty($buffer) && is_callable($this->onMessage)){
call_user_func($this->onMessage,$clientSocket,$buffer);
}
});
$event->add(); //加入事件监听
$base->loop();
});
$event->add(); //加入事件监听
$base->loop(); //调度挂起事件监听
}
/**
* 重启worker进程
*/
public function reload(){
foreach ($this->worker_pid as $index=>$pid){
posix_kill($pid,SIGKILL); //结束进程
var_dump("杀掉的子进程",$pid);
unset($this->worker_pid[$index]);
$this->fork(1); //重新拉起worker
}
}
//捕获信号之后重启worker进程
}
$worker = new Worker('tcp://0.0.0.0:9800');
//开启多进程的端口监听
$worker->reusePort = true;
//连接事件
$worker->onConnect = function ($fd) {
//echo '连接事件触发',(int)$fd,PHP_EOL;
};
$worker->onTask = function ($fd) {
//echo '连接事件触发',(int)$fd,PHP_EOL;
};
//消息接收
$worker->onMessage = function ($conn, $message) {
//事件回调当中写业务逻辑
//var_dump($conn,$message);
$content=$message;
$http_resonse = "HTTP/1.1 200 OK\r\n";
$http_resonse .= "Content-Type: text/html;charset=UTF-8\r\n";
$http_resonse .= "Connection: keep-alive\r\n"; //连接保持
$http_resonse .= "Server: php socket server\r\n";
$http_resonse .= "Content-length: ".strlen($content)."\r\n\r\n";
$http_resonse .= $content;
fwrite($conn, $http_resonse);
if(empty($message)){
if(!is_resource($conn) || feof($message) ){
//触发关闭事件
fclose($conn);
}
}
};
$worker->start(); //启动
问下是哪里出了问题