PHP 多任務(wù)協(xié)程處理

jayzou 發(fā)布于2019-07-01 10:19 / 902人閱讀

摘要：所以本文將探討多任務(wù)協(xié)程這方面的內(nèi)容。我們僅需在處理前進行檢測即可。方法用于執(zhí)行任務(wù)，方法用于讓調(diào)度程序知道何時終止運行。它是一種先進先出數(shù)據(jù)結(jié)構(gòu)，能夠確保每個任務(wù)都能夠獲取足夠的處理時間。

本文首發(fā)于 PHP 多任務(wù)協(xié)程處理,轉(zhuǎn)載請注明出處！

上周有幸和同事一起在 SilverStripe 分享最近的工作事宜。今天我計劃分享 PHP 異步編程，不過由于上周我聊過 ReactPHP；我決定討論一些不一樣的內(nèi)容。所以本文將探討多任務(wù)協(xié)程這方面的內(nèi)容。

另外我還計劃把這個主題加入到我正在籌備的一本 PHP 異步編程的圖書中。雖然這本書相比本文來說會涉及更多細節(jié)，但我覺得本文依然具有實際意義！

那么，開始吧！
new MyIterator(

這就是本文我們要討論的問題。不過我們會從更簡單更熟悉的示例開始。

一切從數(shù)組開始

我們可以通過簡單的遍歷來使用數(shù)組：

$array = ["foo", "bar", "baz"];
 
foreach ($array as $key => $value) {
    print "item: " . $key . "|" . $value . "
";
}
 
for ($i = 0; $i < count($array); $i++) {
    print "item: " . $i . "|" . $array[$i] . "
";
}

這是我們?nèi)粘＞幋a所依賴的基本實現(xiàn)?？梢酝ㄟ^遍歷數(shù)組獲取每個元素的鍵名和鍵值。

當然，如果我們希望能夠知道在何時可以使用數(shù)組。PHP 提供了一個方便的內(nèi)置函數(shù):

print is_array($array) ? "yes" : "no"; // yes

類數(shù)組處理

有時，我們需要對一些數(shù)據(jù)使用相同的方式進行遍歷處理，但它們并非數(shù)組類型。比如對 DOMDocument 類進行處理：

$document = new DOMDocument();
$document->loadXML("");

$elements = $document->getElementsByTagName("div");
print_r($elements); // DOMNodeList Object ( [length] => 1 )

這顯然不是一個數(shù)組，但是它有一個 length 屬性。我們能像遍歷數(shù)組一樣，對其進行遍歷么？我們可以判斷它是否實現(xiàn)了下面這個特殊的接口：

print ($elements instanceof Traversable) ? "yes" : "no"; // yes

這真的太有用了。它不會導致我們在遍歷非可遍歷數(shù)據(jù)時觸發(fā)錯誤。我們僅需在處理前進行檢測即可。

不過，這會引發(fā)另外一個問題：我們能否讓自定義類也擁有這個功能呢？回答是肯定的！第一個實現(xiàn)方法類似如下：

class MyTraversable implements Traversable
{
    //  在這里編碼...
}

如果我們執(zhí)行這個類，我們將看到一個錯誤信息：

PHP Fatal error: Class MyTraversable must implement interface Traversable as part of either Iterator or IteratorAggregate

Iterator（迭代器）

我們無法直接實現(xiàn) Traversable，但是我們可以嘗試第二種方案：

class MyTraversable implements Iterator
{
    //  在這里編碼...
}

這個接口需要我們實現(xiàn) 5 個方法。讓我們完善我們的迭代器：

class MyTraversable implements Iterator
{
    protected $data;

    protected $index = 0;

    public function __construct($data)
    {
        $this->data = $data;
    }

    public function current()
    {
        return $this->data[$this->index];
    }

    public function next()
    {
        return $this->data[$this->index++];
    }

    public function key()
    {
        return $this->index;
    }

    public function rewind()
    {
        $this->index = 0;
    }

    public function valid()
    {
        return $this->index < count($this->data);
    }
}

這邊我們需要注意幾個事項：

我們需要存儲構(gòu)造器方法傳入的 $data 數(shù)組，以便后續(xù)我們可以從中獲取它的元素。

還需要一個內(nèi)部索引（或指針）來跟蹤 current 或 next 元素。

rewind() 僅僅重置 index 屬性，這樣 current() 和 next() 才能正常工作。

