基于 swoole 下 异步队列和毫秒定时任务 API

• 说明

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

• 当我们使用定时任务时说明:

1. 服务端代码

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

<?php /**
 * Created by PhpStorm
 * User: pl
 * Date: 2020/5/22
 * Time: 15:23
 */


class TaskServer
{
    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();
    }


    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);
            $type = $data['data']['type'];
            $time = intval($data['data']['timing']);
            unset($data['data']['timing']);
            unset($data['data']['type']);
            if (intval($type) == 1) {
                $this->request_curl($data['url'], $data['data'], $data['data']['http_type']);
            } else {
                Swoole\Timer::after($time, function () use ($data) {
                    $this->request_curl($data['url'], $data['data'], $data['data']['http_type']);
                });
            }
        } 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 TaskServer();

2.客户端代码

<?php const API_KEY = 'xxx';

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($_SERVER['REQUEST_METHOD']!='POST')  throw new ErrorException('路由不存在','404');

if(!array_key_exists('HTTP_TOKEN',$_SERVER)){
    header('Content-type: application/json');
    echo json_encode(['code'=>'200007','token is not empty']);exit();
}
if($_SERVER['HTTP_TOKEN']!= API_KEY )  {
    header('Content-type: application/json');
    echo json_encode(['code'=>'200007','error token']);exit();
}


$params = $_POST;


$client = new ClientRequest($params);
$client->connect();

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

基于 swoole 下 异步队列 API 基于 swoole 下 异步队列 API 基于 swoole下 异步队列API

最后补充:基于swoole 一个简单的异步队列就完成了。可以将此队列封装成api队列接口，将它丢到task里面去慢慢执行吧~哈哈
<a name="introduction1"></a>

补充新增的一次定时器任务，支持毫秒级

3.定时任务和异步接口

说明:

1、在swoole 中 毫秒【如 1000 表示 1 秒，v4.2.10 以下版本最大不得超过 86400000【一天】】
2、后台定时器操作接口：操作方法如下
当操作时间在一天内 则直接执行定时器
3、大于一天时
系统执行一个定时crontab任务 ->
每隔12-24小时 运行一次 将数据库执行接口时间大于当前时间且不超过一天的数据数据执行该接口
关 于用户执行撤销则通过推送数据接口去判断该数据是否进行推送
4、当该需要给执行接口添加参数时 直接将参数放入接口

接口文档

ClientRequest.php

请求方法 post

参数

参数	是否必选	备注	限制	新增
token	是	密钥【header】xxxx
url	是	执行接口的地址
type	是	1、异步执行该接口 2、定时执行该接口
http_type	是	get、post
timing	否	定时执行该接口的时间 单位【毫秒】

​