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

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

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

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

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

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

那么，開始吧！
new MyIterator(

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

一切從數(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所依賴的基本實(shí)現(xiàn)。可以通過遍歷數(shù)組獲取每個(gè)元素的鍵名和鍵值。

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

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

類數(shù)組處理

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

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

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

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

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

這真的太有用了。它不會(huì)導(dǎo)致我們?cè)诒闅v非可遍歷數(shù)據(jù)時(shí)觸發(fā)錯(cuò)誤。我們僅需在處理前進(jìn)行檢測即可。

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

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

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

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

Iterator（迭代器）

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

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

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

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è)事項(xiàng)：

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

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

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

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

我們可以向下面這樣運(yùn)行這段代碼：

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

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

IteratorAggregate（聚合迭代器）

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

class MyIteratorAggregate implements IteratorAggregate
{
    protected $data;

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

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

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

兄弟莫急！先讓我們比較一些代碼。首先，我們?cè)诓皇褂蒙善鞯那闆r下從文件中讀取每一行數(shù)據(jù)：

$content = file_get_contents(__FILE__);

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

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

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

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 . "
";
}

我知道這看起來更加復(fù)雜。不錯(cuò)，不過這是因?yàn)槲覀儧]有使用 file_get_contents() 函數(shù)。一個(gè)生成器看起來就像是一個(gè)函數(shù)，但是它會(huì)在每次獲取到 yield 關(guān)鍵詞是停止運(yùn)行。

生成器看起來有點(diǎn)像迭代器：

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

盡管它不是迭代器，它是一個(gè) 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
// )

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

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

可以將數(shù)據(jù)發(fā)送到生成器中?？聪孪旅孢@個(gè)生成器：

current() . "
"; // foo

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

我們已經(jīng)見過 yield 的用法。我們可以通過擴(kuò)展這個(gè)生成器來接收數(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() 將輸出傳入到生成器打印輸入的位置。你需要習(xí)慣這種用法。

拋出異常（Throw）

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

看看下面這個(gè)示例：

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

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

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

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

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

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

這里我們通過一個(gè)普通閉包將乘法生成器封裝起來?，F(xiàn)在讓我們驗(yà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...

如果我們希望能夠通過生成器處理異常？我們?cè)鯓硬拍軐惓魅肷善髂兀?/p>

$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ù)并拋出異常。它們是可中斷和可恢復(fù)的函數(shù)。有些語言把這些函數(shù)叫做……

我們可以使用協(xié)程（coroutines）來構(gòu)建異步代碼。讓我們來創(chuàng)建一個(gè)簡單的任務(wù)調(diào)度程序。首先我們需要一個(gè) 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ù)使用，然后實(shí)現(xiàn)一個(gè)簡單的 run() 和 finished() 方法。run() 方法用于執(zhí)行任務(wù)，finished() 方法用于讓調(diào)度程序知道何時(shí)終止運(yùn)行。

然后我們需要一個(gè) 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 用于維護(hù)一個(gè)待執(zhí)行的任務(wù)隊(duì)列。run() 會(huì)彈出隊(duì)列中的所有任務(wù)并執(zhí)行它，直到運(yùn)行完整個(gè)隊(duì)列任務(wù)。如果某個(gè)任務(wù)沒有執(zhí)行完畢，當(dāng)這個(gè)任務(wù)本次運(yùn)行完成后，我們將再次入列。

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

我們可以像這樣運(yùn)行這段代碼：

$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();

運(yùn)行時(shí)，我們將看到如下執(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é)果。不過有個(gè)問題發(fā)生在首次運(yùn)行每個(gè)任務(wù)時(shí)，它們都執(zhí)行了兩次。我們可以對(duì) Task 類稍作修改來修復(fù)這個(gè)問題：

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)用，從生成器當(dāng)前有效的指針讀取運(yùn)行。后續(xù)調(diào)用可以從下一個(gè)指針讀取運(yùn)行...

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

RecoilPHP

RecoilPHP 是一套基于協(xié)程的類庫，它最令人印象深刻的是用于 ReactPHP 內(nèi)核?？梢詫⑹录h(huán)在 RecoilPHP 和 RecoilPHP 之間進(jìn)行交換，而你的程序無需架構(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í)行標(biāo)準(zhǔn)的 DNS 查找。不用太過糾結(jié)與 resolve() 函數(shù)內(nèi)部。重要的是這個(gè)函數(shù)不是生成器，而是一個(gè)函數(shù)！

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

想知道 RecoilPHP 到底有何不同？還希望掌握更多細(xì)節(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 來完成一些令人稱奇的工作。每次運(yùn)行 resolve() 時(shí)，RecoilPHP 會(huì)管理由 $resoler->resolve() 返回的 promise 對(duì)象，然后將數(shù)據(jù)發(fā)送給生成器。此時(shí)我們就像在編寫同步代碼一樣。與我們?cè)谄渌徊侥Ｐ椭惺褂没卣{(diào)代碼不同，這里只有一個(gè)指令列表。

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

IcicleIO

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

讓我們看一個(gè) 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)建一個(gè)服務(wù)器實(shí)例，然后將其傳入外部生成器.

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

內(nèi)部生成器寫入消息到 socket。當(dāng) socket 可讀時(shí)運(yùn)行。

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

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

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

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

如果你有問題，可以隨時(shí)問我。

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

原文

Co-operative PHP Multitasking

GPU云服務(wù)器云服務(wù)器 php 處理任務(wù)表 php協(xié)程curl php協(xié)程擴(kuò)展 java任務(wù)處理

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

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

發(fā)表評(píng)論

登陸后可評(píng)論

0條評(píng)論

jayzou

男|高級(jí)講師

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

TA的文章

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

閱讀 2010·2021-11-24 09:39
《正規(guī)軍的Python進(jìn)階之路｜Python技能樹測評(píng)》

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

閱讀 3359·2021-09-06 15:10
小程序自定義導(dǎo)航欄（適配所有手機(jī)）

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

閱讀 1208·2019-08-30 15:43
學(xué)習(xí)小結(jié)

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

閱讀 2779·2019-08-30 12:51
重學(xué)前端學(xué)習(xí)筆記（二十七）--JavaScript的詞法

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

成人无码视频,亚洲精品久久久久av无码,午夜精品久久久久久毛片,亚洲中文字幕日韩无码

資訊專欄INFORMATION COLUMN

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

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

相關(guān)文章

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

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

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

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

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

發(fā)表評(píng)論

0條評(píng)論

jayzou

男|高級(jí)講師

TA的文章

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

《正規(guī)軍的Python進(jìn)階之路｜Python技能樹測評(píng)》

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

小程序自定義導(dǎo)航欄（適配所有手機(jī)）

css零散小技巧

學(xué)習(xí)小結(jié)

css創(chuàng)建叉和勾

重學(xué)前端學(xué)習(xí)筆記（二十七）--JavaScript的詞法

最新活動(dòng)

資訊專欄INFORMATION COLUMN

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

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

相關(guān)文章

發(fā)表評(píng)論

0條評(píng)論

男|高級(jí)講師

TA的文章

最新活動(dòng)

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