parallel\Future 类

    (0.8.0)

    期货

    期货表示任务的返回值或未捕获的异常，并公开用于取消的 API。

    示例 #1 显示期货作为返回值的示例

    <?php
    $runtime     = new \parallel\Runtime;
    $future = $runtime->run(function(){
    return     "World";
    });
    printf("Hello %s\n", $future->value());
    ?>

    上面的示例将输出类似于以下内容

    Hello World

    期货的行为也允许它用作简单的同步点，即使任务没有显式返回值。

    示例 #2 显示期货作为同步点的示例

    <?php
    $runtime     = new \parallel\Runtime;
    $future = $runtime->run(function(){
    echo     "in child ";
    for (    $i = 0; $i < 500; $i++) {
    if (    $i % 10 == 0) {
    echo     ".";
    }
    }
    echo     " leaving child";
    });

    $future->value();
    echo     "\nparent continues\n";
    ?>

    上面的示例将输出类似于以下内容

    in child .................................................. leaving child
parent continues

    类概要

    final class parallel\Future {
    /* 解析 */
    public value(): 混合类型
    /* 状态 */
    public cancelled(): 布尔型
    public done(): 布尔型
    /* 取消 */
    public cancel(): 布尔型
    }

    目录

    改进此页面

    了解如何改进此页面 • 提交拉取请求 • 报告错误
    ＋添加备注

    用户贡献的备注 1 个备注

    -1
    Thierry Kauffmann
    2 年前
    <?php

    /**
    * 示例并行 Runtime/Future 处理
    * 使用生成器生成要处理的项目列表
    *
    * 并行处理的项目来自生成器
    * 它可以是任何东西：例如获取 mysql 数组、DirectoryIterator...
    * 因此，并行处理的项目数量事先未知
    *
    * 此算法动态地将项目分配给每个并行线程
    * 线程完成工作后
    * 它将被分配一个新的项目进行处理
    * 直到所有项目都被处理完毕（生成器关闭）
    *
    * 在此示例中，我们在 5 个并行线程中处理 50 个项目
    * 它以这种形式输出（输出在每次运行时都会改变）：
    *
    * ThreadId: 1 => Item: 1 (Start)
    * ThreadId: 2 => Item: 2 (Start)
    * ThreadId: 3 => Item: 3 (Start)
    * ThreadId: 4 => Item: 4 (Start)
    * ThreadId: 5 => Item: 5 (Start)
    * ThreadId: 5 => Item: 5 Sleep: 3s (End)
    * ThreadId: 5 => Item: 6 (Start)
    * ThreadId: 3 => Item: 3 Sleep: 4s (End)
    * ThreadId: 3 => Item: 7 (Start)
    * ThreadId: 2 => Item: 2 Sleep: 6s (End)
    * ThreadId: 2 => Item: 8 (Start)
    * ...
    * ThreadId: 4 => Item: 44 Sleep: 6s (End)
    * ThreadId: 4 => Item: 49 (Start)
    * ThreadId: 3 => Item: 46 Sleep: 5s (End)
    * ThreadId: 3 => Item: 50 (Start)
    * ThreadId: 2 => Item: 43 Sleep: 9s (End)
    * Destroy ThreadId: 2
    * ThreadId: 1 => Item: 47 Sleep: 5s (End)
    * Destroy ThreadId: 1
    * ThreadId: 4 => Item: 49 Sleep: 7s (End)
    * Destroy ThreadId: 4
    * ThreadId: 5 => Item: 48 Sleep: 10s (End)
    * Destroy ThreadId: 5
    * ThreadId: 3 => Item: 50 Sleep: 10s (End)
    * Destroy ThreadId: 3
    */

    // 使用生成器生成要处理的项目列表
    function generator(int $item_count) {
    echo     "项目数量： $item_count\n";
    for (    $i=1; $i <= $item_count; $i++) {
    yield     $i;
    }
    }

    function     testConcurrency(int $concurrency, int $max_item_count) {

    // 将 item_count 设置为随机数，以便每次运行都不同
    $item_count= rand(1, $max_item_count);
    // 创建生成器
    $generator = generator($item_count);

    // 在每个线程中执行的函数。例如，随机时间内有一个快照！
    // 你可以在此处进行一些计算或传输文件或其他操作...
    $producer = function (int $item_id) {
    $seconds = rand(1, 10);
    sleep($seconds);
    return [    'item_id' => $item_id, 'sleep_seconds' => $seconds];
    };

    // 用一个初始化为“false”值的 Future 填充线程
    for ($thread_id = 1; $thread_id <= $concurrency; $thread_id++) {
    $threads[$thread_id] = ['runtime' => new \parallel\Runtime(), 'future' => false];
    }

    // 为整个处理创建无限循环
    while (true) {
    // 遍历线程，直到生成器关闭且所有线程都被销毁
    foreach ($threads as $thread_id => &$thread) {
    if (    $thread['future'] === false and $generator->valid()) {
    // 线程处于非活动状态且生成器仍有值：在线程中运行一些内容
    $item_id = $generator->current();
    $thread['future'] = $thread['runtime']->run($producer, [$item_id]);
    echo     "ThreadId: $thread_id => Item: $item_id (Start)\n";
    $generator->next();
    } elseif (    $thread['future'] === false) {
    // 如果生成器已关闭，则销毁线程
    echo "Destroy ThreadId: $thread_id\n";
    unset(    $threads[$thread_id]);
    } elseif (    $thread['future']->done()) {
    // 线程完成工作。获取结果值。
    $item = $thread['future']->value();
    echo     "ThreadId: $thread_id => Item: {$item['item_id']} Sleep: {$item['sleep_seconds']}s (End)\n";
    // 线程准备再次运行
    $thread['future'] = false;
    }
    }

    // 当所有线程都被销毁时，退出无限循环
    if (empty($threads)) break;
    }
    }

    $concurrency = 5;
    $item_count = 50;

    testConcurrency($concurrency, $item_count);

    ?>
    ＋添加备注
    To Top