xyzshop/helper/event/util.php

<?php
/**
 * Created by PhpStorm.
 * User: stanley-king
 * Date: 2018/7/28
 * Time: 下午12:49
 */

namespace event;

use Log;

class util
{
    static public function listen($host,$port)
    {
        $listen_fd = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);

        if(!socket_set_option($listen_fd, SOL_SOCKET, SO_REUSEADDR, 1)) {
            Log::record("socket_set_option 地址重用失败.",Log::DEBUG);
            return false;
        }
        if(!socket_set_nonblock($listen_fd)) {
            $err = socket_last_error();
            Log::record("socket_set_blocking error : {$err}",Log::DEBUG);
            return false;
        }
        if(!socket_bind($listen_fd, $host, $port)) {
            echo "无法绑定socket {$host} : {$port},请退出之前进程.\n";
            return false;
        }
        if(!socket_listen($listen_fd)) {
            echo "无法监听socket,请退出之前进程.\n";
            return false;
        }

        return $listen_fd;
    }

    static public function listen_block($host,$port)
    {
        $listen_fd = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);

        if(!socket_set_option($listen_fd, SOL_SOCKET, SO_REUSEADDR, 1)) {
            Log::record("socket_set_option 地址重用失败.",Log::DEBUG);
            return false;
        }
        if(!socket_bind($listen_fd, $host, $port)) {
            echo "无法绑定socket {$host} : {$port},请退出之前进程.\n";
            return false;
        }
        if(!socket_listen($listen_fd)) {
            echo "无法监听socket,请退出之前进程.\n";
            return false;
        }

        return $listen_fd;
    }

    static function fork_child($callback, $params)
    {
        if (($pid = pcntl_fork()) === 0)
        {
            //ob_end_clean(); // Discard the output buffer and close
            fclose(STDIN);  // Close all of the standard
            fclose(STDOUT); // file descriptors as we
            fclose(STDERR); // are running as a daemon.

            Log::record("pcntl_fork pid = {$pid}",Log::DEBUG);
            if(!empty($params)) {
                call_user_func_array($callback,$params);
            } else {
                call_user_func($callback);
            }
            exit();
        }
        elseif($pid === -1)
        {
            Log::record("pid = {$pid},could not fork process",Log::DEBUG);
            die('could not fork process');
        }
        else
        {
            $ret = pcntl_waitpid($pid,$status,WNOHANG);
            Log::record("pcntl_waitpid pid = {$pid},ret={$ret}",Log::DEBUG);

            if($ret == 0) {
                Log::record("fork_child: successful ret == 0 PID: {$pid}",Log::DEBUG);
                return $pid;
            }
            elseif($ret == -1) {
                Log::record("fork_child: ret == -1  PID: {$pid}",Log::DEBUG);

            }
            else {
                Log::record("fork_child: ret == 0 child exited PID: {$pid}.",Log::DEBUG);
            }
        }
    }

    static function fork_listen($fd, $callback, $count)
    {
        $params = [$fd];
        $count = intval($count);
        if($count <= 0) {
            return call_user_func_array($callback,$params);
        }
        else
        {
            $childs = [];
            while ($count--) {
                $childs[] = util::fork_child($callback,$params);
            }
        }

        return $childs;
    }

    static function fork_listenex($fd, $callback, $count)
    {
        self::fork_listen($fd,$callback,$count);
        while ($count) {
            Log::record("waitting child quit......",Log::DEBUG);
            $ret = pcntl_wait($status);
            Log::record("pcntl_waitpid ret={$ret}",Log::DEBUG);
            if($ret > 0)
            {
                if(pcntl_wifexited($status)) {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child normal quit ret={$ret} status = {$status}",Log::DEBUG);
                } else {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child abnormal quit ret={$ret} status = {$status}",Log::DEBUG);
                }
                self::fork_listen($fd,$callback,1);
            }
            else {
                //子进程已经退出干净了.
                break;
            }
        }
    }

    static function join()
    {
        do {
            Log::record("waitting child quit......",Log::DEBUG);
            $ret = pcntl_wait($status);
            Log::record("pcntl_waitpid ret={$ret}",Log::DEBUG);
            if($ret > 0)
            {
                if(pcntl_wifexited($status)) {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child normal quit ret={$ret} status = {$status}",Log::DEBUG);
                } else {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child abnormal quit ret={$ret} status = {$status}",Log::DEBUG);
                }
            }
            else {
                //子进程已经退出干净了.
                break;
            }
        }
        while(true);
    }

    static function fork_worker($callback, $count)
    {
        $count = intval($count);
        if($count <= 0) {
            return call_user_func($callback);
        }
        else
        {
            while ($count--) {
                util::fork_child($callback,NULL);
            }
        }
    }

    static function fork_workerex($callback, $count)
    {
        self::fork_worker($callback,$count);
        while ($count) {
            Log::record("waitting child quit......",Log::DEBUG);
            $ret = pcntl_wait($status);
            Log::record("pcntl_waitpid ret={$ret}",Log::DEBUG);
            if($ret > 0)
            {
                if(pcntl_wifexited($status)) {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child normal quit ret={$ret} status = {$status}",Log::DEBUG);
                } else {
                    $ret = pcntl_wexitstatus($status);
                    Log::record("child abnormal quit ret={$ret} status = {$status}",Log::DEBUG);
                }
                self::fork_worker($callback,1);
            }
            else {
                //子进程已经退出干净了.
                break;
            }
        }
    }
}