鍵名并非只能是數(shù)字類型！這里使用數(shù)組索引是為了保證示例足夠簡單。

我們可以向下面這樣運行這段代碼：

$iterator = new MyTraversable(["foo", "bar", "baz"]);
 
foreach ($iterator as $key => $value) {
    print "item: " . $key . "|" . $value . "
";
}

這看起來需要處理太多工作，但是這是能夠像數(shù)組一樣使用 foreach/for 功能的一個簡潔實現(xiàn)。

IteratorAggregate（聚合迭代器）

還記得第二個接口拋出的 Traversable 異常么？下面看一個比實現(xiàn) Iterator 接口更快的實現(xiàn)吧：

class MyIteratorAggregate implements IteratorAggregate
{
    protected $data;

    public function __construct($data)
    {
        $this->data = $data;
    }

    public function getIterator()
    {
        return new ArrayIterator($this->data);
    }
}

這里我們作弊了。相比于實現(xiàn)一個完整的 Iterator，我們通過 ArrayIterator() 裝飾。不過，這相比于通過實現(xiàn)完整的 Iterator 簡化了不少代碼。

兄弟莫急！先讓我們比較一些代碼。首先，我們在不使用生成器的情況下從文件中讀取每一行數(shù)據(jù)：

$content = file_get_contents(__FILE__);

$lines = explode("
", $content);

foreach ($lines as $i => $line) {
    print $i . ". " . $line . "
";
}

這段代碼讀取文件自身，然后會打印出每行的行號和代碼。那么為什么我們不使用生成器呢！

function lines($file) {
    $handle = fopen($file, "r");

    while (!feof($handle)) {
        yield trim(fgets($handle));
    }

    fclose($handle);
}

foreach (lines(__FILE__) as $i => $line) {
    print $i . ". " . $line . "
";
}

我知道這看起來更加復雜。不錯，不過這是因為我們沒有使用 file_get_contents() 函數(shù)。一個生成器看起來就像是一個函數(shù)，但是它會在每次獲取到 yield 關(guān)鍵詞是停止運行。

生成器看起來有點像迭代器：

print_r(lines(__FILE__)); // Generator Object ( )

盡管它不是迭代器，它是一個 Generator。它的內(nèi)部定義了什么方法呢？

print_r(get_class_methods(lines(__FILE__)));
 
// Array
// (
//     [0] => rewind
//     [1] => valid
//     [2] => current
//     [3] => key
//     [4] => next
//     [5] => send
//     [6] => throw
//     [7] => __wakeup
// )

如果你讀取一個大文件，然后使用 memory_get_peak_usage()，你會注意到生成器的代碼會使用固定的內(nèi)存，無論這個文件有多大。它每次進度去一行。而是用 file_get_contents() 函數(shù)讀取整個文件，會使用更大的內(nèi)存。這就是在迭代處理這類事物時，生成器的能給我們帶來的優(yōu)勢！

Send（發(fā)送數(shù)據(jù)）

可以將數(shù)據(jù)發(fā)送到生成器中。看下下面這個生成器：

current() . "
"; // foo

注意這里我們?nèi)绾卧?call_user_func() 函數(shù)中封裝生成器函數(shù)的？這里僅僅是一個簡單的函數(shù)定義，然后立即調(diào)用它獲取一個新的生成器實例...

我們已經(jīng)見過 yield 的用法。我們可以通過擴展這個生成器來接收數(shù)據(jù)：

$generator = call_user_func(function() {
    $input = (yield "foo");

    print "inside: " . $input . "
";
});

print $generator->current() . "
";

$generator->send("bar");

數(shù)據(jù)通過 yield 關(guān)鍵字傳入和返回。首先，執(zhí)行 current() 代碼直到遇到 yield，返回 foo。send() 將輸出傳入到生成器打印輸入的位置。你需要習慣這種用法。

拋出異常（Throw）

由于我們需要同這些函數(shù)進行交互，可能希望將異常推送到生成器中。這樣這些函數(shù)就可以自行處理異常。

看看下面這個示例：

$multiply = function($x, $y) {
    yield $x * $y;
};

print $multiply(5, 6)->current(); // 30

現(xiàn)在讓我們將它封裝到另一個函數(shù)中：

$calculate = function ($op, $x, $y) use ($multiply) {
    if ($op === "multiply") {
        $generator = $multiply($x, $y);

        return $generator->current();
    }
};

print $calculate("multiply", 5, 6); // 30

