PHP code example of kode / parallel

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

kode / parallel example snippets

function Kode\Parallel\map;
use function Kode\Parallel\engine;

// 自动按 CPU 核心数并行处理，结果保持输入键序
$sizes = map(['a.jpg', 'b.jpg', 'c.jpg'], static fn(string $file): int => strlen($file));

echo '当前引擎: ' . engine() . PHP_EOL;   // parallel / process / sync
print_r($sizes);

Kode\Parallel\Runtime\Runtime;

$runtime = new Runtime();

// 并行执行 5 个任务
$futures = [];
for ($i = 0; $i < 5; $i++) {
    $futures[] = $runtime->run(
        fn($args) => $args['n'] * $args['n'],
        ['n' => $i + 1]
    );
}

foreach ($futures as $f) {
    echo "结果: " . $f->get() . "\n";
}

$runtime->close();

use Kode\Parallel\Runtime\Runtime;
use Kode\Parallel\Concurrency\Channel;

$runtime = new Runtime();
$ch = Channel::make();

$runtime->run(fn() => $ch->send(range(1, 100)));
$runtime->run(fn() => $ch->send(range(101, 200)));

$sum = fn() => array_sum($ch->recv()) + array_sum($ch->recv());
echo "总和: " . $runtime->run($sum)->get() . "\n";

$runtime->close();

use Kode\Parallel\Fiber\Fiber;

$fiber = new Fiber(function() {
    $value = Fiber::suspend('暂停');
    return "收到: $value";
});

echo $fiber->start() . "\n";       // 暂停
echo $fiber->resume('恢复数据') . "\n"; // 收到: 恢复数据

use Kode\Parallel\Engine\EngineFactory;
use Kode\Parallel\Runtime\Runtime;

EngineFactory::available();            // ['parallel' => true, 'sync' => true]  (ZTS + ext-parallel)
EngineFactory::detect();               // 'parallel'

$runtime = new Runtime();              // 自动选择（真线程）
$runtime = new Runtime(null, 'sync');  // 显式指定同步回退

// 需要多进程时，把 kode/process 接入统一调度（详见 docs/ENGINE.md）
EngineFactory::register(
    name: 'process',
    factory: static fn(?string $bootstrap, int $workers) => new MyProcessEngine($bootstrap, $workers),
    supported: static fn(): bool => extension_loaded('pcntl'),
);

use Kode\Parallel\Future\Futures;

$futures = $runtime->runAll([
    fn(array $a) => file_get_contents('https://example.com/a'),
    fn(array $a) => file_get_contents('https://example.com/b'),
]);

$results  = Futures::all($futures, timeoutMs: 5000);   // 任一失败即抛出
$settled  = Futures::settle($futures);                 // 全部结束，返回状态数组
$fastest  = Futures::any($futures, 3000);              // 第一个成功的结果

use Kode\Parallel\Pool\WorkerPool;

$pool = new WorkerPool(concurrency: 8);

$results = $pool->map(range(1, 100), static fn(int $n): int => $n * $n);
$report  = $pool->mapSettled($urls, $fetch);   // 不因单个失败中断
$stream  = $pool->map($rowGenerator, $render); // 输入可为数组或任意可迭代（生成器只遍历一次，先归整再派发）
$stats   = $pool->stats();                     // engine / submitted / completed / failed / pending

$pool->close();

use Kode\Parallel\Pool\ThreadPool;

$pool = new ThreadPool(8);

// submit 立即返回，不等待线程空闲；可一口气预排海量任务
$f = $pool->submit(static fn(array $a): int => heavy($a['x']), ['x' => $i]);

// 队列积压由工作线程异步消费；需要时再阻塞取结果
$value = $f->get();

// 也可等全部完成
$pool->wait();
$pool->close();

use Kode\Parallel\Concurrency\Lock;
use Kode\Parallel\Concurrency\Atomic;
use Kode\Parallel\Concurrency\Barrier;
use Kode\Parallel\Concurrency\Channel;

// 互斥锁（可命名，跨进程共享）
$lock = Lock::named('order');
$lock->withLock(function () {
    // 临界区
});

// 原子计数器（跨进程安全的自增，CAS 支持）
$counter = new Atomic(0);
$counter->inc();
$counter->compareAndSwap(1, 10);

// 屏障：N 个参与者到齐后整体放行，并自动进入下一代
$barrier = Barrier::named(4, 'phase-1');
$barrier->wait();

// 单运行时消息通道（有界/无界）
$ch = Channel::bounded(8);
$ch->send($item);
$item = $ch->recv();

use Kode\Parallel\Future\Futures;

$future = runtime()->run(fn($a) => $a['x'] + 1, ['x' => 41]);

$chained = $future
    ->then(fn($v) => $v * 2)        // 成功时变换
    ->map(fn($v) => "result:$v")    // 同 then，仅做映射
    ->catch(fn($e) => "fallback");  // 失败时兜底

echo $chained->get();               // "result:84"

// 非阻塞选择：返回第一个已就绪的 Future（对标 parallel\Events::poll / Swoole Channel::select）
$ready = Futures::select([$f1, $f2, $f3], timeoutMs: 1000);

$runtime = new Runtime();
$runtime = new Runtime('/path/to/bootstrap.php');

$future = $runtime->run($task, $args);
$future = $runtime->run(fn($a) => $a['x'] + $a['y'], ['x' => 1, 'y' => 2]);

