Revolt是什么？

Revolt是并發(fā)PHP應(yīng)用程序的堅(jiān)如磐石的事件循環(huán)。通常的PHP應(yīng)用程序?qū)⒋蟛糠謺r(shí)間花在等待I/O上。雖然PHP是單線程的，但可以使用協(xié)作多任務(wù)來允許并發(fā)性，方法是使用等待時(shí)間來做不同的事情。

PHP的傳統(tǒng)同步執(zhí)行流程很容易理解。一次只做一件事。如果查詢數(shù)據(jù)庫，則發(fā)送查詢并等待數(shù)據(jù)庫服務(wù)器的響應(yīng)。一旦你有了答案，你就可以開始做下一件事。

ReactPHP和其他庫已經(jīng)在PHP中提供了很長一段時(shí)間的協(xié)作多任務(wù)。然而，它們的事件驅(qū)動(dòng)特性與許多現(xiàn)有的接口不兼容，需要不同的思維模型。PHP 8.1內(nèi)置了fibers，它提供了協(xié)作多線程。調(diào)用可以是異步的，沒有promise或回調(diào)，同時(shí)仍然允許非阻塞I/O。

每個(gè)使用協(xié)同多任務(wù)的應(yīng)用程序都需要一個(gè)調(diào)度器（也稱為事件循環(huán)），這個(gè)包提供了這個(gè)調(diào)度器。Revolt是結(jié)合了React和ReactPHP的事件循環(huán)實(shí)現(xiàn)的多年經(jīng)驗(yàn)的結(jié)果。然而，它并不是一個(gè)用于編寫并發(fā)PHP應(yīng)用程序的成熟框架，而只是提供了必要的公共基礎(chǔ)。不同的（強(qiáng)烈的）固執(zhí)己見的庫可以在它的基礎(chǔ)上構(gòu)建，React和ReactPHP將繼續(xù)共存。

Revolt 支持事件

• Defer 回調(diào)在事件循環(huán)的下一次迭代中執(zhí)行。如果有延遲調(diào)度，事件循環(huán)不會(huì)在迭代之間等待。
• Delay 在指定的秒數(shù)后執(zhí)行回調(diào)。秒的分?jǐn)?shù)可以表示為浮點(diǎn)數(shù)。
• Repeat 在指定的秒數(shù)后重復(fù)執(zhí)行回調(diào)。秒的分?jǐn)?shù)可以表示為浮點(diǎn)數(shù)。
• Stream readable 當(dāng)流上有數(shù)據(jù)要讀取或連接關(guān)閉時(shí)，將執(zhí)行回調(diào)。
• Stream writable 當(dāng)寫緩沖區(qū)中有足夠的空間來接受要寫入的新數(shù)據(jù)時(shí)，就會(huì)執(zhí)行回調(diào)。
• Signal 當(dāng)進(jìn)程從操作系統(tǒng)接收到特定信號(hào)時(shí)執(zhí)行回調(diào)。

安裝

composer require revolt/event-loop

注意：此包可以作為Composer依賴項(xiàng)安裝在PHP 8.1及更高版本上。

示例

<?php

require __DIR__ . '/vendor/autoload.php';

use Revolt\EventLoop;

$suspension = EventLoop::getSuspension();

$repeatId = EventLoop::repeat(1, function (): void {
    print '++ Executing callback created by EventLoop::repeat()' . PHP_EOL;
});

EventLoop::delay(5, function () use ($suspension, $repeatId): void {
    print '++ Executing callback created by EventLoop::delay()' . PHP_EOL;

    EventLoop::cancel($repeatId);
    $suspension->resume(null);

    print '++ Suspension::resume() is async!' . PHP_EOL;
});

print '++ Suspending to event loop...' . PHP_EOL;

$suspension->suspend();

print '++ Script end' . PHP_EOL;

在執(zhí)行上面的例子時(shí)，你應(yīng)該看到這樣的輸出：

