摘要：在中的應用官網源碼解讀號外號外歡迎大家我們開發組定了一個就線下聚一次的小目標上一篇源碼解讀反響還不錯不少同學推薦再加一篇講解一下中使用到的功能幫助大家開啟的實戰之旅服務器開發涉及到的相關技術領域的知識非常多不日積月累打好基礎是很難真正

date: 2017-12-14 21:34:51
title: swoole 在 swoft 中的應用

swoft 官網: https://www.swoft.org/
swoft 源碼解讀: http://naotu.baidu.com/file/8...
號外號外, 歡迎大家 star, 我們開發組定了一個 star 1000+ 就線下聚一次的小目標

上一篇 blog - swoft 源碼解讀 反響還不錯, 不少同學推薦再加一篇, 講解一下 swoft 中使用到的 swoole 功能, 幫助大家開啟 swoole 的 實戰之旅.

服務器開發涉及到的相關技術領域的知識非常多, 不日積月累打好基礎, 是很難真正做好的. 所以我建議:

swoole wiki 最好看 3 遍, 包括評論. 第一遍快速過一遍, 形成大致印象; 第二遍邊看邊敲代碼; 第三遍可以選擇衍生的開源框架進行實戰. swoft 就是不錯的選擇.

swoole wiki 發展到現在已經 1400+ 頁, 確實會有點難啃, 勇敢的少年呀, 加油.

swoole 在 swoft 中的應用:

SwooleServer: swoole2.0 協程 Server

SwooleHttpServer: swoole2.0 協程 http Server, 繼承自 SwooleServer

SwooleCoroutineClient: 協程客戶端, swoole 封裝了 tcp / http / redis / mysql

SwooleCoroutine: 協程工具集, 獲取當前協程id，反射調用等能力

SwooleProcess: 進程管理模塊, 可以在 SwooleServer 之外擴展更多功能

SwooleAsync: 異步文件 IO

SwooleTimer: 基于 timerfd + epoll 實現的異步毫秒定時器，可完美的運行在 EventLoop 中

SwooleEvent: 直接操作底層 epoll/kqueue 事件循環(EventLoop)的接口

SwooleLock: 在 PHP 代碼中可以很方便地創建一個鎖, 用來實現數據同步

SwooleTable: 基于共享內存實現的超高性能數據結構

使用 swoole 的 http server 相較 tcp server 還是要簡單一些, 只需要關心:

SwooleHttpServer

SwooleHttpRequest

SwooleHttpResponse

先看 http server:

// SwoftServerHttpServer
public function start()
{
    // http server
    $this->server = new SwooleHttpServer($this->httpSetting["host"], $this->httpSetting["port"], $this->httpSetting["model"], $this->httpSetting["type"]);

    // 設置事件監聽
    $this->server->set($this->setting);
    $this->server->on("start", [$this, "onStart"]);
    $this->server->on("workerStart", [$this, "onWorkerStart"]);
    $this->server->on("managerStart", [$this, "onManagerStart"]);
    $this->server->on("request", [$this, "onRequest"]);
    $this->server->on("task", [$this, "onTask"]);
    $this->server->on("pipeMessage", [$this, "onPipeMessage"]);
    $this->server->on("finish", [$this, "onFinish"]);

    // 啟動RPC服務
    if ((int)$this->serverSetting["tcpable"] === 1) {
        $this->listen = $this->server->listen($this->tcpSetting["host"], $this->tcpSetting["port"], $this->tcpSetting["type"]);
        $tcpSetting = $this->getListenTcpSetting();
        $this->listen->set($tcpSetting);
        $this->listen->on("connect", [$this, "onConnect"]);
        $this->listen->on("receive", [$this, "onReceive"]);
        $this->listen->on("close", [$this, "onClose"]);
    }

    $this->beforeStart();
    $this->server->start();
}

使用 swoole server 十分簡單:

傳入配置 server 配置信息, new 一個 swoole server

設置事件監聽, 這一步需要大家對 swoole 的進程模型非常熟悉, 一定要看懂下面 2 張圖

啟動服務器

swoft 在使用 http server 時, 還會根據配置信息, 來判斷是否同時新建一個 RPC server, 使用 swoole 的 多端口監聽 來實現.

再來看 Request 和 Response, 提醒一下, 框架設計的時候, 要記住 規范先行:

PSR-7: HTTP message interfaces

phper 比較熟悉的應該是 $_GET $_POST $_COOKIE $_FILES $_SERVER 這些全局變量, 這些在 swoole 中都得到了支持, 并且提供了更多方便的功能:

// SwooleHttpRequest $request
$request->get(); // -> $_GET
$request->post(); // -> $_POST
$request->cookie(); // -> $_COOKIE
$request->files(); // -> $_FILES
$request->server(); // -> $_SERVER

