欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

基于 swoole 下 异步消息队列 API

程序员文章站 2022-07-01 15:51:59
说明 1. 在 Server 程序中如果需要执行很耗时的操作,比如一个聊天服务器发送广播,Web 服务器中发送邮件。如果直接去执行这些函数就会阻塞当前进程,导致服务器响应变慢。 Swoole 提供了异步任务处理的功能,可以投递一个异步任务到 TaskWorker 进程池中执行,不影响当前请求的处理速 ......

说明

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

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

1. 服务端代码

执行服务端监听端口 9501。通过设置 daemonize 这个参数,以守护进程在系统去维护这个 taskworker 进程池。我们客户端将消息传递给服务端,服务端异步将数据请求放入进程池队列运行,从大大缩短了响应时间。
<?php
/**
 * created by phpstorm
 * user: pl
 * date: 2020/4/26
 * time: 10:29
 */


class taskservers
{
    private $server;

    public function __construct()
    {

        $this->server = new swoole\server('127.0.0.1',9501);

        $this->server->set([
            'task_worker_num' => 3,     //开启的进程数 一般为cup核数 1-4倍
            'daemonize'       => 1,     //已守护进程执行该程序
            'max_request'     => 10000,  //worker进程最大任务数
            'dispatch_mode' => 2,        //设置为争抢模式
            'task_ipc_mode'   => 3,     //设置为消息队列模式
        ]);
        $this->server->on('receive',array($this,'onreceive'));
        $this->server->on('task',array($this,'ontask'));
        $this->server->on('finish',array($this,'onfinish'));
        $this->server->start();
    }

    /**
     * @param swoole_server $server
     * @param $fd
     * @param $form_id
     * @param $data
     *  开始投递异步任务
     */
    public function onreceive(swoole_server $server , $fd , $form_id , $data)
    {
        $this->server->task($data);
    }

    /**
     * @param swoole_server $server
     * @param $fd
     * @param $from_id
     * @param $data
     *  执行异步任务
     */
    public function ontask( $server , $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( $server , $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 = false)
    {
            $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;
                    }
                    else{
                        $url .= '&'. $key . '=' . $value;
                      }
                    $num ++;
                }
                $num = 0;
            }
            //区分get和post
            curl_setopt ( $ch, curlopt_url, $url ); //url地址
            curl_setopt ( $ch, curlopt_header, 0 ); //头信息不输出
            //如果成功只将结果返回,不自动输出任何内容
            curl_setopt ( $ch, curlopt_returntransfer, 1 );  
            //post类型就实现此结果
            if( $request_type == 'post')
            {
               //设置为post方式
                curl_setopt ( $ch, curlopt_post, 1 ); 
                 //post数据
                curl_setopt ( $ch, curlopt_postfields, $request_data );  
                //当post数据大于1024时强制执行
                curl_setopt ( $ch, curlopt_httpheader, array("expect:")); 
            }
            //判断是否绕过证书
            if( $is_ssl )
            {
                curl_setopt($ch, curlopt_ssl_verifypeer, false);//绕过ssl验证
                curl_setopt($ch, curlopt_ssl_verifyhost, false);
            }
            if(!empty($headers))  curl_setopt($ch, curlopt_httpheader, $headers);
            $result = curl_exec ( $ch ); //执行
            if ( $result == false) return false;
            curl_close ( $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);
        fclose($file_log);
    }
}
$task = new taskservers();

  

2. 客户端代码

<?php
/**
 * created by phpstorm
 * user: 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'];
        unset($params['url']);
        $array['data'] = $params;
        $this->client->send(json_encode($array, json_unescaped_unicode));
    }
}
if (!empty($_get)) {
    $params = $_get;
}
if (!empty($_post)) {
    $params = $_post;
}
$client = new clientrequest($params);
$client->connect();

  

开始执行服务端程序,php taskserver.php 我们以接口形式上去调用另外一个耗时接口。简单对比一下响应速度。

 

 

基于 swoole 下 异步消息队列 API

 

 

 

基于 swoole 下 异步消息队列 API

更多学习内容请访问:

腾讯t3-t4标准精品php架构师教程目录大全,只要你看完保证薪资上升一个台阶(持续更新)