-

Swoole Task异步任务介绍

swoole 的异步任务task系统可以很方便的为我们在开发的过程中调用异步任务的执行，而无需等待。

常见使用场景：

task模块用来做一些异步的慢速任务，比如webim中发广播，发送邮件，异步订单处理、异步支付处理等。

  • task进程必须是同步阻塞的
  • task进程支持定时器

node.js 假如有10万个连接，要发广播时，那会循环10万次，这时候程序不能做任何事情，不能接受新的连接，也不能收包发包。

而swoole不同，丢给task进程之后，worker进程可以继续处理新的数据请求。任务完成后会异步地通知worker进程告诉它此任务已经完成。

当然task模块的作用还不仅如此，实现PHP的数据库连接池，异步队列等，还需要进一步挖掘。

简单实例：

$serv = new Swoole\Server("127.0.0.1", 9502);
$serv->set(array('task_worker_num' => 4));
$serv->on('Receive', function($serv, $fd, $from_id, $data) {
    $task_id = $serv->task("Async");
    echo "Dispath AsyncTask: id=$task_id\n";
});
$serv->on('Task', function ($serv, $task_id, $from_id, $data) {
    echo "New AsyncTask[id=$task_id]".PHP_EOL;
    $serv->finish("$data -> OK");
});
$serv->on('Finish', function ($serv, $task_id, $data) {
    echo "AsyncTask[$task_id] Finish: $data".PHP_EOL;
});
$serv->start();

swoole_server task异步任务介绍

投递一个异步任务到task_worker池中。此函数是非阻塞的，执行完毕会立即返回。worker进程可以继续处理新的请求。

int swoole_server::task(mixed $data, int $dst_worker_id = -1) 
$task_id = $serv->task("some data");
//swoole-1.8.6或更高版本
$serv->task("taskcallback", -1, function (swoole_server $serv, $task_id, $data) {
    echo "Task Callback: ";
    var_dump($task_id, $data);
});
  • $data要投递的任务数据，可以为除资源类型之外的任意PHP变量
  • $dst_worker_id可以制定要给投递给哪个task进程，传入ID即可，范围是0 - (serv->task_worker_num -1)
  • 调用成功，返回值为整数$task_id，表示此任务的ID。如果有finish回应，onFinish回调中会携带$task_id参数
  • 调用失败，返回值为false
  • 未指定目标Task进程，调用task方法会判断Task进程的忙闲状态，底层只会向处于空闲状态的Task进程投递任务
  • 1.8.6版本增加了第三个参数，可以直接设置onFinish函数，如果任务设置了回调函数，Task返回结果时会直接执行制定的回调函数，不再执行Server的onFinish回调

$dst_worker_id在1.6.11+后可用，默认为随机投递
$task_id是从0-42亿的整数，在当前进程内是唯一的
task方法不能在task进程/用户自定义进程中调用

此功能用于将慢速的任务异步地去执行，比如一个聊天室服务器，可以用它来进行发送广播。当任务完成时，在task进程中调用$serv->finish("finish")告诉worker进程此任务已完成。当然swoole_server->finish是可选的。

task底层使用Unix Socket管道通信，是全内存的，没有IO消耗。单进程读写性能可达100万/s，不同的进程使用不同的管道通信，可以最大化利用多核。

AsyncTask功能在1.6.4版本增加，默认不启动task功能，需要在手工设置task_worker_num来启动此功能
task_worker的数量在swoole_server::set参数中调整，如task_worker_num => 64，表示启动64个进程来接收异步任务

配置参数

swoole_server->task/taskwait/finish 3个方法当传入的$data数据超过8K时会启用临时文件来保存。当临时文件内容超过server->package_max_length 时底层会抛出一个警告。

WARN: task package is too big.

server->package_max_length 默认为2M

注意事项

  • 使用swoole_server_task必须为Server设置onTask和onFinish回调，否则swoole_server->start会失败
  • task操作的次数必须小于onTask处理速度，如果投递容量超过处理能力，task会塞满缓存区，导致worker进程发生阻塞。worker进程将无法接收新的请求
  • 使用addProcess添加的用户进程中无法使用task投递任务，请使用sendMessage接口与工作进程通信

Swoole taskwait

函数原型：

string $result = swoole_server->taskwait(mixed $task_data, float $timeout = 0.5, int $dst_worker_id = -1);

taskwait与task方法作用相同，用于投递一个异步的任务到task进程池去执行。与task不同的是taskwait是阻塞等待的，直到任务完成或者超时返回。

$result为任务执行的结果，由$serv->finish函数发出。如果此任务超时，这里会返回false。

taskwait是阻塞接口，如果你的Server是全异步的请使用swoole_server::task和swoole_server::finish,不要使用taskwait
第3个参数可以制定要给投递给哪个task进程，传入ID即可，范围是0 - serv->task_worker_num
$dst_worker_id在1.6.11+后可用，默认为随机投递
taskwait方法不能在task进程中调用

onTask

在task_worker进程内被调用。worker进程可以使用swoole_server_task函数向task_worker进程投递新的任务。当前的Task进程在调用onTask回调函数时会将进程状态切换为忙碌，这时将不再接收新的Task，当onTask函数返回时会将进程状态切换为空闲然后继续接收新的Task。

function onTask((swoole_server $serv, int $task_id, int $src_worker_id, string $data));
  • $task_id是任务ID，由swoole扩展内自动生成，用于区分不同的任务。$task_id和$src_worker_id组合起来才是全局唯一的，不同的worker进程投递的任务ID可能会有相同
  • $src_worker_id来自于哪个worker进程
  • $data 是任务的内容

返回执行结果到worker进程

1.7.2以上的版本，在onTask函数中 return字符串，表示将此内容返回给worker进程。worker进程中会触发onFinish函数，表示投递的task已完成。

  • return的变量可以是任意非null的PHP变量

1.7.2以前的版本，需要调用swoole_server->finish()函数将结果返回给worker进程


© 2023 mianshi8.net MIT license