這里我們通過一個普通閉包將乘法生成器封裝起來?，F(xiàn)在讓我們驗證無效參數(shù)：

$calculate = function ($op, $x, $y) use ($multiply) {

    if ($op === "multiply") {
        $generator = $multiply($x, $y);

        if (!is_numeric($x) || !is_numeric($y)) {
            throw new InvalidArgumentException();
        }

        return $generator->current();
    }
};

print $calculate("multiply", 5, "foo"); // PHP Fatal error...

如果我們希望能夠通過生成器處理異常？我們怎樣才能將異常傳入生成器呢！

$multiply = function ($x, $y) {
    try {
        yield $x * $y;
    } catch (InvalidArgumentException $exception) {
        print "ERRORS!";
    }
};

$calculate = function ($op, $x, $y) use ($multiply) {

    if ($op === "multiply") {
        $generator = $multiply($x, $y);

        if (!is_numeric($x) || !is_numeric($y)) {
            $generator->throw(new InvalidArgumentException());
        }

        return $generator->current();
    }
};
print $calculate("multiply", 5, "foo"); // PHP Fatal error...

棒呆了！我們不僅可以像迭代器一樣使用生成器。還可以通過它們發(fā)送數(shù)據(jù)并拋出異常。它們是可中斷和可恢復的函數(shù)。有些語言把這些函數(shù)叫做……

我們可以使用協(xié)程（coroutines）來構(gòu)建異步代碼。讓我們來創(chuàng)建一個簡單的任務(wù)調(diào)度程序。首先我們需要一個 Task 類：

class Task
{
    protected $generator;

    public function __construct(Generator $generator)
    {
        $this->generator = $generator;
    }

    public function run()
    {
        $this->generator->next();
    }

    public function finished()
    {
        return !$this->generator->valid();
    }
}

Task 是普通生成器的裝飾器。我們將生成器賦值給它的成員變量以供后續(xù)使用，然后實現(xiàn)一個簡單的 run() 和 finished() 方法。run() 方法用于執(zhí)行任務(wù)，finished() 方法用于讓調(diào)度程序知道何時終止運行。

然后我們需要一個 Scheduler 類：

class Scheduler
{
    protected $queue;

    public function __construct()
    {
        $this->queue = new SplQueue();
    }

    public function enqueue(Task $task)
    {
        $this->queue->enqueue($task);
    }

    pulic function run()
    {
        while (!$this->queue->isEmpty()) {
            $task = $this->queue->dequeue();
            $task->run();

            if (!$task->finished()) {
                $this->queue->enqueue($task);
            }
        }
    }
}

Scheduler 用于維護一個待執(zhí)行的任務(wù)隊列。run() 會彈出隊列中的所有任務(wù)并執(zhí)行它，直到運行完整個隊列任務(wù)。如果某個任務(wù)沒有執(zhí)行完畢，當這個任務(wù)本次運行完成后，我們將再次入列。

SplQueue 對于這個示例來講再合適不過了。它是一種 FIFO（先進先出：fist in first out） 數(shù)據(jù)結(jié)構(gòu)，能夠確保每個任務(wù)都能夠獲取足夠的處理時間。

我們可以像這樣運行這段代碼：

$scheduler = new Scheduler();

$task1 = new Task(call_user_func(function() {
    for ($i = 0; $i < 3; $i++) {
        print "task1: " . $i . "
";
        yield;
    }
}));

$task2 = new Task(call_user_func(function() {
    for ($i = 0; $i < 6; $i++) {
        print "task2: " . $i . "
";
        yield;
    }
}));

$scheduler->enqueue($task1);
$scheduler->enqueue($task2);

$scheduler->run();

運行時，我們將看到如下執(zhí)行結(jié)果：

task 1: 0
task 1: 1
task 2: 0
task 2: 1
task 1: 2
task 2: 2
task 2: 3
task 2: 4
task 2: 5

這幾乎就是我們想要的執(zhí)行結(jié)果。不過有個問題發(fā)生在首次運行每個任務(wù)時，它們都執(zhí)行了兩次。我們可以對 Task 類稍作修改來修復這個問題：

class Task
{
    protected $generator;

    protected $run = false;

    public function __construct(Generator $generator)
    {
        $this->generator = $generator;
    }

    public function run()
    {
        if ($this->run) {
            $this->generator->next();
        } else {
            $this->generator->current();
        }

        $this->run = true;
    }