// 更方便的方法
$request->header(); // 原生 php 需要從 $_SERVER 中取
$request->rawContent(); // 獲取原始的POST包體

這里強調一下 $request->rawContent(), phper 可能用 $_POST 比較 6, 導致一些知識不知道: post 的數據的格式. 因為這個知識, 所以 $_POST 不是所有時候都能取到數據的, 大家可以網上查找資料, 或者自己使用 postman 這樣的工具自己測試驗證一下. 在 $_POST 取不到數據的情況下, 會這樣處理:

$post = file_get_content("php://input");

$request->rawContent() 和這個等價的.

swoft 封裝 Request 對象的方法, 和主流框架差不多, 以 laravel 為例(實際使用 symfony 的方法):

// SymfonyRequest::createFromGlobals()
public static function createFromGlobals()
{
    // With the php"s bug #66606, the php"s built-in web server
    // stores the Content-Type and Content-Length header values in
    // HTTP_CONTENT_TYPE and HTTP_CONTENT_LENGTH fields.
    $server = $_SERVER;
    if ("cli-server" === PHP_SAPI) {
        if (array_key_exists("HTTP_CONTENT_LENGTH", $_SERVER)) {
            $server["CONTENT_LENGTH"] = $_SERVER["HTTP_CONTENT_LENGTH"];
        }
        if (array_key_exists("HTTP_CONTENT_TYPE", $_SERVER)) {
            $server["CONTENT_TYPE"] = $_SERVER["HTTP_CONTENT_TYPE"];
        }
    }

    $request = self::createRequestFromFactory($_GET, $_POST, array(), $_COOKIE, $_FILES, $server); // xglobal,

    if (0 === strpos($request->headers->get("CONTENT_TYPE"), "application/x-www-form-urlencoded")
        && in_array(strtoupper($request->server->get("REQUEST_METHOD", "GET")), array("PUT", "DELETE", "PATCH"))
    ) {
        parse_str($request->getContent(), $data);
        $request->request = new ParameterBag($data);
    }

    return $request;
}

SwooleHttpResponse 也是支持常見功能:

// SwooleHttpResponse $response
$response->header($key, $value); // -> header("$key: $valu", $httpCode)
$response->cookie(); // -> setcookie()
$response->status(); // http 狀態碼

當然, swoole 還提供了常用的功能:

$response->sendfile(); // 給客戶端發送文件
$response->gzip(); // nginx + fpm 的場景, nginx 處理掉了這個
$response->end(); // 返回數據給客戶端
$response->write(); // 分段傳輸數據, 最后調用 end() 表示數據傳輸結束

phper 注意下這里的 write() 和 end(), 這里有一個 http chunk 的知識點. 需要返回大量數據給客戶端(>=2M)時, 需要分段(chunk)進行發送. 所以先用 write() 發送數據, 最后用 end() 表示結束. 數據量不大時, 直接調用 end($html) 返回就可以了.

在框架具體實現上, 和上面一樣, laravel 依舊用的 SymfonyResponse, swoft 也是實現 PSR-7 定義的接口, 對 SwooleHttpResponse 進行封裝.

swoft 使用 SwooleServer 來實現 RPC 服務, 其實在上面的多端口監聽, 也是為了開啟 RPC 服務. 注意一下多帶帶啟用中回調函數的區別:

// SwoftServerRpcServer
public function start()
{
    // rpc server
    $this->server = new Server($this->tcpSetting["host"], $this->tcpSetting["port"], $this->tcpSetting["model"], $this->tcpSetting["type"]);

    // 設置回調函數
    $listenSetting = $this->getListenTcpSetting();
    $setting = array_merge($this->setting, $listenSetting);
    $this->server->set($setting);
    $this->server->on("start", [$this, "onStart"]);
    $this->server->on("workerStart", [$this, "onWorkerStart"]);
    $this->server->on("managerStart", [$this, "onManagerStart"]);
    $this->server->on("task", [$this, "onTask"]);
    $this->server->on("finish", [$this, "onFinish"]);
    $this->server->on("connect", [$this, "onConnect"]);
    $this->server->on("receive", [$this, "onReceive"]);
    $this->server->on("pipeMessage", [$this, "onPipeMessage"]); // 接收管道信息時觸發的回調函數
    $this->server->on("close", [$this, "onClose"]);

    // before start
    $this->beforeStart();
    $this->server->start();
}

swoole 自帶的協程的客戶端, swoft 都封裝進了連接池, 用來提高性能. 同時, 為了業務使用方便, 既有協程連接, 也有同步連接, 方便業務使用時無縫切換.

同步/協程連接的實現代碼:

