與PHP中的Pthreads平行編程 - 基本面

關鍵要點

• 避免在Web 服務器環境中使用pthreads： 由於安全性和可擴展性問題，不應在FCGI 等Web 服務器環境中使用pthreads，因為它在這些環境中無法有效處理多個線程。
• 將pthreads 用於一次性任務或IO 綁定操作： 對於執行一次或需要大量IO 操作的任務，使用pthreads 可以幫助卸載主執行線程，並通過在單獨的線程中處理這些操作來提高性能。
• 回收線程以優化資源： 為每個任務創建新線程可能會佔用大量資源；相反，請通過 Worker 或 Pool 類重用線程，以便更有效地管理和執行多個任務。
• 了解pthreads 的不變性和Volatile 類： 默認情況下，擴展Threaded 的對象的屬性是不可變的，以避免性能下降，Volatile 類提供了一種在必要時管理可變屬性的方法。
• 實現同步以確保線程安全： 為防止數據損壞並確保多個線程訪問共享資源時的一致結果，請使用pthreads 提供的同步方法，例如同步塊和Threaded::wait 和Threaded::notify 等方法。

本文由 Christopher Pitt 審核。感謝所有 SitePoint 的同行評審者，使 SitePoint 內容盡善盡美！

---

PHP 開發人員似乎很少利用並行性。同步、單線程編程的簡單性確實很有吸引力，但有時使用一點並發可以帶來一些值得的性能改進。

在本文中，我們將了解如何使用 pthreads 擴展在 PHP 中實現線程。這需要安裝 ZTS（Zend 線程安全）版本的 PHP 7.x，以及安裝 pthreads v3。 （在撰寫本文時，PHP 7.1 用戶需要從 pthreads repo 的主分支安裝——請參閱本文的部分內容，了解有關從源代碼構建第三方擴展的詳細信息。）

快速說明一下：pthreads v2 面向 PHP 5.x，不再受支持；pthreads v3 面向 PHP 7.x，並且正在積極開發中。

非常感謝 Joe Watkins（pthreads 擴展的創建者）校對並幫助改進我的文章！

何時不使用 pthreads

在我們繼續之前，我想首先說明您不應該（以及不能）使用 pthreads 擴展的情況。

在 pthreads v2 中，建議不要在 Web 服務器環境（即在 FCGI 進程中）中使用 pthreads。從 pthreads v3 開始，此建議已強制執行，因此您現在根本不能在 Web 服務器環境中使用它。這樣做的兩個主要原因是：

1. 在這種環境中使用多個線程是不安全的（會導致 IO 問題以及其他問題）。
2. 它不能很好地擴展。例如，假設您有一個 PHP 腳本，該腳本創建一個新線程來處理一些工作，並且該腳本在每次請求時都會執行。這意味著對於每個請求，您的應用程序都會創建一個新線程（這是一個 1:1 線程模型——一個線程對應一個請求）。如果您的應用程序每秒處理 1,000 個請求，那麼它每秒就會創建 1,000 個線程！在單個機器上運行如此多的線程很快就會使它不堪重負，並且隨著請求速率的增加，這個問題只會加劇。

這就是為什麼線程在這種環境中不是一個好解決方案的原因。如果您正在尋找線程作為 IO 阻塞任務（例如執行 HTTP 請求）的解決方案，那麼讓我向您指出異步編程的方向，這可以通過 Amp 等框架實現。 SitePoint 發布了一些關於此主題的優秀文章（例如編寫異步庫和使用 PHP 修改 Minecraft），如果您感興趣的話。

言歸正傳，讓我們直接進入正題！

處理一次性任務

有時，您希望以多線程方式處理一次性任務（例如執行一些 IO 綁定任務）。在這種情況下，可以使用 Thread 類創建一個新線程，並在該單獨線程中運行一些工作單元。

例如：

$task = new class extends Thread {
    private $response;

    public function run()
    {
        $content = file_get_contents("http://google.com");
        preg_match("~<title>(.+)</title>~", $content, $matches);
        $this->response = $matches[1];
    }
};

$task->start() && $task->join();

var_dump($task->response); // string(6) "Google"

在上面，run 方法是我們將在新線程中執行的工作單元。調用 Thread::start 時，將生成新線程並調用 run 方法。然後，我們將生成的線程重新加入到主線程（通過 Thread::join），這將阻塞，直到單獨的線程完成執行。這確保了在嘗試輸出結果（存儲在 $task->response 中）之前，任務已完成執行。

將線程相關的邏輯（包括必須定義 run 方法）污染類的職責可能並不理想。我們可以通過讓它們擴展 Threaded 類來隔離這些類，然後可以在其他線程中運行它們：

