Swoole來實現實時異步任務隊列

假如要發100封郵件，for循環100遍，用戶直接揭竿而起，什麼破網站！
但實際上，咱們極可能有超過1萬的郵件。怎麼處理這個延遲的問題？
答案就是用異步。把「發郵件」這個操做封裝，而後後臺異步地執行1萬遍。這樣的話，用戶提交網頁後，他所等待的時間只是「把發郵件任務請求推送進隊列裏」的時間。而咱們的後臺服務將在用戶看不見的地方跑。
在實現「異步隊列」這點上，有人採用MySQL表或者redis來存放待發送的郵件，而後，每分鐘定時讀取待發送列表，而後處理。這即是定時異步任務隊列。但當前提交的任務要一分鐘後才能執行，在某些實時性要求應用場景裏仍是不快。有些場景要求，只有一提交任務，便立刻執行，但用戶不須要等待返回結果。
本文將探討用php擴展swoole實現實時異步任務隊列的方案。

服務端

在打算放置腳本的目錄（你也能夠自行新建）新建Server.php，代碼以下

 1 <?php  
 2   
 3 class Server  
 4 {  
 5     private $serv;  
 6   
 7     public function __construct()  
 8     {  
 9         $this->serv = new swoole_server("0.0.0.0", 9501);  
10         $this->serv->set(array(  
11             'worker_num' => 1, //通常設置爲服務器CPU數的1-4倍  
12             'daemonize' => 1, //以守護進程執行  
13             'max_request' => 10000,  
14             'dispatch_mode' => 2,  
15             'task_worker_num' => 8, //task進程的數量  
16             "task_ipc_mode " => 3, //使用消息隊列通訊，並設置爲爭搶模式  
17             //"log_file" => "log/taskqueueu.log" ,//日誌  
18         ));  
19         $this->serv->on('Receive', array($this, 'onReceive'));  
20         // bind callback  
21         $this->serv->on('Task', array($this, 'onTask'));  
22         $this->serv->on('Finish', array($this, 'onFinish'));  
23         $this->serv->start();  
24     }  
25   
26     public function onReceive(swoole_server $serv, $fd, $from_id, $data)  
27     {  
28         //echo "Get Message From Client {$fd}:{$data}\n";  
29         // send a task to task worker.  
30         $serv->task($data);  
31     }  
32   
33     public function onTask($serv, $task_id, $from_id, $data)  
34     {  
35         $array = json_decode($data, true);  
36         if ($array['url']) {  
37             return $this->httpGet($array['url'], $array['param']);  
38         }  
39     }  
40   
41     public function onFinish($serv, $task_id, $data)  
42     {  
43         //echo "Task {$task_id} finish\n";  
44         //echo "Result: {$data}\n";  
45     }  
46   
47     protected function httpGet($url, $data)  
48     {  
49         if ($data) {  
50             $url .= '?' . http_build_query($data);  
51         }  
52         $curlObj = curl_init(); //初始化curl，  
53         curl_setopt($curlObj, CURLOPT_URL, $url); //設置網址  
54         curl_setopt($curlObj, CURLOPT_RETURNTRANSFER, 1); //將curl_exec的結果返回  
55         curl_setopt($curlObj, CURLOPT_SSL_VERIFYPEER, FALSE);  
56         curl_setopt($curlObj, CURLOPT_SSL_VERIFYHOST, FALSE);  
57         curl_setopt($curlObj, CURLOPT_HEADER, 0); //是否輸出返回頭信息  
58         $response = curl_exec($curlObj); //執行  
59         curl_close($curlObj); //關閉會話  
60         return $response;  
61     }  
62   
63 }  
64   
65 $server = new Server();

客戶端

啓動服務後，讓咱們看看如何調用服務。新建測試文件Client_test.php

 1 <?php  
 2   
 3 class Client  
 4 {  
 5     private $client;  
 6   
 7     public function __construct()  
 8     {  
 9         $this->client = new swoole_client(SWOOLE_SOCK_TCP);  
10     }  
11   
12     public function connect()  
13     {  
14         if (!$this->client->connect("127.0.0.1", 9501, 1)) {  
15             throw new Exception(sprintf('Swoole Error: %s', $this->client->errCode));  
16         }  
17     }  
18   
19     public function send($data)  
20     {  
21         if ($this->client->isConnected()) {  
22             if (!is_string($data)) {  
23                 $data = json_encode($data);  
24             }  
25   
26             return $this->client->send($data);  
27         } else {  
28             throw new Exception('Swoole Server does not connected.');  
29         }  
30     }  
31   
32     public function close()  
33     {  
34         $this->client->close();  
35     }  
36 }  
37   
38 $data = array(  
39     "url" => "http://192.168.10.19/send_mail",  
40     "param" => array(  
41         "username" => 'test',  
42         "password" => 'test'  
43     )  
44 );  
45 $client = new Client();  
46 $client->connect();  
47 if ($client->send($data)) {  
48     echo 'success';  
49 } else {  
50     echo 'fail';  
51 }  
52 $client->close();

在上面代碼中，url即爲任務所在地址，param爲所需傳遞參數。
保存好代碼，在命令行或者瀏覽器中執行Client_test.php,便實現了異步任務隊列。你所填寫的URL，將會在每次異步任務被提交後，以HTTP GET的方式異步執行。