    public function finished()
    {
        return !$this->generator->valid();
    }
}

我們需要調(diào)整首次 run() 方法調(diào)用，從生成器當前有效的指針讀取運行。后續(xù)調(diào)用可以從下一個指針讀取運行...

有些人基于這個思路實現(xiàn)了一些超贊的類庫。我們來看看其中的兩個...

RecoilPHP

RecoilPHP 是一套基于協(xié)程的類庫，它最令人印象深刻的是用于 ReactPHP 內(nèi)核?？梢詫⑹录h(huán)在 RecoilPHP 和 RecoilPHP 之間進行交換，而你的程序無需架構(gòu)上的調(diào)整。

我們來看一下 ReactPHP 異步 DNS 解決方案：

function resolve($domain, $resolver) {
    $resolver
        ->resolve($domain)
        ->then(function ($ip) use ($domain) {
            print "domain: " . $domain . "
";
            print "ip: " . $ip . "
";
        }, function ($error) {            
            print $error . "
";
        })
}

function run()
{
    $loop = ReactEventLoopFactory::create();
 
    $factory = new ReactDnsResolverFactory();
 
    $resolver = $factory->create("8.8.8.8", $loop);
 
    resolve("silverstripe.org", $resolver);
    resolve("wordpress.org", $resolver);
    resolve("wardrobecms.com", $resolver);
    resolve("pagekit.com", $resolver);
 
    $loop->run();
}
 
run();

resolve() 接收域名和 DNS 解析器，并使用 ReactPHP 執(zhí)行標準的 DNS 查找。不用太過糾結(jié)與 resolve() 函數(shù)內(nèi)部。重要的是這個函數(shù)不是生成器，而是一個函數(shù)！

run() 創(chuàng)建一個 ReactPHP 事件循環(huán)，DNS 解析器（這里是個工廠實例）解析若干域名。同樣，這個也不是一個生成器。

想知道 RecoilPHP 到底有何不同？還希望掌握更多細節(jié)！

use RecoilRecoil;
 
function resolve($domain, $resolver)
{
    try {
        $ip = (yield $resolver->resolve($domain));
 
        print "domain: " . $domain . "
";
        print "ip: " . $ip . "
";
    } catch (Exception $exception) {
        print $exception->getMessage() . "
";
    }
}
 
function run()
{
    $loop = (yield Recoil::eventLoop());
 
    $factory = new ReactDnsResolverFactory();
 
    $resolver = $factory->create("8.8.8.8", $loop);
 
    yield [
        resolve("silverstripe.org", $resolver),
        resolve("wordpress.org", $resolver),
        resolve("wardrobecms.com", $resolver),
        resolve("pagekit.com", $resolver),
    ];
}
 
Recoil::run("run");

通過將它集成到 ReactPHP 來完成一些令人稱奇的工作。每次運行 resolve() 時，RecoilPHP 會管理由 $resoler->resolve() 返回的 promise 對象，然后將數(shù)據(jù)發(fā)送給生成器。此時我們就像在編寫同步代碼一樣。與我們在其他一步模型中使用回調(diào)代碼不同，這里只有一個指令列表。

RecoilPHP 知道它應(yīng)該管理一個有執(zhí)行 run() 函數(shù)時返回的 yield 數(shù)組。RoceilPHP 還支持基于協(xié)程的數(shù)據(jù)庫（PDO）和日志庫。

IcicleIO

IcicleIO 為了一全新的方案實現(xiàn) ReactPHP 一樣的目標，而僅僅使用協(xié)程功能。相比 ReactPHP 它僅包含極少的組件。但是，核心的異步流、服務(wù)器、Socket、事件循環(huán)特性一個不落。

讓我們看一個 socket 服務(wù)器示例：

use IcicleCoroutineCoroutine;
use IcicleLoopLoop;
use IcicleSocketClientClientInterface;
use IcicleSocketServerServerInterface;
use IcicleSocketServerServerFactory;
 
$factory = new ServerFactory();
 
