PHP使用swoole来实现实时异步任务队列

时间:2022-08-27 15:40:51

转载来自第七星尘的技术博客《PHP使用swoole来实现实时异步任务队列》

关于异步任务队列

用户打开了我们的网站。他要做的就是勾选需要发邮件的代理商列表,然后把结算邮件发出去。
假如我们需要发1封邮件,我们写个函数执行即可。考虑到网络可能会稍微有点延迟,但是是可以接受的,用户会乖乖等你的网页发完邮件了再关闭网页。
假如我们要发布10封邮件,用一个for循环,循环10遍执行发邮件操作。这时候,也许10倍的网络延迟会让用户稍微有点不耐烦,但勉强可以等吧。
假如要发100封邮件,for循环100遍,用户直接揭竿而起,什么破网站!
但实际上,我们很可能有超过1万的邮件。怎么处理这个延迟的问题?
答案就是用异步。把“发邮件”这个操作封装,然后后台异步地执行1万遍。这样的话,用户提交网页后,他所等待的时间只是“把发邮件任务请求推送进队列里”的时间。而我们的后台服务将在用户看不见的地方跑。
在实现“异步队列”这点上,有人采用mysql表或者redis来存放待发送的邮件,然后,每分钟定时读取待发送列表,然后处理。这便是定时异步任务队列。但当前提交的任务要一分钟后才能执行,在某些实时性要求应用场景里还是不快。有些场景要求,只有一提交任务,便马上执行,但用户不需要等待返回结果。
在云平台SAE和BAE上,都有taskqueue服务来解决上面的问题。而如果是自己假设服务器,则如何解决?本文将探讨用php扩展swoole实现实时异步任务队列的方案。

安装swoole

pecl 安装:
pecl install swoole

看命令行提示,如果它提示说没有写php.ini,则自己手动在PHP.ini后面加上:
extension = "swoole.so"

服务端

在打算放置脚本的目录(你也可以自行新建)新建Server.php,代码如下:

<?php class Server { private $serv; public function __construct() { $this->serv = new swoole_server("0.0.0.0", 9501); $this->serv->set(array( 'worker_num' => 1, //一般设置为服务器CPU数的1-4倍 'daemonize' => 1, //以守护进程执行 'max_request' => 10000, 'dispatch_mode' => 2, 'task_worker_num' => 8, //task进程的数量 "task_ipc_mode " => 3 , //使用消息队列通信,并设置为争抢模式 //"log_file" => "log/taskqueueu.log" ,//日志 )); $this->serv->on('Receive', array($this, 'onReceive')); // bind callback $this->serv->on('Task', array($this, 'onTask')); $this->serv->on('Finish', array($this, 'onFinish')); $this->serv->start(); } public function onReceive( swoole_server $serv, $fd, $from_id, $data ) { //echo "Get Message From Client {$fd}:{$data}n"; // send a task to task worker. $serv->task( $data ); } public function onTask($serv,$task_id,$from_id, $data) { $array = json_decode( $data , true ); if ($array['url']) { return $this->httpGet( $array['url'] , $array['param'] ); } } public function onFinish($serv,$task_id, $data) { //echo "Task {$task_id} finishn"; //echo "Result: {$data}n"; } protected function httpGet($url,$data){ if ($data) { $url .='?'.http_build_query($data) ; } $curlObj = curl_init(); //初始化curl, curl_setopt($curlObj, CURLOPT_URL, $url); //设置网址 curl_setopt($curlObj, CURLOPT_RETURNTRANSFER, 1); //将curl_exec的结果返回 curl_setopt($curlObj, CURLOPT_SSL_VERIFYPEER, FALSE); curl_setopt($curlObj, CURLOPT_SSL_VERIFYHOST, FALSE); curl_setopt($curlObj, CURLOPT_HEADER, 0); //是否输出返回头信息 $response = curl_exec($curlObj); //执行 curl_close($curlObj); //关闭会话 return $response; } } $server = new Server(); 

由于服务端是异步、常驻内存的,因此必须通过命令行来启动。在命令行执行以上代码以启动服务
php Server.php
执行完毕后关闭命令行窗口即可。服务会在后台以守护进程运行

客户端

启动服务后,让我们看看如何调用服务。新建测试文件Client_test.php
代码如下:

<?php class Client { private $client; public function __construct() { $this->client = new swoole_client(SWOOLE_SOCK_TCP); } public function connect() { if( !$this->client->connect("127.0.0.1", 9501 , 1) ) { echo "Connect Error"; } $data = array( "url" => "http://192.168.10.19/send_mail" , "param" => array( "username"=>'test', "password" => 'test' ) ); $json_data = json_encode($data); $this->client->send( $json_data ); } } $client = new Client(); $client->connect(); 

在上面代码中,url即为任务所在地址,param为所需传递参数。
保存好代码,在命令行或者浏览器中执行Client_test.php,便实现了异步任务队列。你所填写的URL,将会在每次异步任务被提交后,以HTTP GET的方式异步执行。

查看与关闭

swoole好像没有很便捷的关闭方式。所以只能直接通过关闭进程来关闭。
查看命令:
ps -ef | grep php
结束单个进程:
kill -9 {进程号}
结束所有进程的命令:
killall -9 php