与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中文网其他相关文章！