socket = $sock; } public function fork() { for ($i=0; $i < $this->workerNumber; $i ) { $pid = pcntl_fork(); if ($pid == -1) { die('could not fork'); } else if ($pid == 0) { // we are the child $this->accept(); exit; } } for ($i=0; $i < $this->workerNumber; $i ) { $status=0; $pid=pcntl_wait($status); //Protect against Zombie children echo "子进程回收了:$pid".PHP_EOL; } } public function accept() { $client_id = 0; while (true) { if (($msgsock = socket_accept($this->socket)) === false) { echo "socket_accept() failed: reason: " . socket_strerror(socket_last_error($this->socket)) . "\n"; } else { $client_id = 1; if (is_callable($this->onConnect)) { call_user_func($this->onConnect, (int)$msgsock); } } if (false === ($buf = @socket_read($msgsock, 2048))) { echo "socket_read() failed: reason: " . socket_strerror(socket_last_error($msgsock)) . "\n"; } else { if (is_callable($this->onMessage)) { call_user_func($this->onMessage, $msgsock, $buf); } } socket_close($msgsock); } } public function run() { $this->fork(); }} $worker = new Worker('tcp://0.0.0.0:9800'); $worker->onConnect = function ($fd) { echo '新的连接来了'.$fd. 'pid' . posix_getpid() . PHP_EOL;};$worker->onMessage = function ($conn, $message) { // echo "读取到数据: {$message}"; $content = "Hello World"; $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; socket_write($conn, $http_resonse, strlen($http_resonse));};$worker->run();
暂无评论