parallel\Channel 类

    (0.9.0)

    无缓冲通道

    无缓冲通道将在调用 parallel\Channel::send() 时阻塞，直到有接收方，并在调用 parallel\Channel::recv() 时阻塞，直到有发送方。这意味着无缓冲通道不仅是任务之间共享数据的途径，也是一种简单的同步方法。

    无缓冲通道是任务之间共享数据的最快方式，需要最少的复制。

    带缓冲通道

    带缓冲通道在调用 parallel\Channel::send() 时不会阻塞，直到容量达到，调用 parallel\Channel::recv() 将阻塞，直到缓冲区中有数据。

    通道上的闭包

    并行通道的一个强大功能是它们允许在任务（和运行时）之间交换闭包。

    当闭包通过通道发送时，闭包会被缓冲，它不会改变传输闭包的通道的缓冲，但它确实会影响闭包内部的静态作用域：发送到不同运行时或同一运行时的同一个闭包将不会共享它们的静态作用域。

    这意味着，无论何时执行通过通道传输的闭包，静态状态都将与闭包被缓冲时的状态相同。

    匿名通道

    匿名通道构造函数允许程序员避免为每个通道分配名称：parallel 将为匿名通道生成一个唯一的名称。

    类摘要

    final class parallel\Channel {
    /* 匿名构造函数 */
    public __construct()
    public __construct(int $capacity)
    /* 访问 */
    public make(string $name): Channel
    public make(string $name, int $capacity): Channel
    public open(string $name): Channel
    /* 共享 */
    public recv(): mixed
    public send(mixed $value): void
    /* 关闭 */
    public close(): void
    /* 无限缓冲的常量 */
    const Infinite;
    }

    目录

    改进此页面

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

    用户贡献注释 4 个注释

    4
    hdvianna
    4 年前
    这是一个使用通道为消费者生成数据的示例。在此示例中，生产者 Runtime 实例将发送消费者应休眠的秒数。

    <?php

    use parallel\{Runtime, Channel};

    main($argv);

    function     main(array $argv)
    {
    if (    count($argv) !== 3) {
    echo     "Type: hello-parallel.php <number-of-tasks> <maximum-time-of-sleep (in seconds)>" . PHP_EOL;
    echo     "Example: hello-parallel.php 5 3" . PHP_EOL;
    die;
    } else {
    $numberOfTasks = intval($argv[1]);
    $maximumTimeOfSleep = intval($argv[2]);
    $t1 = microtime(true);
    parallelize($numberOfTasks, $maximumTimeOfSleep);
    $endTime = microtime(true) - $t1;
    echo     PHP_EOL."Finished $numberOfTasks task(s) in {$endTime}s".PHP_EOL;
    }
    }

    function     parallelize(int $numberOfTasks, int $maximumTimeOfSleep)
    {
    $channel = new Channel();

    $taskIds = array_map(function () use ($maximumTimeOfSleep) {
    return     $id = uniqid("task::");
    },     range(0, $numberOfTasks - 1));

    $timesToSleep = array_map(function () use ($maximumTimeOfSleep) {
    return     rand(1, $maximumTimeOfSleep);
    },     $taskIds);

    $producer = new Runtime();
    $producerFuture = $producer->run(function (Channel $channel, array $timesToSleep) {
    foreach (    $timesToSleep as $timeToSleep) {
    $channel->send($timeToSleep);
    }
    }, [    $channel, $timesToSleep]);

    $consumerFutures = array_map(function (string $id) use ($channel) {
    $runtime = new Runtime();
    return     $runtime->run(function (string $id, Channel $channel) {
    $timeToSleep = $channel->recv();
    echo     "Hello from $id. I will sleep for $timeToSleep second(s).".PHP_EOL;
    sleep($timeToSleep);
    echo     "$id slept for $timeToSleep second(s).".PHP_EOL;
    return     $timeToSleep;
    }, [    $id, $channel]);
    },     $taskIds);

    wait($consumerFutures);
    wait([$producerFuture]);
    }

    function     wait(array $futures)
    {
    return     array_map(function ($future) {
    return     $future->value();
    },     $futures);
    }
    2
    rustysun
    4 年前
    这是一个使用无缓冲通道的示例。
    <?php

    use parallel\{Channel,Runtime};

    $sum=function(array $a, Channel $ch) {
    $sum=0;
    foreach (    $a as $v) {
    $sum+=$v;
    }
    $ch->send($sum);
    };
    try {
    $a=[7, 2, 8, 1, 4, 0, 9, 10];
    //无缓冲通道
    $runtime=new Runtime;
    $ch2=new Channel;
    $runtime->run($sum, [array_slice($a, 0, $num), $ch2]);
    $runtime->run($sum, [array_slice($a, $num), $ch2]);
    //从通道接收
    $x=$ch2->recv();
    $y=$ch2->recv();
    $ch2->close();
    echo     "\nch2:", $x, "\t", $y, "\t", $x + $y, "\n";
    } catch(    Error $err) {
    echo     "\nError:", $err->getMessage();
    } catch(    Exception $e) {
    echo     "\nException:", $e->getMessage();
    }

    //输出
    //ch2:18 23 41
    2
    gam6itko
    2 年前
    <?php

    // 计算阶乘的奇怪方式 ^_^
    // 我们创建一个线程并使用带缓冲的通道进行同步
    // 每次只执行一个线程中的 fact

    use parallel\{Channel, Future, Runtime};

    for (    $n = 0; $n <= 10; $n++) {
    echo     "!$n = " . factorial($n) . PHP_EOL;
    }

    /**
    * 创建 $n 个线程。
    */
    function factorial(int $n): int
    {
    // 带缓冲的通道 - 用于同步线程 ^_^
    $channel = new Channel(1);
    $futureList = [];
    for (    $i = 2; $i <= $n; $i++) {
    $runtime = new Runtime();
    $futureList[] = $runtime->run(
    static function (    Channel $channel, $multiplier): void {
    $f = $channel->recv();
    $channel->send($f * $multiplier);
    },
    [    $channel, $i]
    );
    }

    $channel->send(1);

    // 等待所有线程完成
    do {
    $allDone = array_reduce(
    $futureList,
    function (    bool $c, Future $future): bool {

    return     $c && $future->done();
    },
    true
    );
    } while (    false === $allDone);

    return     $channel->recv();
    }

    // 输出:
    // !0 = 1
    // !1 = 1
    // !2 = 2
    // !3 = 6
    // !4 = 24
    // !5 = 120
    // !6 = 720
    // !7 = 5040
    // !8 = 40320
    // !9 = 362880
    // !10 = 3628800
    1
    rustysun
    4 年前
    <?php
    use parallel\Channel;

    function     sum(array $a, Channel $ch) {
    $sum=0;
    foreach (    $a as $v) {
    $sum+=$v;
    }
    $ch->send($sum);
    }

    try {
    $a=[7, 2, 8, 1, 4, 0, 9, 10];
    $ch1=Channel::make('sum', 2);
    $ch2=new Channel;
    $num=count($a) / 2;
    sum(array_slice($a, 0, $num), $ch1);
    sum(array_slice($a, $num), $ch1);

    //从通道接收
    $x=$ch1->recv();
    $y=$ch1->recv();
    $ch1->close();
    echo     "\nch1:", $x, "\t", $y, "\t", $x + $y, "\n";
    } catch(    Error $err) {
    echo     "\nError:", $err->getMessage();
    } catch(    Exception $e) {
    echo     "\nException:", $e->getMessage();
    }
    ＋添加注释
    To Top