phalcon: queueing使用心得,需要安装相应的软体

http://flyhighest.com/archives/50

原本没有用过phalcon的消息队列，本来以为很简单，结果搞了半天，把步骤记录一下。

1. phalcon的官网上没有说需要安装beanstalkd，我猜写文档的以为我等默认会安装，找了半天没有找到windows下的beanstalk，有个国人写的beanstalkd-win，没敢用。
2. 找了一台Ubuntu，安装上了beanstalkd，命令： sudo apt-get install beanstalkd
3. 启动：beanstalkd -l 10.0.1.5 -p 11300
4. 打开11300端口，命令：iptables -A INPUT -ptcp –dport 11300 -j ACCEPT
5. https://github.com/shruthi-ananth/phalcon-beanstalk 这里有个例子，用来学习phalcon和beanstalk的集成，
   • Once you have Beanstalk server up, you can run producer and worker tasks in the console using ‘php console.php producer main’ and ‘php console.php worker main’. The producer puts a job on the queue and that works fine. The worker(Phalcon implementation) does not pick up a job from the queue to execute unlike the PheanstalkWorker Task.(‘php console.php pheanstalkworker main’). You use ‘php console.php pheanstalkstats main’ to check the server stats and keep both workers running to see which worker picks the job.
6. 把例子程序里的所有beanstalkd的host和端口号改一下。
7. 开四个cmd，一个运行php console.php worker main，一个运行 php console.php pheanstalkworker main， 一个运行php console.php pheanstalkstats main ，最后一个运行php console.php producer main。就可以看到任务写入，任务执行过程了
8. 基本上phalcon的消息队列环境就搭好了。
9. 照着phalcon官网上的例子写，就可以了：https://docs.phalconphp.com/zh/latest/reference/queue.html

Note：

1. 抄了一段别人写的代码,  如果把worker的这段代码用supervisor做个守护进程，就更好了。参考http://www.phpddt.com/php/supervisor.html
   • This is not really Phalcon related but here it goes anyway. This shows a simple example of using the queue to generate a response. Your API endpoint only creates a job in the queue and can return immediately.When the item is picked from the queue the request to the external API is made and when a reponse is recieved it calls the callback supplied by the original callee. This example could be improved by splitting the job in two, one job for external api retrieval and a seperate job for the callback. This way the extarnal Api isn’t called agian when the callback fails. This also allows for multiple callback attempts by keeping a counter for failed deliveries.
   • ApiController

     public function myApiEndpointAction($myArgument, $callbackUrl) {
         $queue->put(array(
             'jobName' => $myArgument,
             'callbackUrl' => $callbackUrl
         ));
     
         return new Reponse("Item put in queue", 201);
     }

     Worker

     while (($job = $queue->peekReady()) !== false) {
     
         $jobBody = $job->getBody();
     
         if(isset($jobBody['jobName']) {
             $myArgument = $jobBody['jobName']; 
             $callbackUrl = $jobBody['callbackUrl']; 
     
             //Call external API
             $external = curl_init();
             curl_setopt_array($external, array(
                 CURLOPT_RETURNTRANSFER => true,
                 CURLOPT_URL => 'http://external.api.com'
             ));
     
             //The reponse from the external API
             $result = curl_exec($external);
             if(!$result) {
                 //External api failed
                 //Don't delete job, will be pickedup again the next round
                 curl_close($external);
                 return;
             }
             curl_close($external);
     
             //Return the response to the original callee
             $callback = curl_init();
             curl_setopt_array($callback, array(
                 CURLOPT_RETURNTRANSFER => true,
                 CURLOPT_URL => $callbackUrl, //The url which the original api caller gave
                 CURLOPT_POST => true,
                 CURLOPT_POSTFIELDS => array(
                     data => $result //The response from the external api call
                 )
             ));
             $deliverd = curl_exec($callback);
             if(!$deliverd) {
                 curl_close($callback);
                 return;
             }
             curl_close($callback);
         }
     
         //Remove the job from the queue
         $job->delete();
     }

PHP进程的生命周期很短，遇到意外情况也会中断，如果跟想要PHP在后台不断的跑脚本，就需重启它。Supervisor是一个python开发的基于*nix上的管理和监控进程的client/server程序。当PHP进程中断，会重新启动它。
安装：

1. wget http://pypi.python.org/packages/source/s/supervisor/supervisor-3.0b1.tar.gz
2. tar –zxvf supervisor–3.0b1.tar.gz
3. cd supervisor–3.0b1
4. easy_install supervisor #安装setuptools(yum -y install python-setuptools)

配置：
在/etc/supervisord.conf后添加：

1. [program:php]
2. command= /usr/local/php54/bin/php –f /root/supervisor/test.php #被监控进程
3. ;process_name=%(process_num)02d
4. ;numprocs=5 #启动几个进程
5. autostart=true #随着supervisord的启动而启动
6. autorestart=true #自动启动
7. startsecs=1 #程序重启时候停留在runing状态的秒数
8. startretries=10 #启动失败时的最多重试次数
9. redirect_stderr=true #重定向stderr到stdout
10. stdout_logfile=/root/supervisor/test.log #stdout文件

命令：
supervisord ：启动supervisor sever
supervisorctl：启动supervisor client的命令行窗口。