基于 swoole 下异步消息队列 API

更新时间:2023-04-08 13:12:57 阅读：评论：0

说明

1. 在 rver 程序中如果需要执行很耗时的操作，比如一个聊天服务器发送广播，web 服务器中发送邮件。如果直接去执行这些函数就会阻塞当前进程，导致服务器响应变慢。

swoole 提供了异步任务处理的功能，可以投递一个异步任务到taskworker进程池中执行，不影响当前请求的处理速度。(官网说明)

1. 服务端代码

执行服务端监听端口
9501。通过设置
daemonize这个参数，以守护进程在系统去维护这个
taskworker进程池。我们客户端将消息传递给服务端，服务端异步将数据请求放入进程池队列运行，从大大缩短了响应时间。

<?php/** * created by phpstorm * ur: pl * date: 2020/4/26 * time: 10:29 */class taskrvers{    private $rver;    public function __construct()    {        $this->rver = new swoole\rver('127.0.0.1',9501);        $this->rver->t([            'task_worker_num' => 3,     //开启的进程数 一般为cup核数 1-4倍            'dae电脑安全monize'       => 1,     //已守护进程执行该程序            'max_request'     => 10000,  //worker进程最大任务数            'dispatch_mode' => 2,        //设置为争抢模式            'task_ipc_mode'   => 3,     //设置为消息队列模式        ]);        $this->rver->on('receive',array($this,'onreceive'));        $this->rver->on('task',array($this,'ontask'));        $this->rver->on('finish',array($this,'onfinish'));        $this->rver->s会计工作年度总结tart();    }    /**     * @param swoole_rver $rver     * @param $fd     * @param $form_id     * @param $data     *  开始投递异步任务     */    public function onreceive(swoole_rver $rver , $fd , $form_id , $data)    {        $this->rver->task($data);    }    /**     * @param swoole_rver $rver     * @param $fd     * @param $from_id     * @param $data     *  执行异步任务     */    public function ontask( $rver , $fd , $from_id , $data)    {        $data = json_decode($data,true);        try {            $log_txt =  date('y-m-d h:i:s')."开始执行任务".php_eol ;            $this->log($log_txt);            return  $this->request_curl($data['url'],$data['data'],'post');        }catch (\exception $exception){            $log_txt =  date('y-m-d h:i:s')."执行任务失败发生错误".php_eol ;            $this->log($log_txt);        }    }    public function onfinish( $rver , $task_id, $data)    {        $log_txt =  date('y-m-d h:i:s')."$data".php_eol ;        $this->log($log_txt);    }   public function request_curl($url = '', $request_data = '', $request_type = 'get', $headers = [], $is_ssl = fal)    {            $ch = curl_init (); //curl初始化            if( $request_type == 'get' && !empty( $request_data) )            {                $num = 0;                foreach ( $request_data as $key => $value )                {                    if($num == 0)                    {                        $url .= '?' . $key.'='.$value;                    }                    el{                        $url .= '&'. $key . '=' . $value;                      }                    $num ++;                }                $num = 0;            }            //区分get和post            curl_topt ( $ch, curlopt_url, $url ); //url地址            curl_topt ( $ch, curlopt_header, 0 ); //头信息不输出            //如果成功只将结果返回，不自动输出任何内容            curl_topt ( $ch, curlopt_returntransfer, 1 );              //post类型就实现此结果            if( $request_type == 'post')            {               //设置为post方式                curl_topt ( $ch, curlopt_post, 1 );                  //post数据                curl_topt ( $ch, curlopt_postfields, $request_data );                  //当post数据大于1024时强制执行                curl_topt ( $ch, curlopt_httpheader, array("expect:"));             }            //判断是否绕过证书            if( $is_ssl )            {                curl_topt($ch, curlopt_ssl_verifypeer, fal);//绕过ssl验证                curl_topt($ch, curlopt_ssl_verifyhost, fal);            }            if(!empty($headers))  curl_topt($ch, curlopt_httpheader, $headers);            $result = curl_exec 喜居宝地千年旺 福照家门万事兴横批是什么( $ch ); //执行            if ( $result == fal) return fal;            curl_clo ( $ch ); //关闭资源    帮派宣言    return $result;    }    public function log($log_txt)    {        $log ='log/'.date('y_m_d').'log';        if(!file_exists($log)) {            touch($log);            chown($log,0777);        } 安徽财贸职业学院       $file_log = fopen($log,"a");        fputs($file_log,$log_txt);        fclo($file_log);    }}$task = new taskrvers();

2. 客户端代码

<?php/** * created by phpstorm * ur: pl * date: 2020/4/26 * time: 11:54 */class clientrequest{    private $client;    private $params; //请求参数    public function __construct($params)    {        $this->client = new swoole_client(swoole_sock_tcp | swoole_keep);        $this->params = $params;    }    public function connect()    {        if (!$this->client->connect('127.0.0.1', 9501, 1)) {            return json_encode([                'code' => 500,                'err_msg' => '链接异步客户端失败'            ]);        }        /**         * 注意请求格式         * $params['url'] 接口地址         * $params['type']接口请求方式         * $params['data']参数         */        $params = $this->params;        $array['url'] = $params['url'];        unt($params['url']);        $array['data'] = $params;        $this->client->nd(json_encode($array, json_unescaped_unicode));    }}if (!empty($_get)) {    $params = $_get;}if (!empty($_post)) {    $params = $_post;}$client = new clientrequest($params);$client->connect();