++ Suspending to event loop...
++ Executing callback created by EventLoop::repeat()
++ Executing callback created by EventLoop::repeat()
++ Executing callback created by EventLoop::repeat()
++ Executing callback created by EventLoop::repeat()
++ Executing callback created by EventLoop::delay()
++ Suspension::resume() is async!
++ Script end

這個(gè)輸出說明了事件循環(huán)內(nèi)部發(fā)生的事情就像它自己獨(dú)立的程序一樣。您的腳本將不會(huì)繼續(xù)通過 $suspension->suspend() 點(diǎn)，除非掛起點(diǎn)通過 $suspension->resume() 或 $suspension->throw() 恢復(fù)。

雖然一個(gè)應(yīng)用程序可以而且經(jīng)常幾乎完全在事件循環(huán)的范圍內(nèi)發(fā)生，但我們也可以使用事件循環(huán)來做一些事情，比如下面的例子，它為交互式控制臺(tái)輸入施加了一個(gè)短暫的超時(shí)：

<?php

require __DIR__ . '/vendor/autoload.php';

use Revolt\EventLoop;

if (\stream_set_blocking(STDIN, false) !== true) {
    \fwrite(STDERR, "Unable to set STDIN to non-blocking" . PHP_EOL);
    exit(1);
}

print "Write something and hit enter" . PHP_EOL;

$suspension = EventLoop::getSuspension();

$readableId = EventLoop::onReadable(STDIN, function ($id, $stream) use ($suspension): void {
    EventLoop::cancel($id);

    $chunk = \fread($stream, 8192);

    print "Read " . \strlen($chunk) . " bytes" . PHP_EOL;

    $suspension->resume(null);
});

$timeoutId = EventLoop::delay(5, function () use ($readableId, $suspension) {
    EventLoop::cancel($readableId);
    
    print "Timeout reached" . PHP_EOL;

    $suspension->resume(null);
});

$suspension->suspend();

EventLoop::cancel($readableId);
EventLoop::cancel($timeoutId);

顯然，我們可以在這個(gè)例子中簡單地同步使用 fgets(STDIN) 。我們只是在演示可以根據(jù)需要進(jìn)出事件循環(huán)，以混合同步任務(wù)和非阻塞任務(wù)。

Timers 定時(shí)器

事件循環(huán)公開了幾種調(diào)度計(jì)時(shí)器的方法。

Deferred 回調(diào)

• defer() 調(diào)度回調(diào)在事件循環(huán)的下一次迭代中執(zhí)行。
• 此方法保證了一個(gè)干凈的調(diào)用堆棧，以避免循環(huán)的當(dāng)前迭代中其他事件的饑餓。defer()回調(diào)總是在事件循環(huán)的下一個(gè)tick中執(zhí)行。
• 在 defer()計(jì)時(shí)器執(zhí)行之后，它會(huì)被事件循環(huán)自動(dòng)垃圾收集，因此應(yīng)用程序不需要手動(dòng)取消關(guān)聯(lián)的回調(diào)。
• 像所有事件回調(diào)一樣，defer() 計(jì)時(shí)器可以被禁用和重新啟用。如果您在調(diào)度它和它實(shí)際運(yùn)行之間禁用此回調(diào)，則事件循環(huán)將無法對(duì)其進(jìn)行垃圾收集，直到它執(zhí)行為止。因此，如果 defer() 回調(diào)從未真正執(zhí)行以釋放任何相關(guān)資源，則必須手動(dòng)取消該回調(diào)。

案例

<?php
/**
 * @author Tinywan(ShaoBo Wan)
 * @email 756684177@qq.com
 * @date 2024/1/31 18:24
 */

require 'vendor/autoload.php';

use Revolt\EventLoop;

echo "line 1\n";

EventLoop::defer(function (): void {
    echo "line 3\n";
});

echo "line 2\n";

EventLoop::run();

