关于php:PHP-Swoole-实现异步任务队列

4次阅读

共计 3994 个字符，预计需要花费 10 分钟才能阅读完成。

最近接手一个对接短信的需要，这个需要自身并没有什么难度，间接依照服务商的要求申请具体的接口就好了。

最开始是应用传统的同步阻塞形式实现了一遍，用户体验并不好，发送短信须要期待，期待服务商的接口返回内容，才持续向下执行。

因为最近在学习 Swoole，Swoole 中有一个“异步工作”，就特地适宜以下利用场景：

须要执行耗时操作，会阻塞主过程
用户不须要期待返回后果

联合官网手册和 Latent 的基于 swoole 下异步音讯队列 API，最终简略封装了一个解决 API 的类，实现如下：

服务端是基于本地 Tcp，监听 9501 端口。

<?php
class taskServer{
    const HOST = "127.0.0.1";
    const PORT = 9501;
    public $server = null;

    public function __construct()
    {$this->server = new SWoole\Server(self::HOST, self::PORT);
        $this->server->set(array(
            "enable_coroutine" => false,     // 敞开协程
            "worker_num" => 2,               // 开启的过程数 个别为 cup 核数 1-4 倍
            "task_worker_num" => 2,          // task 过程的数量
            'daemonize' => true,             // 以守护过程的形式启动
        ));

        // 注册事件
        $this->server->on("connect", [$this, "onConnect"]);
        $this->server->on("receive", [$this, "onReceive"]);
        $this->server->on("close", [$this, "onClose"]);
        $this->server->on("task", [$this, "onTask"]);
        $this->server->on("finish", [$this, "onFinish"]);

        // 启用服务
        $this->server->start();}

    /**
     * 监听连贯事件
     * @param $server
     * @param $fd
     */
    public function onConnect($server, $fd){echo "连贯胜利".PHP_EOL;}

    /**
     * 监听客户端发送的音讯
     * @param $server       "Server 对象"
     * @param $fd           "惟一标示"
     * @param $form_id
     * @param $data         "客户端发送的数据"
     */
    public function onReceive($server, $fd, $form_id, $data){
        // 投递工作
        $server->task($data);
        $server->send($fd, "这是客户端向服务端发送的信息：{$data}");
    }

    /**
     * 监听异步工作 task 事件
     * @param $server
     * @param $task_id
     * @param $worker_id
     * @param $data
     * @return string
     */
    public function onTask($server, $task_id, $worker_id, $data){$data = json_decode($data, true);
        echo "开始执行异步工作".PHP_EOL;
        try {
            // 开始执行工作
            $this->addLog(date('Y-m-d H:i:s')."开始执行工作".PHP_EOL );
            // 告诉 worker（必须 return，否则不会调用 onFinish）return $this->curl($data['url'], $data['data'], $data['type']);
        } catch (Exception $exception) {
            // 执行工作失败
            $this->addLog(date('Y-m-d H:i:s')."执行工作失败".PHP_EOL);
        }
    }

    /**
     * 监听 finish 事件
     * @param $server
     * @param $task_id
     * @param $data
     */
    public function onFinish($server, $task_id, $data){$this->addLog(date("Y-m-d H:i:s")."异步工作执行实现".PHP_EOL);
        print_r("来自服务端的音讯：{$data}");
    }

    /**
     * 监听敞开连贯事件
     * @param $server
     * @param $fd
     */
    public function onClose($server, $fd){echo "敞开 TCP 连贯".PHP_EOL;}

    /**
     * 发动 Get 或 Post 申请
     * @param string $url           申请地址
     * @param array $request_data   申请参数
     * @param string $request_type  申请类型
     * @param array $headers        头信息
     * @param bool $is_ssl          是否是 ssl
     * @return bool|string
     */
    public function curl($url = '', $request_data = [], $request_type ='get', $headers = [], $is_ssl = false)
    {$curl = curl_init (); // 初始化
        // 设置 URL
        curl_setopt($curl, CURLOPT_URL, $url);
        // 不返回 Response 头部信息
        curl_setopt ($curl, CURLOPT_HEADER, 0);
        // 如果胜利只将后果返回，不主动输入任何内容
        curl_setopt ($curl, CURLOPT_RETURNTRANSFER, 1);
        // 设置申请参数
        curl_setopt ($curl, CURLOPT_POSTFIELDS, http_build_query($request_data));
        // TRUE 时追踪句柄的申请字符串
        curl_setopt($curl, CURLINFO_HEADER_OUT, true);
        // Post 类型减少以下解决
        if($request_type == 'post') {
            // 设置为 POST 形式
            curl_setopt ($curl, CURLOPT_POST, 1);
            // 设置头信息
            curl_setopt($curl, CURLOPT_HTTPHEADER, array('Content-Type: application/json', 'Content-Length:' . strlen(json_encode($request_data))));
            // 设置申请参数
            curl_setopt ($curl, CURLOPT_POSTFIELDS, json_encode($request_data));
            // 当 POST 数据大于 1024 时强制执行
            curl_setopt ($curl, CURLOPT_HTTPHEADER, array("Expect:"));
        }

        // 判断是否绕过证书
        if($is_ssl) {
            // 绕过 ssl 验证
            curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, false);
            curl_setopt($curl, CURLOPT_SSL_VERIFYHOST, false);
        }
        if(!empty($headers))  curl_setopt($curl, CURLOPT_HTTPHEADER, $headers);
        // 执行
        $result = curl_exec ($curl);
        if ($result == FALSE) return false;
        // 敞开资源
        curl_close ($curl);
        return $result;
    }

    /**
     * 写入日志
     * @param $content
     */
    public function addLog($content){$path = dirname(__FILE__)."/logs/";
        if (!is_dir($path))
            mkdir($path,0777,true);

        $file_name = $path.date("Y_m_d") . ".log";
        if (!file_exists($file_name)) {touch($file_name);
            chown($file_name, 0777);
        }

        $file_log = fopen($file_name, "a");
        fputs($file_log, $content);
        fclose($file_log);
    }
}

$server = new taskServer();

这里的客户端能够是 cli 脚本，也能够是对应控制器中的具体方法，只有能连贯 Swoole 监听的 Tcp 就行。

<?php
namespace app\admin\controller;

class Index extends Base
{public function index(){$client = new \Swoole\Client(SWOOLE_SOCK_TCP);
          if (!$client->connect('0.0.0.0', 9501)) {return json("connect failed. Error: {$client->errCode}\n");
          }
          $data = [
              "url" => "https://api.paasoo.com/json",
              "data" => [
                  "key" => "key",
                  "secret" => "secret",
                  "from" => "sms",
                  "to" => "mobile_phone",
                  "text" => "test",
              ],
              "type" => "get"
          ];
        $client->send(json_encode($data));
        return json($client->recv());
    }
}