swoole task怎么用

class Server
{
 private $serv;

 public function __construct() {
 $this->serv = new swoole_server('0.0.0.0', 9501);
 $this->serv->set([
 'worker_num' => 2, //開啟2個worker進(jìn)程
 'max_request' => 4, //每個worker進(jìn)程 max_request設(shè)置為4次
 'task_worker_num' => 4, //開啟4個task進(jìn)程
 'dispatch_mode' => 2, //數(shù)據(jù)包分發(fā)策略 - 固定模式
 ]);

 $this->serv->on('Start', [$this, 'onStart']);
 $this->serv->on('Connect', [$this, 'onConnect']);
 $this->serv->on("Receive", [$this, 'onReceive']);
 $this->serv->on("Close", [$this, 'onClose']);
 $this->serv->on("Task", [$this, 'onTask']);
 $this->serv->on("Finish", [$this, 'onFinish']);

 $this->serv->start();
 }

 public function onStart($serv) {
 echo "#### onStart ####".PHP_EOL;
 echo "SWOOLE ".SWOOLE_VERSION . " 服務(wù)已啟動".PHP_EOL;
 echo "master_pid: {$serv->master_pid}".PHP_EOL;
 echo "manager_pid: {$serv->manager_pid}".PHP_EOL;
 echo "########".PHP_EOL.PHP_EOL;
 }

 public function onConnect($serv, $fd) {
 echo "#### onConnect ####".PHP_EOL;
 echo "客戶端:".$fd." 已連接".PHP_EOL;
 echo "########".PHP_EOL.PHP_EOL;
 }

 public function onReceive($serv, $fd, $from_id, $data) {
 echo "#### onReceive ####".PHP_EOL;
 echo "worker_pid: {$serv->worker_pid}".PHP_EOL;
 echo "客戶端:{$fd} 發(fā)來的Email:{$data}".PHP_EOL;
 $param = [
 'fd' => $fd,
 'email' => $data
 ];
 $rs = $serv->task(json_encode($param));
 if ($rs === false) {
 echo "任務(wù)分配失敗 Task ".$rs.PHP_EOL;
 } else {
 echo "任務(wù)分配成功 Task ".$rs.PHP_EOL;
 }
 echo "########".PHP_EOL.PHP_EOL;
 }

 public function onTask($serv, $task_id, $from_id, $data) {
 echo "#### onTask ####".PHP_EOL;
 echo "#{$serv->worker_id} onTask: [PID={$serv->worker_pid}]: task_id={$task_id}".PHP_EOL;

 //業(yè)務(wù)代碼
 for($i = 1 ; $i <= 5 ; $i ++ ) {
 sleep(2);
 echo "Task {$task_id} 已完成了 {$i}/5 的任務(wù)".PHP_EOL;
 }

 $data_arr = json_decode($data, true);
 $serv->send($data_arr['fd'] , 'Email:'.$data_arr['email'].',發(fā)送成功');
 $serv->finish($data);
 echo "########".PHP_EOL.PHP_EOL;
 }

 public function onFinish($serv,$task_id, $data) {
 echo "#### onFinish ####".PHP_EOL;
 echo "Task {$task_id} 已完成".PHP_EOL;
 echo "########".PHP_EOL.PHP_EOL;
 }

 public function onClose($serv, $fd) {
 echo "Client Close.".PHP_EOL;
 }
}

$server = new Server();

<?php
class Client
{
 private $client;
 public function __construct() {
 $this->client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
 $this->client->on('Connect', [$this, 'onConnect']);
 $this->client->on('Receive', [$this, 'onReceive']);
 $this->client->on('Close', [$this, 'onClose']);
 $this->client->on('Error', [$this, 'onError']);
 }
 public function connect() {
 if(!$fp = $this->client->connect("127.0.0.1", 9501 , 1)) {
 echo "Error: {$fp->errMsg}[{$fp->errCode}]".PHP_EOL;
 return;
 }
 }
 public function onConnect($cli) {
 fwrite(STDOUT, "輸入Email:");
 swoole_event_add(STDIN, function() {
 fwrite(STDOUT, "輸入Email:");
 $msg = trim(fgets(STDIN));
 $this->send($msg);
 });
 }
 public function onReceive($cli, $data) {
 echo PHP_EOL."Received: ".$data.PHP_EOL;
 }
 public function send($data) {
 $this->client->send($data);
 }
 public function onClose($cli) {
 echo "Client close connection".PHP_EOL;
 }
 public function onError() {
 }
}
$client = new Client();
$client->connect();