$runtime->isRunning();
$runtime->pendingCount();   // 尚未回收的任务数（run/isRunning 会自动回收已完成的 Future）
$runtime->getBootstrap();
$runtime->close();

use Kode\Parallel\Concurrency\Channel;

// 无界限通道
$ch = Channel::make();

// 有界限通道（容量为10）
$ch = Channel::bounded(10);

// 发送/接收
$ch->send($data);
$data = $ch->recv();
$ch->close();
$ch->isEmpty();

$future->get();        // 阻塞获取
$future->getOrNull();   // 非阻塞获取
$future->wait(1000);    // 等待1秒
$future->cancel();      // 取消任务

use function Kode\Parallel\Util\pipe;

// 类似 Unix 的管道操作
$result = pipe(
    '  Hello World  ',
    'trim',
    'strtoupper',
    fn($s) => str_replace('WORLD', 'PHP', $s)
);

echo $result; // 输出: HELLO PHP

use function Kode\Parallel\Util\clone_with;

class Color {
    public function __construct(
        public int $red,
        public int $green,
        public int $blue,
        public int $alpha = 255
    ) {}
}

$blue = new Color(79, 91, 147);
$transparent = clone_with($blue, ['alpha' => 128]);

// CurlMulti 自动支持连接复用
$curl = new CurlMulti();
$curl->get('https://api.example.com/1');
$curl->get('https://api.example.com/2');
$curl->get('https://api.example.com/3');
$results = $curl->execute(); // 自动复用连接

use Kode\Parallel\Fiber\Fiber;
use Kode\Parallel\Fiber\FiberManager;

// 单个 Fiber
$fiber = new Fiber(function($input) {
    $step1 = Fiber::suspend('第一步完成');
    return "最终结果: {$step1}";
});

$fiber->start();
$result = $fiber->resume('恢复数据');

// Fiber 管理器
$manager = new FiberManager();
$manager->spawn('task1', fn() => compute1());
$manager->spawn('task2', fn() => compute2());
$manager->startAll();
$results = $manager->collect();

use Kode\Parallel\Concurrency\Lock;
use Kode\Parallel\Concurrency\Semaphore;
use Kode\Parallel\Concurrency\Atomic;

// Lock - 互斥锁（可命名，跨进程共享）
$lock = Lock::named('order');
$lock->withLock(function() {
    // 临界区代码
});

// Semaphore - 信号量（限流，最多3个并发）
$sem = new Semaphore(3);
$sem->withPermits(1, function() {
    // 限流执行
});

// Atomic - 原子计数器（跨进程安全自增 + CAS）
$counter = new Atomic(0);
$counter->inc();
$counter->compareAndSwap(1, 10);

use Kode\Parallel\Pipe\Pipe;

$pipe = Pipe::make('my_pipe');
$pipe->write('Hello');
$data = $pipe->read();

use Kode\Parallel\Curl\CurlMulti;

$curl = new CurlMulti();
$curl->get('https://api.example.com/users');
$curl->post('https://api.example.com/posts', ['title' => 'Hello']);
$results = $curl->execute();

use Kode\Parallel\Runtime\Runtime;
use Kode\Parallel\Concurrency\Channel;

$runtime = new Runtime();
$channel = Channel::bounded(10);

$producer = fn() => produce($channel);
$consumer = fn() => consume($channel);

$runtime->run($producer);
$runtime->run($consumer);

use Kode\Parallel\Pool\WorkerPool;

$pool = new WorkerPool(11);                       // 线程数 ≈ CPU 核数

// 直接 map() 即可，元素较多时内部自动批量合并；批大小 auto（元素数 / 线程数 / 4，封顶 1024）
$results = $pool->map($messages, static fn (mixed $m): string => render($m));

// 群发场景用容错版：单条失败不连累同批其他元素
$settled = $pool->mapBatchSettled($recipients, static fn (array $u): string => send($u));

foreach ($settled as $key => $r) {
    if ($r['status'] === 'rejected') {
        // reason 恒为 ParallelException；原始异常类名见 getContext()['class']
        $retry[$key] = $r['reason']->getMessage();
    }
}

$pool->close();

use Kode\Parallel\Curl\CurlMulti;

$urls = [];
foreach ($users as $u) {
    $urls['u' . $u['id']] = "https://api.example.com/notify?uid={$u['id']}";
}

// 滑动窗口：始终保持 16 个在途连接，完成一个补一个
$results = CurlMulti::fetch($urls, concurrency: 16, timeout: 30);

// ✅ 推荐：简单、单一职责
$task = new Task(fn($args) => processData($args['data']));

// ❌ 避免：复杂业务逻辑
$task = new Task(function($args) {
    // 大量代码...
});

try {
    $runtime = new Runtime('/invalid/path.php');
} catch (ParallelException $e) {
    echo "错误: " . $e->getMessage() . "\n";
}

$runtime = new Runtime();
try {
    $result = $runtime->run($task)->get();
} finally {
    $runtime->close();
}

// ❌ 不行
$task = new Task(function() { yield 1; });

// ✅ 改用返回数组
$task = new Task(function() { return [1, 2, 3]; });

// ✅ 使用 Channel
$channel = Channel::make();
$runtime->run(fn($args) => $args['ch']->send($data), ['ch' => $channel]);