// RedisConnect -> 使用 swoole 協程客戶端
public function createConnect()
{
    // 連接信息
    $timeout = $this->connectPool->getTimeout();
    $address = $this->connectPool->getConnectAddress();
    list($host, $port) = explode(":", $address);

    // 創建連接
    $redis = new SwooleCoroutineRedis();
    $result = $redis->connect($host, $port, $timeout);
    if ($result == false) {
        App::error("redis連接失敗，host=" . $host . " port=" . $port . " timeout=" . $timeout);
        return;
    }

    $this->connect = $redis;
}

// SyncRedisConnect -> 使用 Redis 同步客戶端
public function createConnect()
{
    // 連接信息
    $timeout = $this->connectPool->getTimeout();
    $address = $this->connectPool->getConnectAddress();
    list($host, $port) = explode(":", $address);

    // 初始化連接
    $redis = new Redis();
    $redis->connect($host, $port, $timeout);
    $this->connect = $redis;
}

swoft 中實現連接池的代碼在 src/Pool 下實現, 由三部分組成:

Connect: 即上面代碼中的連接

Balancer: 負載均衡器, 目前實現了 隨機/輪詢 2 種方式

Pool: 連接池, 調用 Balancer, 返回 Connect

詳細內容可以參考之前的 blog - swoft 源碼解讀

作為首個使用 Swoole2.0 原生協程的框架, swoft 希望將協程的能力擴展到框架的核心設計中. 使用 SwoftBaseCoroutine 進行封裝, 方便整個應用中使用:

public static function id()
{
    $cid = SwCoroutine::getuid(); // swoole 協程
    $context = ApplicationContext::getContext();

    if ($context == ApplicationContext::WORKER || $cid !== -1) {
        return $cid;
    }
    if ($context == ApplicationContext::TASK) {
        return Task::getId();
    }
    if($context == ApplicationContext::CONSOLE){
        return Console::id();
    }

    return Process::getId();
}

如同這段代碼所示, Swoft 希望將方便易用的協程的能力, 擴展到 Console/Worker/Task/Process 等等不同的應用場景中

原生的 call_user_func() / call_user_func_array() 中無法使用協程 client, 所以 swoole 在協程組件中也封裝的了相應的實現, swoft 中也有使用到, 請自行閱讀源碼.

進程管理模塊, 適合處理和 Server 比較獨立的常駐進程任務, 在 swoft 中, 在以下場景中使用到:

協程定時器 CronTimerProcess

協程執行命令 CronExecProcess

熱更新進程 ReloadProcess

swoft 使用 SwoftProcess 對 SwooleProcess 進行了封裝:

// SwoftProcess
public static function create(
    AbstractServer $server,
    string $processName,
    string $processClassName
) {
    ...

    // 創建進程
    $process = new SwooleProcess(function (SwooleProcess $process) use ($processClass, $processName) {
        // reload
        BeanFactory::reload();
        $initApplicationContext = new InitApplicationContext();
        $initApplicationContext->init();

        App::trigger(AppEvent::BEFORE_PROCESS, null, $processName, $process, null);
        PhpHelper::call([$processClass, "run"], [$process]);
        App::trigger(AppEvent::AFTER_PROCESS);
    }, $iout, $pipe); // 啟動 SwooleProcess 并綁定回調函數即可

    return $process;
}

swoft 在日志場景下使用 SwooleAsync 來提高性能, 同時保留了原有的同步方式, 方便進行切換

// SwoftLogFileHandler
private function aysncWrite(string $logFile, string $messageText)
{
    while (true) {
        $result = SwooleAsync::writeFile($logFile, $messageText, null, FILE_APPEND); // 使用起來很簡單
        if ($result == true) {
            break;
        }
    }
}

服務器出于性能考慮, 通常都是 常駐內存 的, 傳統的 php-fpm 也是, 修改了配置需要 reload 服務器才能生效. 也因為此, 服務器領域出現了新的需求 -- 熱更新. swoole 在進程管理上已經做了很多優化, 這里摘抄部分 wiki 內容:

Swoole提供了柔性終止/重啟的機制
SIGTERM: 向主進程/管理進程發送此信號服務器將安全終止
SIGUSR1: 向主進程/管理進程發送SIGUSR1信號，將平穩地restart所有worker進程

目前大家采用的, 比較常見的方案, 是基于 Linux Inotify 特性, 通過監測文件變更來觸發 swoole server reload. PHP 中有 Inotify 擴展, 方便使用, 具體實現在 SwoftBaseInotify 中:

public function run()
{
    $inotify = inotify_init();

    // 設置為非阻塞
    stream_set_blocking($inotify, 0);

    $tempFiles = [];
    $iterator = new RecursiveDirectoryIterator($this->watchDir);
    $files = new RecursiveIteratorIterator($iterator);
    foreach ($files as $file) {
        $path = dirname($file);

        // 只監聽目錄
        if (!isset($tempFiles[$path])) {
            $wd = inotify_add_watch($inotify, $path, IN_MODIFY | IN_CREATE | IN_IGNORED | IN_DELETE);
            $tempFiles[$path] = $wd;
            $this->watchFiles[$wd] = $path;
        }
    }

    // swoole Event add
    $this->addSwooleEvent($inotify);
}

