swoole的应用场景：实现异步任务队列

应用场景

假如要发100封邮件，for循环100遍，这种方法显然是不可取的。

在一些比较繁杂的业务里，我们很可能有超过1万的邮件要群发。那我们怎么处理这个延迟的问题？

答案就是用异步。把“发邮件”这个操作封装，然后后台异步地执行1万遍。这样的话，用户提交网页后，他所等待的时间只是“把发邮件任务请求推送进队列里”的时间。而我们的后台服务将在用户看不见的地方跑。

在实现“异步队列”这点上，有人采用MySQL表或者redis来存放待发送的邮件，然后，每分钟定时读取待发送列表，然后处理。这便是定时异步任务队列。但当前提交的任务要一分钟后才能执行，在某些实时性要求高的应用场景里还是不快，比如发送短信的场景，只要一提交任务，便要马上执行，用户不需要等待返回结果。

以下将探讨用php扩展swoole实现实时异步任务队列发送短信的方案。

服务端

第一步：创建tcp服务器

第二步：设置服务器的相关属性

第三步：设置服务端的相关回调函数处理任务

具体代码如下：tcp_server.php

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'));
 $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){
 $serv->task($data);
 }
 public function onTask($serv, $task_id, $from_id, $data){
 $data = json_decode($data,true);
 if(!empty($data)){
 return $this->sendsms($data['mobile'],$data[' message ']); 
 }
 }
 public function onFinish($serv, $task_id, $data){
 echo "Task {$task_id} finish\n";
 }
 public function sendsms($mobile,$text)
 {
 $timestamp = date("Y-m-d H-i-s");
 $pid = "888888888";
 $send_sign = md5($pid.$timestamp."abcdefghijklmnopqrstuvwxyz");
 $post_data = array(); 
 $post_data['partner_id'] = $pid; 
 $post_data['timestamp'] =$timestamp; 
 $post_data['mobile'] = $mobile; 
 $post_data['message'] = $text; 
 $post_data['sign'] = $send_sign; 
 $url=' 
 $o=""; 
 forea ch  ($post_data as $k=>$v) 
 { 
 $o.= "$k=". urlencode ($v)."&"; 
 } 
 $post_data=substr($o,0,-1); 
 $ch = curl_init(); 
 curl_setopt($ch, CURLOPT_POST, 1); 
 curl_setopt($ch, CURLOPT_HEADER, 0); 
 curl_setopt($ch, CURLOPT_URL,$url); 
 //为了支持 cookie  
 //curl_setopt($ch, CURLOPT_COOKIEJAR, 'cookie.txt'); 
 curl_setopt($ch, CURLOPT_POSTFIELDS, $post_data); 
 curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
 $result = curl_exec($ch); 
 if(strpos($result,"success")!==false)
 {
 $outstr=1;
 }
 else
 {
 $outstr=502;
 }
 return $outstr;
 }
}
$server = new Server();
?>
 

客户端

启动后端服务后，客户端首先创建tcp客户端服务器，然后连接tcp后端服务器，并向后端tcp服务器发送数据，具体代码如下：client.php

client= new swoole_client(SWOOLE_SOCK_TCP);//默认同步tcp客户端，添加参数SWOOLE_SOCK_ASYNC为异步
 }
 public function  connect (){
 if(!$this->client->connect('127.0.0.1',9501,1)){
 throw new Exception(sprintf('Swoole Error: %s', $this->client->errCode));
 }
 }
 public function send($data){
 if($this->client->isConnected()){
 $data = json_encode($data);
 //print $data; 
 if($this->client->send($data)){
 return 1; 
 }else{
 throw new Exception(sprintf('Swoole Error: %s', $this->client->errCode));
 }
 }else{
 throw new Exception('Swoole Server does not connected.'); 
 }
 }
 public function close(){
 $this->client->close();
 }
}
$client= new Client();
$client->connect();
$data=array(
 'mobile'=>'18511487955',
 'message'=>'you mobile 18511487955'
);
if($client->send($data)){
 echo 'succ';
}else{
 echo 'fail';
}
?>