Use of swarm task task

Swoole asynchronous Task, It mainly realizes the execution of calling asynchronous tasks .

Common scenes ： Asynchronous payment processing 、 Asynchronous order processing 、 Asynchronous log processing 、 Send mail asynchronously / SMS, etc. .

Swoole The implementation of worker The process handles data requests , Assigned to task Process execution .

The official introduction ：

task Bottom use Unix Socket Pipeline communication , It's full memory , No, IO Consume . The read-write performance of a single process can reach 100 ten thousand /s, Different processes use different pipes to communicate , Can maximize the use of multi-core .

Local version ：PHP 7.2.6、Swoole 4.3.1.

Not much to say , Look at the renderings first ：

Code

server.php

class Server

{

    private $serv;

 

    public function __construct() {

        $this->serv = new swoole_server('0.0.0.0', 9501);

        $this->serv->set([

            'worker_num'      => 2, // Turn on 2 individual worker process 

            'max_request'     => 4, // Every worker process  max_request Set to 4 Time 

            'task_worker_num' => 4, // Turn on 4 individual task process 

            'dispatch_mode'   => 2, // Packet distribution strategy  -  Fixed mode 

        ]);

 

        $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 . "  Service started ".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 " client :".$fd."  Connected ".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 " client :{$fd}  It's from Email:{$data}".PHP_EOL;

        $param = [

            'fd'    => $fd,

            'email' => $data

        ];

        $rs = $serv->task(json_encode($param));

        if ($rs === false) {

            echo " Task assignment failed  Task ".$rs.PHP_EOL;

        } else {

            echo " Task assigned successfully  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;

 

        // Business code 

        for($i = 1 ; $i <= 5 ; $i ++ ) {

            sleep(2);

            echo "Task {$task_id}  It's done  {$i}/5  The task of ".PHP_EOL;

        }

 

        $data_arr = json_decode($data, true);

        $serv->send($data_arr['fd'] , 'Email:'.$data_arr['email'].', Send successfully ');

        $serv->finish($data);

        echo "########".PHP_EOL.PHP_EOL;

    }

 

    public function onFinish($serv,$task_id, $data) {

        echo "#### onFinish ####".PHP_EOL;

        echo "Task {$task_id}  Completed ".PHP_EOL;

        echo "########".PHP_EOL.PHP_EOL;

    }

 

    public function onClose($serv, $fd) {

        echo "Client Close.".PHP_EOL;

    }

}

 

$server = new Server();

client.php

<?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, " Input Email:");

        swoole_event_add(STDIN, function() {

            fwrite(STDOUT, " Input 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();