一、寫在前面

21年的時候有機(jī)會第一次接觸了webman，初次接觸這種模式的框架時，感覺還是挺新穎的。目前已經(jīng)使用了1年多，感覺挺不錯的。之前在論壇看到了第三方項目中對PHPForker的介紹，于是有機(jī)會學(xué)習(xí)了一下，并重新回過頭來看webman的啟動流程。

二、說明

以下流程圖是個人理解，有不對的地方請指出來，我將修正它。
同時聲明下該流程的幾個點(diǎn)

1. 該流程核心關(guān)注接收并響應(yīng)請求的流程，針對定時器部分暫未列出
2. 默認(rèn)處理流程涉及的系統(tǒng)為Linux
3. 涉及信號的部分未列出
4. 監(jiān)聽類型為tcp

三、流程圖

如果圖片查看不完整的話，看這里img

四、簡單的server服務(wù)

<?php
declare(strict_types=1);

namespace Stream;

class Timer
{
    /**
     * @var SplPriorityQueue
     */
    private static SplPriorityQueue $queue;

    public function __construct()
    {
        // 初始化優(yōu)先隊列
        self::$queue = new SplPriorityQueue();
        // 定義extra
        self::$queue->setExtractFlags(SplPriorityQueue::EXTR_BOTH);
    }

    public function init(): void
    {
        // 注冊信號
        pcntl_signal(SIGALRM, [$this, 'handler'], false);
    }

    /**
     * @param callable $func 回調(diào)函數(shù)
     * @param float $interval
     * @return void
     */
    public function addTimer(callable $func, float $interval): void
    {
        // 如果不存在任務(wù)，則創(chuàng)建一個信號
        if (self::$queue->count() === 0) {
            pcntl_alarm(1);
        }
        $now = hrtime(true) / 1e-9;
        $nextRunTime = $now + $interval;
        self::$queue->insert(
            [
                'interval' => $interval,
                'func' => $func
            ],
            -$nextRunTime
        );
        $count = self::$queue->count();
        printf('queue count:%d' . PHP_EOL, $count);
    }

    public function handler(): void
    {
        pcntl_alarm(1);
        $this->tick();
    }

    public function tick(): void
    {
        $count = self::$queue->count();
        $now = hrtime(true) / 1e-9;
        while ($count--) {
            $data = self::$queue->top();
            $runTime = -$data['priority'];
            if ($runTime <= $now) {
                self::$queue->extract();
                call_user_func($data['data']['func']);
                $this->addTimer($data['data']['func'], $data['data']['interval']);
            }
        }
    }

}

class Server
{
    private string $server = 'tcp://127.0.0.1:5501';

    public function listenSelect(): void
    {
        $mainSocket = stream_socket_server($this->server, $errorCode, $errorMsg);

        stream_set_blocking($mainSocket, false);

        $read = [];
        $write = $except = null;

        $read[$this->server] = $mainSocket;

        while (true) {
            // 調(diào)用等待信號的處理器，Timer部分
            pcntl_signal_dispatch();

            $tmpRead = $read;
            $tmpWrite = $write;

            try {
                $select = stream_select($tmpRead, $tmpWrite, $except, 1, 0);
            } catch (\Throwable $e) {
                printf($e->getMessage() . PHP_EOL);
                continue;
            }

            if ($select === false) {
                continue;
            }

            foreach ($tmpRead as $sid => $socket) {
                if ($socket === $mainSocket) {
                    // 說明有新的鏈接進(jìn)入
                    $newSocket = stream_socket_accept($socket, 0, $newSocketPeer);
                    if ($newSocket === false) {
                        print '接受connection失?。? . $newSocketPeer . PHP_EOL;
                        continue;
                    }
                    $socketAddress = 'tcp://' . stream_socket_get_name($newSocket, true);
                    $read[$socketAddress] = $newSocket;
                    print '接受connection成功：' . $newSocketPeer . PHP_EOL;
                }
                else {
                    // 從客戶端讀取數(shù)據(jù)，如何確定是當(dāng)前這個client發(fā)來的數(shù)據(jù)？
                    // 因為select拿到數(shù)據(jù)，說明一定是有新的數(shù)據(jù)被讀到（返回的read是有新數(shù)據(jù)到達(dá)的socket）
                    $msg = fread($socket, 65535);
                    if ($msg === '' || $msg === false) {
                        // 移除read
                        foreach ($read as $k => $v) {
                            if ($socket === $v) {
                                unset($read[$k]);
                            }
                        }
                        fclose($socket);
                        print '關(guān)閉connection：' . $sid . PHP_EOL;
                    } else {
                        // 打印信息，并寫回
                        print '收到connection消息：' . $msg . PHP_EOL;
                        if (in_array($socket, $read, true)) {
                            fwrite($socket, '已收到信息，' . date('H:i:s') . PHP_EOL);
                        }
                    }
                }
            }

//            usleep(100000);
        }

    }
}

$timer = new Timer();
$timer->addTimer(
    static function() {
        $time = time();
        printf('this is timer: hello world: %d' . PHP_EOL, $time);
    },
    1
);

$timer->init();
(new Server())->listenSelect();

四、中途遇到的幾個問題

4.1、關(guān)于stream_select一直返回

起初在本地測試的過程中，針對stream_select(&$read, &$write, &$except, $tv_sec, $tv_usec)，我學(xué)著的PHPForker的做法，將新連接進(jìn)來的socket不僅放入$read中，同時也放入$write中，這樣一旦啟動之后新連接進(jìn)入后，原本的stream_select應(yīng)該阻塞直到超時的，卻并沒有被阻塞。通過debug調(diào)試發(fā)現(xiàn)，這時每次select都有write返回，也就是該新加進(jìn)去的socket。

這里起初，我針對stream_select的什么時候返回的理解，除了超時場景下返回外，另外的就是監(jiān)聽的$read、$write、$except的文件描述符有新變化才返回。所以當(dāng)?shù)谝淮螌阎鲃颖O(jiān)聽的socket放進(jìn)到read時（這里后面統(tǒng)一稱為mainSocket），如果此時有客戶端連接進(jìn)來，那么下一次在select監(jiān)聽的時候，就會發(fā)現(xiàn)read中的mainSocket有就緒的狀態(tài)，所以這時我們將受理連接，也就是通過stream_socket_accept去接收，得到一個客戶端的socket。如果此時將該socket不僅放入$read去監(jiān)聽，同時也放到$write去監(jiān)聽，那接下來每次select的時候，都會立即返回。為什么呢？

后面我翻到了有其他人也有同樣的疑惑http://wtbis.cn/q/1307，看到這個我才大概明白。前面放進(jìn)去的$write一直處于可寫狀態(tài)，所以每次select都能拿到。

4.2、理解Linux下的select、poll、epoll模型

這里找到了幾篇解說挺好的文章，可以參考下，它有一個系列PHP socket初探，PHP Socket初探—-先從一個簡單的socket服務(wù)器開始

五、推薦文章

1、php socket通信中stream_select方法的理解
2、stream_select ($read, $write, $except, $timeout ); 函數(shù)問題
3、PHP Socket初探—-先從一個簡單的socket服務(wù)器開始

原文見這里