摘要：在中的應(yīng)用官網(wǎng)源碼解讀號(hào)外號(hào)外歡迎大家我們開發(fā)組定了一個(gè)就線下聚一次的小目標(biāo)上一篇源碼解讀反響還不錯(cuò)不少同學(xué)推薦再加一篇講解一下中使用到的功能幫助大家開啟的實(shí)戰(zhàn)之旅服務(wù)器開發(fā)涉及到的相關(guān)技術(shù)領(lǐng)域的知識(shí)非常多不日積月累打好基礎(chǔ)是很難真正

date: 2017-12-14 21:34:51
title: swoole 在 swoft 中的應(yīng)用

swoft 官網(wǎng): https://www.swoft.org/
swoft 源碼解讀: http://naotu.baidu.com/file/8...
號(hào)外號(hào)外, 歡迎大家 star, 我們開發(fā)組定了一個(gè) star 1000+ 就線下聚一次的小目標(biāo)

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

服務(wù)器開發(fā)涉及到的相關(guān)技術(shù)領(lǐng)域的知識(shí)非常多, 不日積月累打好基礎(chǔ), 是很難真正做好的. 所以我建議:

swoole wiki 最好看 3 遍, 包括評(píng)論. 第一遍快速過(guò)一遍, 形成大致印象; 第二遍邊看邊敲代碼; 第三遍可以選擇衍生的開源框架進(jìn)行實(shí)戰(zhàn). swoft 就是不錯(cuò)的選擇.

swoole wiki 發(fā)展到現(xiàn)在已經(jīng) 1400+ 頁(yè), 確實(shí)會(huì)有點(diǎn)難啃, 勇敢的少年呀, 加油.

swoole 在 swoft 中的應(yīng)用:

SwooleServer: swoole2.0 協(xié)程 Server

SwooleHttpServer: swoole2.0 協(xié)程 http Server, 繼承自 SwooleServer

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

SwooleCoroutine: 協(xié)程工具集, 獲取當(dāng)前協(xié)程id，反射調(diào)用等能力

SwooleProcess: 進(jìn)程管理模塊, 可以在 SwooleServer 之外擴(kuò)展更多功能

SwooleAsync: 異步文件 IO

SwooleTimer: 基于 timerfd + epoll 實(shí)現(xiàn)的異步毫秒定時(shí)器，可完美的運(yùn)行在 EventLoop 中

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

SwooleLock: 在 PHP 代碼中可以很方便地創(chuàng)建一個(gè)鎖, 用來(lái)實(shí)現(xiàn)數(shù)據(jù)同步

SwooleTable: 基于共享內(nèi)存實(shí)現(xiàn)的超高性能數(shù)據(jù)結(jié)構(gòu)

使用 swoole 的 http server 相較 tcp server 還是要簡(jiǎn)單一些, 只需要關(guān)心:

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"]);

    // 設(shè)置事件監(jiān)聽
    $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"]);

    // 啟動(dòng)RPC服務(wù)
    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 十分簡(jiǎn)單:

傳入配置 server 配置信息, new 一個(gè) swoole server

設(shè)置事件監(jiān)聽, 這一步需要大家對(duì) swoole 的進(jìn)程模型非常熟悉, 一定要看懂下面 2 張圖

啟動(dòng)服務(wù)器

swoft 在使用 http server 時(shí), 還會(huì)根據(jù)配置信息, 來(lái)判斷是否同時(shí)新建一個(gè) RPC server, 使用 swoole 的 多端口監(jiān)聽 來(lái)實(shí)現(xiàn).

再來(lái)看 Request 和 Response, 提醒一下, 框架設(shè)計(jì)的時(shí)候, 要記住 規(guī)范先行:

PSR-7: HTTP message interfaces

phper 比較熟悉的應(yīng)該是 $_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包體

這里強(qiáng)調(diào)一下 $request->rawContent(), phper 可能用 $_POST 比較 6, 導(dǎo)致一些知識(shí)不知道: post 的數(shù)據(jù)的格式. 因?yàn)檫@個(gè)知識(shí), 所以 $_POST 不是所有時(shí)候都能取到數(shù)據(jù)的, 大家可以網(wǎng)上查找資料, 或者自己使用 postman 這樣的工具自己測(cè)試驗(yàn)證一下. 在 $_POST 取不到數(shù)據(jù)的情況下, 會(huì)這樣處理:

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