private function addSwooleEvent($inotify)
{
    // swoole Event add
    Event::add($inotify, function ($inotify) { // 使用 SwooleEvent
        // 讀取有事件變化的文件
        $events = inotify_read($inotify);
        if ($events) {
            $this->reloadFiles($inotify, $events);
        }
    }, null, SWOOLE_EVENT_READ);
}

swoft 在 CircuitBreaker(熔斷器) 中的 HalfOpenState(半開狀態) 使用到了, 并且這塊的實現比較復雜, 推薦閱讀源碼:

// CircuitBreaker
public function init()
{
    // 狀態初始化
    $this->circuitState = new CloseState($this);
    $this->halfOpenLock = new SwooleLock(SWOOLE_MUTEX); // 初始化互斥鎖
}

// HalfOpenState
public function doCall($callback, $params = [], $fallback = null)
{
    // 加鎖
    $lock = $this->circuitBreaker->getHalfOpenLock();
    $lock->lock();
    list($class ,$method) = $callback;

    ....

    // 釋放鎖
    $lock->unlock();

    ...
}

鎖的使用, 難點主要在了解各種不同鎖使用的場景, 目前 swoole 支持:

文件鎖 SWOOLE_FILELOCK

讀寫鎖 SWOOLE_RWLOCK

信號量 SWOOLE_SEM

互斥鎖 SWOOLE_MUTEX

自旋鎖 SWOOLE_SPINLOCK

定時器基本都會使用到, phper 用的比較多的應該是 crontab 了. 基于這個考慮, swoft 對 Timer 進行了封裝, 方便 phper 用 熟悉的姿勢 繼續使用.

swoft 對 SwooleTimer 進行了簡單的封裝, 代碼在 BaseTimer 中:

// 設置定時器
public function addTickTimer(string $name, int $time, $callback, $params = [])
{
    array_unshift($params, $name, $callback);

    $tid = SwooleTimer::tick($time, [$this, "timerCallback"], $params);

    $this->timers[$name][$tid] = $tid;

    return $tid;
}

// 清除定時器
public function clearTimerByName(string $name)
{
    if (!isset($this->timers[$name])) {
        return true;
    }
    foreach ($this->timers[$name] as $tid => $tidVal) {
        SwooleTimer::clear($tid);
    }
    unset($this->timers[$name]);

    return true;
}

SwooleTable 是在內存中開辟一塊區域, 實現類似關系型數據庫表(Table)這樣的數據結構, 關于 SwooleTable 的實現原理, rango 寫過專門的文章 swoole_table 實現原理剖析, 推薦閱讀.

SwooleTable 在使用上需要注意以下幾點:

類似關系型數據庫, 需要提前定義好 表結構

需要預先判斷數據的大小(行數)

注意內存, swoole 會更根據上面 2 個定義, 在調用 SwooleTable->create() 時分配掉這些內存

swoft 中則是使用這一功能, 來實現 crontab 方式的任務調度:

private $originTable;
private $runTimeTable;

private $originStruct = [
    "rule"       => [SwooleTable::TYPE_STRING, 100],
    "taskClass"  => [SwooleTable::TYPE_STRING, 255],
    "taskMethod" => [SwooleTable::TYPE_STRING, 255],
    "add_time"   => [SwooleTable::TYPE_STRING, 11],
];

private $runTimeStruct = [
    "taskClass"  => [SwooleTable::TYPE_STRING, 255],
    "taskMethod" => [SwooleTable::TYPE_STRING, 255],
    "minte"      => [SwooleTable::TYPE_STRING, 20],
    "sec"        => [SwooleTable::TYPE_STRING, 20],
    "runStatus"  => [SwooleTABLE::TYPE_INT, 4],
];

// 使用 SwooleTable
private function createOriginTable(): bool
{
    $this->setOriginTable(new SwooleTable("origin", self::TABLE_SIZE, $this->originStruct));

    return $this->getOriginTable()->create();
}

老生常談了, 很多人吐槽 swoole 坑, 文檔不好. 說句實話, 要敢于直面自己服務器開發能力不足的現實. 我經常提的一句話:

要把 swoole 的 wiki 看 3 遍.

寫這篇 blog 的初衷是給大家介紹一下 swoole 在 swoft 中的應用場景, 幫助大家嘗試進行 swoole 落地. 希望這篇 blog 能對你有所幫助, 也希望你能多多關注 swoole 社區, 關注 swoft 框架, 能感受到服務器開發帶來的樂趣.