$coroutine = Coroutine::call(function (ServerInterface $server) {
    $clients = new SplObjectStorage();
     
    $handler = Coroutine::async(
        function (ClientInterface $client) use (&$clients) {
            $clients->attach($client);
             
            $host = $client->getRemoteAddress();
            $port = $client->getRemotePort();
             
            $name = $host . ":" . $port;
             
            try {
                foreach ($clients as $stream) {
                    if ($client !== $stream) {
                        $stream->write($name . "connected.
");
                    }
                }
 
                yield $client->write("Welcome " . $name . "!
");
                 
                while ($client->isReadable()) {
                    $data = trim(yield $client->read());
                     
                    if ("/exit" === $data) {
                        yield $client->end("Goodbye!
");
                    } else {
                        $message = $name . ":" . $data . "
";
                        
                        foreach ($clients as $stream) {
                            if ($client !== $stream) {
                                $stream->write($message);
                            }
                        }
                    }
                }
            } catch (Exception $exception) {
                $client->close($exception);
            } finally {
                $clients->detach($client);
                foreach ($clients as $stream) {
                    $stream->write($name . "disconnected.
");
                }
            }
        }
    );
     
    while ($server->isOpen()) {
        $handler(yield $server->accept());
    }
}, $factory->create("127.0.0.1", 6000));
 
Loop::run();

據(jù)我所知，這段代碼所做的事情如下:

在 127.0.0.1 和 6000 端口創(chuàng)建一個服務(wù)器實例，然后將其傳入外部生成器.

外部生成器運行，同時服務(wù)器等待新連接。當服務(wù)器接收一個連接它將其傳入內(nèi)部生成器。

內(nèi)部生成器寫入消息到 socket。當 socket 可讀時運行。

每次 socket 向服務(wù)器發(fā)送消息時，內(nèi)部生成器檢測消息是否是退出標識。如果是，通知其他 socket。否則，其它 socket 發(fā)送這個相同的消息。

打開命令行終端輸入 nc localhost 6000 查看執(zhí)行結(jié)果！

該示例使用 SplObjectStorage 跟蹤 socket 連接。這樣我們就可以向所有 socket 發(fā)送消息。

這個話題可以包含很多內(nèi)容。希望您能看到生成器是如何創(chuàng)建的，以及它們?nèi)绾螏椭帉懙绦蚝彤惒酱a。

如果你有問題，可以隨時問我。

感謝 Nikita Popov（還有它的啟蒙教程 Cooperative multitasking using coroutines (in PHP!) ）,Anthony Ferrara 和 Joe Watkins。這些研究工作澤被蒼生，給我以寫作此篇文章的靈感。關(guān)注他們吧，好么？

原文

Co-operative PHP Multitasking

文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/29045.html

發(fā)表評論

登陸后可評論

0條評論

jayzou

男|高級講師

我要關(guān)注我要私信

TA的文章

#yyds干貨盤點#Java ASM系列：（093）反編譯-方法參數(shù)

閱讀 1937·2021-11-24 09:39
《正規(guī)軍的Python進階之路｜Python技能樹測評》

閱讀 3522·2021-09-28 09:36
ZJI：香港阿里云專線服務(wù)器月減270-400元,E5服務(wù)器月付480元起

閱讀 3291·2021-09-06 15:10
小程序自定義導航欄（適配所有手機）

閱讀 3446·2019-08-30 15:44
css零散小技巧

閱讀 1159·2019-08-30 15:43
學習小結(jié)

閱讀 1802·2019-08-30 14:20
css創(chuàng)建叉和勾

閱讀 2719·2019-08-30 12:51
重學前端學習筆記（二十七）--JavaScript的詞法

閱讀 2038·2019-08-30 11:04

国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

PHP 多任務(wù)協(xié)程處理

相關(guān)文章

***PHP下的異步嘗試二：初識協(xié)程***

**Mix PHP V2 實例：協(xié)程池異步郵件發(fā)送守護程序**

**關(guān)于PHP協(xié)程與阻塞的思考**

***PHP回顧之協(xié)程***

**IMI 基于 Swoole 開發(fā)的協(xié)程 PHP 開發(fā)框架常駐內(nèi)存、協(xié)程異步非阻塞**

發(fā)表評論

0條評論

jayzou

男|高級講師

TA的文章

#yyds干貨盤點#Java ASM系列：（093）反編譯-方法參數(shù)

《正規(guī)軍的Python進階之路｜Python技能樹測評》

ZJI：香港阿里云專線服務(wù)器月減270-400元,E5服務(wù)器月付480元起

小程序自定義導航欄（適配所有手機）

css零散小技巧

學習小結(jié)

css創(chuàng)建叉和勾

重學前端學習筆記（二十七）--JavaScript的詞法

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

PHP 多任務(wù)協(xié)程處理

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！