php +swoole实现异步任务队列

时间:2023-09-18 19:07
假如要发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='http://182.92.149.100/sendsms';  $o="";  foreach ($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';
}
?>

Copyright© 黑海资讯

鄂ICP备2023014923号