輸出

line 1
line 2
line 3

Delayed 回調(diào)

• delay() 計(jì)劃在延遲 n 秒后執(zhí)行回調(diào)
• delay() 回調(diào)在執(zhí)行后也會(huì)被事件循環(huán)自動(dòng)垃圾回收，應(yīng)用程序不應(yīng)該手動(dòng)取消它，除非他們希望在執(zhí)行前完全放棄回調(diào)。
• 被禁用的 delay() 回調(diào)會(huì)重置其延遲時(shí)間，以便重新啟用后，原始延遲時(shí)間再次從零開始。
• 與 defer() 回調(diào)一樣，如果定時(shí)器在創(chuàng)建后被應(yīng)用程序禁用而無法運(yùn)行，則必須手動(dòng)取消計(jì)劃用于一次性執(zhí)行的定時(shí)器以釋放資源。

案例

<?php
/**
 * @author Tinywan(ShaoBo Wan)
 * @email 756684177@qq.com
 * @date 2024/1/31 18:24
 */

require 'vendor/autoload.php';

use Revolt\EventLoop;

EventLoop::delay(3, function (): void {
    print '3 seconds passed';
});

EventLoop::run();

3秒后輸出

3 seconds passed

Periodic 定期回調(diào)

• repeat() 調(diào)度回調(diào)以每 n 秒重復(fù)執(zhí)行一次。
• 與所有其他事件回調(diào)一樣， repeat() 定時(shí)器可以隨時(shí)禁用/重新啟用。
• 與 defer() 和 delay() 回調(diào)不同， repeat() 回調(diào)必須顯式取消以釋放關(guān)聯(lián)的資源。一旦 repeat() 回調(diào)的目的實(shí)現(xiàn)，如果不能通過 cancel() 釋放它們，將導(dǎo)致應(yīng)用程序中的內(nèi)存泄漏。僅僅禁用 repeat() 回調(diào)是不夠的，因?yàn)樗鼈兊臄?shù)據(jù)只有在取消時(shí)才被釋放。

案例

<?php
/**
 * @author Tinywan(ShaoBo Wan)
 * @email 756684177@qq.com
 * @date 2024/1/31 18:49
 */

require 'vendor/autoload.php';

use Revolt\EventLoop;

EventLoop::repeat(0.1, function ($callbackId): void {
    static $i = 0;

    if ($i++ < 3) {
        echo "tick\n";
    } else {
        EventLoop::cancel($callbackId);
    }
});

EventLoop::run();

輸出

tick
tick
tick

定時(shí)器偏差

重復(fù)計(jì)時(shí)器基本上是簡單的延遲計(jì)時(shí)器，在觸發(fā)適當(dāng)?shù)奶幚沓绦蛑皶?huì)自動(dòng)重新調(diào)度。它們受定時(shí)器漂移的影響。多個(gè)計(jì)時(shí)器可能會(huì)堆疊在一起，以防它們作為協(xié)程執(zhí)行。

Fibers 纖程

Revolt被設(shè)計(jì)為可以很好地與纖維一起工作。所有事件回調(diào)都在單獨(dú)的纖程中運(yùn)行，并且可以隨時(shí)掛起它。如果在事件回調(diào)中沒有掛起，則纖程將被重用于將來的事件回調(diào)以保存資源。

掛起允許通過掛起當(dāng)前執(zhí)行上下文來等待事件，直到所討論的事件發(fā)生。它們將掛起當(dāng)前纖程并返回到事件循環(huán)，或者如果從纖程外部（即從 {main} ）調(diào)用，則開始運(yùn)行事件循環(huán)。

應(yīng)使用 Revolt\EventLoop\Suspension API暫停和恢復(fù)光纖。Suspension 對(duì)象可以使用 Revolt\EventLoop::getSuspension() 創(chuàng)建。在獲得 Suspension 對(duì)象之后，可以注冊(cè)事件回調(diào)以調(diào)度當(dāng)前纖程的恢復(fù)。$suspension->suspend() 將掛起當(dāng)前的執(zhí)行上下文，直到它通過 $suspension->resume() 或 $suspension->throw()恢復(fù)。