$request->rawContent() 和這個(gè)等價(jià)的.

swoft 封裝 Request 對(duì)象的方法, 和主流框架差不多, 以 laravel 為例(實(shí)際使用 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 狀態(tài)碼

當(dāng)然, swoole 還提供了常用的功能:

$response->sendfile(); // 給客戶端發(fā)送文件
$response->gzip(); // nginx + fpm 的場(chǎng)景, nginx 處理掉了這個(gè)
$response->end(); // 返回?cái)?shù)據(jù)給客戶端
$response->write(); // 分段傳輸數(shù)據(jù), 最后調(diào)用 end() 表示數(shù)據(jù)傳輸結(jié)束

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

在框架具體實(shí)現(xiàn)上, 和上面一樣, laravel 依舊用的 SymfonyResponse, swoft 也是實(shí)現(xiàn) PSR-7 定義的接口, 對(duì) SwooleHttpResponse 進(jìn)行封裝.

swoft 使用 SwooleServer 來(lái)實(shí)現(xiàn) RPC 服務(wù), 其實(shí)在上面的多端口監(jiān)聽, 也是為了開啟 RPC 服務(wù). 注意一下多帶帶啟用中回調(diào)函數(shù)的區(qū)別:

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

    // 設(shè)置回調(diào)函數(shù)
    $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"]); // 接收管道信息時(shí)觸發(fā)的回調(diào)函數(shù)
    $this->server->on("close", [$this, "onClose"]);

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

swoole 自帶的協(xié)程的客戶端, swoft 都封裝進(jìn)了連接池, 用來(lái)提高性能. 同時(shí), 為了業(yè)務(wù)使用方便, 既有協(xié)程連接, 也有同步連接, 方便業(yè)務(wù)使用時(shí)無(wú)縫切換.

同步/協(xié)程連接的實(shí)現(xiàn)代碼:

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

    // 創(chuàng)建連接
    $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 中實(shí)現(xiàn)連接池的代碼在 src/Pool 下實(shí)現(xiàn), 由三部分組成:

Connect: 即上面代碼中的連接

Balancer: 負(fù)載均衡器, 目前實(shí)現(xiàn)了 隨機(jī)/輪詢 2 種方式

Pool: 連接池, 調(diào)用 Balancer, 返回 Connect

詳細(xì)內(nèi)容可以參考之前的 blog - swoft 源碼解讀

作為首個(gè)使用 Swoole2.0 原生協(xié)程的框架, swoft 希望將協(xié)程的能力擴(kuò)展到框架的核心設(shè)計(jì)中. 使用 SwoftBaseCoroutine 進(jìn)行封裝, 方便整個(gè)應(yīng)用中使用:

public static function id()
{
    $cid = SwCoroutine::getuid(); // swoole 協(xié)程
    $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 希望將方便易用的協(xié)程的能力, 擴(kuò)展到 Console/Worker/Task/Process 等等不同的應(yīng)用場(chǎng)景中

原生的 call_user_func() / call_user_func_array() 中無(wú)法使用協(xié)程 client, 所以 swoole 在協(xié)程組件中也封裝的了相應(yīng)的實(shí)現(xiàn), swoft 中也有使用到, 請(qǐng)自行閱讀源碼.

進(jìn)程管理模塊, 適合處理和 Server 比較獨(dú)立的常駐進(jìn)程任務(wù), 在 swoft 中, 在以下場(chǎng)景中使用到:

協(xié)程定時(shí)器 CronTimerProcess

協(xié)程執(zhí)行命令 CronExecProcess

熱更新進(jìn)程 ReloadProcess

swoft 使用 SwoftProcess 對(duì) SwooleProcess 進(jìn)行了封裝:

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

    // 創(chuàng)建進(jìn)程
    $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); // 啟動(dòng) SwooleProcess 并綁定回調(diào)函數(shù)即可

    return $process;
}

swoft 在日志場(chǎng)景下使用 SwooleAsync 來(lái)提高性能, 同時(shí)保留了原有的同步方式, 方便進(jìn)行切換

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

