zoukankan      html  css  js  c++  java
  • 用swoole实现异步任务队列

    应用场景如下:

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

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

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

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

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

    服务端

    第一步:创建tcp服务器

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

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

    具体代码如下:tcp_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'));
        $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 ";
      }
      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

    <?php
    class Client{
      public $client;
      public function __construct(){
        $this->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';
    }
    ?>

  • 相关阅读:
    遍历及线索化二叉树
    二叉树
    程序的内存布局
    C语言一些易混淆的概念
    C语言标准库函数memcpy和memmove的区别以及内存重叠问题处理
    柔性数组
    一个基于QT简单登录对话框(带验证码功能)
    Qt中的布局管理器
    Qt中的标准对话框
    一个基于QT简单登录对话框
  • 原文地址:https://www.cnblogs.com/myJuly/p/12686250.html
Copyright © 2011-2022 走看看