案例：讓我們暫停主執(zhí)行上下文，直到有數(shù)據(jù)從 STDIN 讀取或超時(shí)到期：

<?php

require __DIR__ . '/vendor/autoload.php';

use Revolt\EventLoop;

if (\stream_set_blocking(STDIN, false) !== true) {
    \fwrite(STDERR, "Unable to set STDIN to non-blocking" . PHP_EOL);
    exit(1);
}

print "Write something and hit enter" . PHP_EOL;

$suspension = EventLoop::getSuspension();

$readableId = EventLoop::onReadable(STDIN, function ($id, $stream) use ($suspension): void {
    EventLoop::cancel($id);

    $chunk = \fread($stream, 8192);

    print "Read " . \strlen($chunk) . " bytes" . PHP_EOL;

    $suspension->resume(null);
});

$timeoutId = EventLoop::delay(5, function () use ($readableId, $suspension) {
    EventLoop::cancel($readableId);
    
    print "Timeout reached" . PHP_EOL;

    $suspension->resume(null);
});

$suspension->suspend();

EventLoop::cancel($readableId);
EventLoop::cancel($timeoutId);

自動(dòng)超時(shí)輸出

Write something and hit enter
Timeout reached

按Enter鍵盤輸出

Write something and hit enter

Read 1 bytes

Signals 信號(hào)

信號(hào)是類Unix操作系統(tǒng)中的標(biāo)準(zhǔn)化消息。

EventLoop::onSignal() 可用于對(duì)發(fā)送到進(jìn)程的信號(hào)作出反應(yīng)。

<?php

require __DIR__ . '/vendor/autoload.php';

use Revolt\EventLoop;

// Let's tick off output once per second, so we can see activity.
EventLoop::repeat(1, function (): void {
    echo "tick: ", date('c'), "\n";
});

// What to do when a SIGINT signal is received
EventLoop::onSignal(SIGINT, function (): void {
    echo "Caught SIGINT! exiting ...\n";
    exit;
});

EventLoop::run();

SIGINT 信號(hào): 當(dāng)用戶按某些終端鍵時(shí), 引發(fā)終端產(chǎn)生的信號(hào). 如Ctrl+C鍵, 這將產(chǎn)生中斷信號(hào)SIGINT. 它將停止一個(gè)已失去控制的程序。

Ctrl+C 輸出

tick: 2024-01-31T11:54:03+00:00
tick: 2024-01-31T11:54:04+00:00
tick: 2024-01-31T11:54:05+00:00
tick: 2024-01-31T11:54:06+00:00
tick: 2024-01-31T11:54:07+00:00

tick: 2024-01-31T11:54:08+00:00
tick: 2024-01-31T11:54:09+00:00

tick: 2024-01-31T11:54:10+00:00
tick: 2024-01-31T11:54:11+00:00

tick: 2024-01-31T11:54:12+00:00
^CCaught SIGINT! exiting ...

從基本原理中可以清楚地看到，信號(hào)回調(diào)可以像任何其他事件回調(diào)一樣被啟用、禁用和取消。一般來說，如果所有回調(diào)都消失了，只有信號(hào)回調(diào)仍然存在，那么您希望退出事件循環(huán)，除非您沒有主動(dòng)等待該事件發(fā)生。

信號(hào)號(hào)可用性

ext-uv 暴露 UV::SIG* 常量用于可觀察信號(hào)。使用 EventDriver 的應(yīng)用程序在注冊(cè)信號(hào)回調(diào)或依賴 ext-pcntl 時(shí)需要手動(dòng)指定適當(dāng)?shù)恼麛?shù)信號(hào)編號(hào)。