服務(wù)器出于性能考慮, 通常都是 常駐內(nèi)存 的, 傳統(tǒng)的 php-fpm 也是, 修改了配置需要 reload 服務(wù)器才能生效. 也因?yàn)榇? 服務(wù)器領(lǐng)域出現(xiàn)了新的需求 -- 熱更新. swoole 在進(jìn)程管理上已經(jīng)做了很多優(yōu)化, 這里摘抄部分 wiki 內(nèi)容:

Swoole提供了柔性終止/重啟的機(jī)制
SIGTERM: 向主進(jìn)程/管理進(jìn)程發(fā)送此信號(hào)服務(wù)器將安全終止
SIGUSR1: 向主進(jìn)程/管理進(jìn)程發(fā)送SIGUSR1信號(hào)，將平穩(wěn)地restart所有worker進(jìn)程

目前大家采用的, 比較常見的方案, 是基于 Linux Inotify 特性, 通過(guò)監(jiān)測(cè)文件變更來(lái)觸發(fā) swoole server reload. PHP 中有 Inotify 擴(kuò)展, 方便使用, 具體實(shí)現(xiàn)在 SwoftBaseInotify 中:

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

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

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

        // 只監(jiān)聽目錄
        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(半開狀態(tài)) 使用到了, 并且這塊的實(shí)現(xiàn)比較復(fù)雜, 推薦閱讀源碼:

// CircuitBreaker
public function init()
{
    // 狀態(tài)初始化
    $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();

    ...
}

鎖的使用, 難點(diǎn)主要在了解各種不同鎖使用的場(chǎng)景, 目前 swoole 支持:

文件鎖 SWOOLE_FILELOCK

讀寫鎖 SWOOLE_RWLOCK

信號(hào)量 SWOOLE_SEM

互斥鎖 SWOOLE_MUTEX

自旋鎖 SWOOLE_SPINLOCK

定時(shí)器基本都會(huì)使用到, phper 用的比較多的應(yīng)該是 crontab 了. 基于這個(gè)考慮, swoft 對(duì) Timer 進(jìn)行了封裝, 方便 phper 用 熟悉的姿勢(shì) 繼續(xù)使用.

swoft 對(duì) SwooleTimer 進(jìn)行了簡(jiǎn)單的封裝, 代碼在 BaseTimer 中:

// 設(shè)置定時(shí)器
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;
}

// 清除定時(shí)器
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 是在內(nèi)存中開辟一塊區(qū)域, 實(shí)現(xiàn)類似關(guān)系型數(shù)據(jù)庫(kù)表(Table)這樣的數(shù)據(jù)結(jié)構(gòu), 關(guān)于 SwooleTable 的實(shí)現(xiàn)原理, rango 寫過(guò)專門的文章 swoole_table 實(shí)現(xiàn)原理剖析, 推薦閱讀.

SwooleTable 在使用上需要注意以下幾點(diǎn):

類似關(guān)系型數(shù)據(jù)庫(kù), 需要提前定義好 表結(jié)構(gòu)

需要預(yù)先判斷數(shù)據(jù)的大小(行數(shù))

注意內(nèi)存, swoole 會(huì)更根據(jù)上面 2 個(gè)定義, 在調(diào)用 SwooleTable->create() 時(shí)分配掉這些內(nèi)存

swoft 中則是使用這一功能, 來(lái)實(shí)現(xiàn) crontab 方式的任務(wù)調(diào)度:

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 坑, 文檔不好. 說(shuō)句實(shí)話, 要敢于直面自己服務(wù)器開發(fā)能力不足的現(xiàn)實(shí). 我經(jīng)常提的一句話:

要把 swoole 的 wiki 看 3 遍.

寫這篇 blog 的初衷是給大家介紹一下 swoole 在 swoft 中的應(yīng)用場(chǎng)景, 幫助大家嘗試進(jìn)行 swoole 落地. 希望這篇 blog 能對(duì)你有所幫助, 也希望你能多多關(guān)注 swoole 社區(qū), 關(guān)注 swoft 框架, 能感受到服務(wù)器開發(fā)帶來(lái)的樂(lè)趣.