class Task extends Threaded
{
    public $response;

    public function someWork()
    {
        $content = file_get_contents('http://google.com');
        preg_match('~<title>(.+)</title>~', $content, $matches);
        $this->response = $matches[1];
    }
}

$task = new Task;

$thread = new class($task) extends Thread {
    private $task;

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

    public function run()
    {
        $this->task->someWork();
    }
};

$thread->start() && $thread->join();

var_dump($task->response);

任何需要在單獨線程中運行的類都必須以某種方式擴展 Threaded 類。這是因為它提供了在不同線程中運行的必要能力，以及提供隱式安全性和有用的接口（用於資源同步等）。

讓我們快速了解一下 pthreads 公開的類層次結構：

<code>Threaded (implements Traversable, Collectable)
    Thread
        Worker
    Volatile
Pool</code>

我們已經了解了 Thread 和 Threaded 類的基礎知識，所以現在讓我們來看看其餘三個（Worker、Volatile 和 Pool）。

回收線程

為每個要並行化的任務啟動一個新線程是昂貴的。這是因為為了在 PHP 內部實現線程，pthreads 必須採用共享無狀態架構。這意味著必須為創建的每個線程複製 PHP 解釋器當前實例的整個執行上下文（包括每個類、接口、特性和函數）。由於這會造成明顯的性能影響，因此應始終盡可能重用線程。可以通過兩種方式重用線程：使用 Worker 或使用 Pool。

Worker 類用於在另一個線程中同步執行一系列任務。這是通過創建一個新的 Worker 實例（這將創建一個新線程），然後將任務堆疊到該單獨線程（通過 Worker::stack）來完成的。

這是一個簡單的示例：

$task = new class extends Thread {
    private $response;

    public function run()
    {
        $content = file_get_contents("http://google.com");
        preg_match("~<title>(.+)</title>~", $content, $matches);
        $this->response = $matches[1];
    }
};

$task->start() && $task->join();

var_dump($task->response); // string(6) "Google"

輸出：

上面通過 Worker::stack 將 15 個任務堆疊到新的 $worker 對像上，然後按堆疊順序處理它們。如上所示，Worker::collect 方法用於在任務完成執行後清理任務。通過在 while 循環中使用它，我們阻塞主線程，直到所有堆疊的任務都完成執行並已清理完畢，然後我們觸發 Worker::shutdown。過早關閉工作程序（即在仍有待執行的任務時）仍將阻塞主線程，直到所有任務都完成執行——任務只是不會被垃圾回收（導致內存洩漏）。

Worker 類提供了一些其他與任務堆棧相關的 method，包括 Worker::unstack 用於刪除最舊的堆疊項，以及 Worker::getStacked 用於執行堆棧上的項目數量。工作程序的堆棧只保存要執行的任務。一旦堆棧中的任務執行完畢，它就會被刪除，然後放在另一個（內部）堆棧上以進行垃圾回收（使用 Worker::collect）。

在執行許多任務時重用線程的另一種方法是使用線程池（通過 Pool 類）。線程池由一組 Worker 驅動，以使任務能夠並發執行，其中並發因子（池運行的線程數）在池創建時指定。

讓我們調整上面的示例以使用工作程序池：

class Task extends Threaded
{
    public $response;

    public function someWork()
    {
        $content = file_get_contents('http://google.com');
        preg_match('~<title>(.+)</title>~', $content, $matches);
        $this->response = $matches[1];
    }
}

$task = new Task;

$thread = new class($task) extends Thread {
    private $task;

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

    public function run()
    {
        $this->task->someWork();
    }
};

$thread->start() && $thread->join();

var_dump($task->response);

輸出：

使用池與使用工作程序之間存在一些顯著差異。首先，池不需要手動啟動，它們會在任務可用時立即開始執行任務。其次，我們將任務提交到池中，而不是堆疊它們。此外，Pool 類不擴展 Threaded，因此它可能不會傳遞到其他線程（與 Worker 不同）。

作為良好實踐，應始終在完成任務後收集工作程序和池的任務，並手動關閉它們。通過 Thread 類創建的線程也應重新加入創建者線程。

pthreads 和（非）可變性

最後一個要介紹的類是 Volatile——pthreads v3 的一個新補充。不變性已成為 pthreads 中的一個重要概念，因為如果沒有它，性能會嚴重下降。因此，默認情況下，本身是 Threaded 對象的 Threaded 類的屬性現在是不可變的，因此在初始賦值後不能重新賦值。現在更傾向於對這些屬性進行顯式可變性，並且仍然可以通過使用新的 Volatile 類來完成。

