PHP-Swoole-实现异步任务队列

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

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

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

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

联合官网手册和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());
    }
}

参考链接

• php应用Swoole来实现实时异步工作队列
• 基于 swoole 下 异步音讯队列 API

喜欢 (0)赏

[搞代码]

分享 (0)