在 AMPHP Parallel 中优雅处理超时并获取成功任务结果

2024-03-20 13:32:47

在 AMPHP Parallel 中处理超时和获取成功任务结果

引言

在使用 AMPHP Parallel 库进行并发处理时，有时需要为任务设置超时以处理长时间运行或失败的情况。然而，在不破坏任务逻辑的情况下获取成功任务的结果可能是一项挑战。本文将探讨一种使用通道来解决此问题的创新方法。

背景：TimeoutCancellation 和异常处理

AMPHP Parallel 库的 TimeoutCancellation 可用于取消在指定时间内未完成的任务。但是，当任务的 run 方法包含 try {} catch (Exception $e) {} 块时，超时取消可能会抛出 CancellationException。如果 try 块中有一个慢任务，这可能会破坏逻辑并导致错误的结果。

使用通道解决问题

为了在不删除 try {} catch (Exception $e) {} 块的情况下解决这个问题，我们可以使用通道。通道是一种同步通信机制，它允许两个协程在没有锁或条件变量的情况下交换数据。

具体而言，我们可以创建一个通道，并在每个任务的 run 方法中向该通道发送成功或失败结果。然后，我们可以在外部协程中从通道接收结果，并根据需要处理超时取消。

代码示例

以下代码片段展示了使用通道解决此问题的方法：

use Amp\Parallel\Parallel;
use Amp\Parallel\Worker;
use Amp\Promise;
use Amp\CancelledException;
use Amp\Channel;

// 创建一个通道
$channel = new Channel(10);

// 创建任务列表
$tasks = [
    new Worker(function () use ($channel) {
        try {
            // 任务逻辑
            $result = 'Success';
            $channel->send($result);
        } catch (Exception $e) {
            $channel->send($e);
        }
    }),
    new Worker(function () use ($channel) {
        try {
            // 任务逻辑
            $result = 'Success';
            $channel->send($result);
        } catch (Exception $e) {
            $channel->send($e);
        }
    }),
    new Worker(function () use ($channel) {
        try {
            // 任务逻辑
            sleep(10); // 模拟长时间运行的任务
            $result = 'Success';
            $channel->send($result);
        } catch (Exception $e) {
            $channel->send($e);
        }
    }),
];

// 创建一个任务超时为 5 秒的 Parallel 实例
$parallel = new Parallel(5);

// 将任务添加到并行实例
$results = [];
foreach ($tasks as $task) {
    $promise = Promise\async(function () use ($parallel, $task) {
        try {
            // 在并行实例中运行任务
            $result = yield $parallel->run($task);
            return $result;
        } catch (CancelledException $e) {
            // 任务超时
            return 'Timeout';
        }
    });
    $results[] = $promise;
}

// 从通道接收结果
$responses = [];
while ($result = $channel->receive(0.1)) {
    $responses[] = $result;
}

// 处理结果
foreach ($results as $key => $result) {
    if ($responses[$key] instanceof Exception) {
        // 任务失败
        echo "Task $key failed with exception: " . $responses[$key]->getMessage() . "\n";
    } else {
        // 任务成功
        echo "Task $key succeeded with result: " . $responses[$key] . "\n";
    }
}