讓我們快速查看一個示例來演示新的不變性約束：

$task = new class extends Thread {
    private $response;

    public function run()
    {
        $content = file_get_contents("http://google.com");
        preg_match("~<title>(.+)</title>~", $content, $matches);
        $this->response = $matches[1];
    }
};

$task->start() && $task->join();

var_dump($task->response); // string(6) "Google"

另一方面，Volatile 類的 Threaded 屬性是可變的：

class Task extends Threaded
{
    public $response;

    public function someWork()
    {
        $content = file_get_contents('http://google.com');
        preg_match('~<title>(.+)</title>~', $content, $matches);
        $this->response = $matches[1];
    }
}

$task = new Task;

$thread = new class($task) extends Thread {
    private $task;

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

    public function run()
    {
        $this->task->someWork();
    }
};

$thread->start() && $thread->join();

var_dump($task->response);

我們可以看到，Volatile 類覆蓋了其父類 Threaded 類強制執行的不變性，以允許重新分配（以及取消設置）Threaded 屬性。

關於可變性和 Volatile 類，還有一個最後一個基本主題需要介紹——數組。當將數組分配給 Threaded 類的屬性時，pthreads 中的數組會自動強制轉換為 Volatile 對象。這是因為在 PHP 中從多個上下文中操作數組根本不安全。

讓我們再次快速查看一個示例以更好地理解：

<code>Threaded (implements Traversable, Collectable)
    Thread
        Worker
    Volatile
Pool</code>

我們可以看到，Volatile 對象可以像對待數組一樣對待，因為它們為基於數組的操作（如上所示）提供了對子集運算符（[]）的支持。但是，Volatile 類不受常見的基於數組的函數（例如 array_pop 和 array_shift）的支持。相反，Threaded 類為我們提供了這些操作作為內置方法。

作為演示：

class Task extends Threaded
{
    private $value;

    public function __construct(int $i)
    {
        $this->value = $i;
    }

    public function run()
    {
        usleep(250000);
        echo "Task: {$this->value}\n";
    }
}

$worker = new Worker();
$worker->start();

for ($i = 0; $i < 15; $i++) {
    $worker->stack(new Task($i));
}

while ($worker->collect());

$worker->shutdown();

其他受支持的操作包括 Threaded::chunk 和 Threaded::merge。

同步

本文將介紹的最後一個主題是 pthreads 中的同步。同步是一種允許控制訪問共享資源的技術。

例如，讓我們實現一個簡單的計數器：

class Task extends Threaded
{
    private $value;

    public function __construct(int $i)
    {
        $this->value = $i;
    }

    public function run()
    {
        usleep(250000);
        echo "Task: {$this->value}\n";
    }
}

$pool = new Pool(4);

for ($i = 0; $i < 15; $i++) {
    $pool->submit(new Task($i));
}

while ($pool->collect());

$pool->shutdown();

如果不使用同步，則輸出不是確定性的。多個線程寫入單個變量而不進行受控訪問會導致更新丟失。

讓我們通過添加同步來糾正這個問題，以便我們獲得正確的輸出 20：

class Task extends Threaded // a Threaded class
{
    public function __construct()
    {
        $this->data = new Threaded();
        // $this->data is not overwritable, since it is a Threaded property of a Threaded class
    }
}

$task = new class(new Task()) extends Thread { // a Threaded class, since Thread extends Threaded
    public function __construct($tm)
    {
        $this->threadedMember = $tm;
        var_dump($this->threadedMember->data); // object(Threaded)#3 (0) {}
        $this->threadedMember = new StdClass(); // invalid, since the property is a Threaded member of a Threaded class
    }
};

同步代碼塊還可以使用 Threaded::wait 和 Threaded::notify（以及 Threaded::notifyOne）相互協作。

以下是來自兩個同步 while 循環的交錯增量：

class Task extends Volatile
{
    public function __construct()
    {
        $this->data = new Threaded();
        $this->data = new StdClass(); // valid, since we are in a volatile class
    }
}

$task = new class(new Task()) extends Thread {
    public function __construct($vm)
    {
        $this->volatileMember = $vm;

        var_dump($this->volatileMember->data); // object(stdClass)#4 (0) {}

        // still invalid, since Volatile extends Threaded, so the property is still a Threaded member of a Threaded class
        $this->volatileMember = new StdClass();
    }
};

