Heim >Backend-Entwicklung >PHP-Tutorial >关于异步任务队列之PHP使用swoole来实现实时异步任务队列。

关于异步任务队列之PHP使用swoole来实现实时异步任务队列。

WBOY
WBOYOriginal
2016-06-20 12:30:221230Durchsuche

背景:关于PHP的异步回调啥的,其实PHP这种语言的运行机制想优雅实现是不可能的,有一个叫swoole的扩展实现这个队列的处理是很不错的,之前有张宴兄弟写过一个叫https的队列处理(基于libevent),后面有韩天峰兄弟写的swoole,更注重了异步IO实现对CPU的IO的吃满(基于自己编写的epoll加队列链表内存分配一堆东西,反正我是看过没看明白有空再研究),但是,作为swoole的粉丝兼顾问,异步问题在PHP业界的一个需求量还是蛮大的,尤其是日志异步写、url访问、邮件异步发、跨机房db特殊的小同步、审核异步队列、框架底层对接口访问排查错误的db和cache接口查询及返回,这些目前对于大并发网站都是需要异步来解决的,但是异步归异步,回调这块也实现异步回调(真大并发想知道结果可能阻塞了PHP进程产生进程等待异步返回而新来的连接没法及时处理的php-fpm进程性雪崩),在实际运用中常规处理办法是开一个新的端口,下面的swoole也是开了新的端口来处理,于httpsqs不同在于swoole可以把简单的curl啥的逻辑也可封装里面,在httpsqs里只是纯队列,一个投递进来,再起一个php的daemon进行读取队列,因为如果真是繁忙,导致异步处理返回慢,这个等待也是太漫长,我觉得看具体业务而看要不要等待,在实际中异步和队列大都用于抛数据以及解耦,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,代码如下:

<?phpclass 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

代码如下:

<?phpclass 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

摘自第七星尘的博客:http://blog.star7th.com/2016/01/1905.html

Stellungnahme:
Der Inhalt dieses Artikels wird freiwillig von Internetnutzern beigesteuert und das Urheberrecht liegt beim ursprünglichen Autor. Diese Website übernimmt keine entsprechende rechtliche Verantwortung. Wenn Sie Inhalte finden, bei denen der Verdacht eines Plagiats oder einer Rechtsverletzung besteht, wenden Sie sich bitte an admin@php.cn