您可能已經註意到，在對 Threaded::wait 的調用周圍添加了其他條件。這些條件至關重要，因為它們只允許同步回調在收到通知並且指定條件為真時恢復。這很重要，因為通知可能來自 Threaded::notify 的調用以外的地方。因此，如果對 Threaded::wait 的調用未包含在條件中，我們將容易受到虛假喚醒調用的影響，這將導致代碼不可預測。

結論

我們已經看到了 pthreads 附帶的五個類（Threaded、Thread、Worker、Volatile 和 Pool），包括介紹每個類的用法。我們還研究了 pthreads 中新的不變性概念，以及對其支持的同步功能的快速瀏覽。涵蓋了這些基礎知識後，我們現在可以開始研究將 pthreads 應用於一些實際用例！這將是我們下一篇文章的主題。

與此同時，如果您對 pthreads 有任何應用程序創意，請隨時在下面的評論區中留下您的想法！

關於在 PHP 中使用 Pthreads 進行並行編程的常見問題解答 (FAQ)

使用 Pthreads 在 PHP 中的先決條件是什麼？

要在 PHP 中使用 Pthreads，您需要具備 PHP 和麵向對象編程的工作知識。您還需要安裝啟用 ZTS（Zend 線程安全）的 PHP。 Pthreads 在標準 PHP 安裝中不可用；它需要使用線程安全構建的 PHP 版本。您可以通過在終端中運行命令“php -i | grep “Thread Safety””來檢查您的 PHP 安裝是否啟用了 ZTS。如果它返回“Thread Safety => enabled”，那麼您可以使用 Pthreads。

如何在 PHP 中安裝 Pthreads？

要安裝 Pthreads，您需要使用 PECL，即 PHP 擴展社區庫。首先，確保您已安裝啟用 ZTS 的 PHP。然後，在您的終端中，運行命令“pecl install pthreads”。如果安裝成功，您需要將行“extension=pthreads.so”添加到您的 php.ini 文件中。這將在每次運行 PHP 時加載 Pthreads 擴展。

如何使用 Pthreads 在 PHP 中創建一個新線程？

要創建一個新線程，您需要定義一個擴展 Pthreads 提供的 Thread 類的類。在此類中，您將覆蓋 run() 方法，這是在新線程中將執行的代碼。然後，您可以創建此類的實例並調用其 start() 方法來啟動新線程。

如何使用 Pthreads 在 PHP 中在線程之間共享數據？

Pthreads 提供 Threaded 類用於在線程之間共享數據。您可以創建一個此類的新的實例並將其傳遞給您的線程。在此對像上設置的任何屬性都將在線程之間安全共享。

如何處理 Pthreads 中的錯誤？

Pthreads 中的錯誤處理類似於標準 PHP 中的錯誤處理。您可以使用 try-catch 塊來捕獲異常。但是，請注意，每個線程都有其自己的範圍，因此一個線程中的異常不會影響其他線程。

我可以在 Laravel 或 Symfony 等 PHP 框架中使用 Pthreads 嗎？

Pthreads 與 Laravel 或 Symfony 等 PHP 框架不兼容。這是因為這些框架並非設計為線程安全的。如果您需要在這些框架中執行並行處理，請考慮使用其他技術，例如隊列或異步任務。

如何調試使用 Pthreads 的 PHP 腳本？

調試使用 Pthreads 的 PHP 腳本可能具有挑戰性，因為每個線程都在其自己的上下文中運行。但是，您可以使用標準調試技術，例如記錄或將數據輸出到控制台。您還可以使用像 Xdebug 這樣的 PHP 調試器，但請注意，並非所有調試器都支持多線程應用程序。

我可以在 Web 服務器環境中使用 Pthreads 嗎？

不建議在 Web 服務器環境中使用 Pthreads。它專為 CLI（命令行界面）腳本設計。在 Web 服務器環境中使用 Pthreads 會導致不可預測的結果，並且通常是不安全的。

如何使用 Pthreads 在 PHP 中停止正在運行的線程？

要停止正在運行的線程，您可以使用 Pthreads 提供的 kill() 方法。但是，應謹慎使用此方法，因為如果線程處於操作過程中，它可能會導致不可預測的結果。通常最好設計您的線程，以便它們可以乾淨地完成其任務。

是否有用於 PHP 中並行編程的 Pthreads 的替代方案？

是的，有幾種用於 PHP 中並行編程的 Pthreads 的替代方案。這些包括 forks，這是一個提供用於創建和管理子進程的接口的 PECL 擴展；以及 parallel，這是 PHP 7.2 中引入的原生 PHP 擴展，它提供了一個更簡單、更安全的並行編程接口。

以上是與PHP中的Pthreads平行編程 - 基本面的詳細內容。更多資訊請關注PHP中文網其他相關文章！