qimall/console/controllers/SwooleCrontabController.php

<?php
/**
 * rabbitMq 消费者
 */

namespace console\controllers;

use common\models\common\CrontabTask;
use common\models\common\CrontabTaskExecute;
use console\controllers\swoole_server\BaseSwooleServer;
use Co;
use Swoole\Process;
use Swoole\Runtime;
use Swoole\Server;
use Swoole\Timer;
use Swoole\Coroutine\Client;
use function Swoole\Coroutine\run;
use Yii;
use yii\console\ExitCode;
use common\helpers\DateHelper;
use linslin\yii2\curl\Curl;

/**
 * 基于swoole定时任务封装的，定时任务处理服务
 * @Author bing
 * @DateTime 2021-10-15 10:20:51
 * @copyright: Copyright (c) 2020 广东七件事集团
 */
class SwooleCrontabController extends BaseSwooleServer
{
    public $host;
    public $port;
    public $sign;
    private $serv;
    public $timeout;
    private $client;

    /**
     * console模式初始化
     * @Author bing
     * @DateTime 2021-10-18 13:16:01
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @return void
     */
    public function init()
    {
        parent::init();
    }

    /**
     * 开启定时任务服务
     * @Author bing
     * @DateTime 2021-10-21 10:28:30
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @return void
     */
    public function actionServerRun(){
        $this->beforeRun();
        // 重启服务
        if ($this->option == 'stop') return ExitCode::OK;
        $this->serv = new Server('0.0.0.0',9501,SWOOLE_PROCESS,SWOOLE_TCP);
        $this->serv->set(array(
            'worker_num' => 2,
            'daemonize' => false,
            'task_worker_num' => 2,
            'open_eof_split' => true,
            'package_eof' => "\r\n",
            'log_file' =>Yii::$app->getRuntimePath() .'/logs/swoole_crontab_server.log',
            // 'ssl_verify_peer' => true,
            // 'ssl_allow_self_signed' => true,
            // 'ssl_cert_file' => __DIR__ . '/task/ca/server/server.crt',
            // 'ssl_key_file' => __DIR__ . '/task/ca/server/server.key',
            // 'ssl_client_cert_file' => __DIR__ . '/task/ca/private/ca.crt',
            // 'ssl_verify_depth' => 10,
        ));
        $this->serv->on('Start', [$this, 'onStart']);
        $this->serv->on('Connect', [$this, 'onConnect']);
        $this->serv->on('Receive', [$this, 'onReceive']);
        $this->serv->on('Close', [$this, 'onClose']);
        $this->serv->on('Task', [$this, 'onTask']);
        $this->serv->on('Finish', [$this, 'onFinish']);
        $this->serv->on('workerStart', [$this, 'onWorkerStart']);
        $this->serv->on('pipeMessage', [$this, 'onPipeMessage']);
        $serv = $this->serv;

        // 准备任务数据
        $crontab_prepare_process = new Process(function ($process) use ($serv) {
            swoole_set_process_name("swoole : crontab_prepare_process {$process->pid}");
            run(function () use ($serv) {
                CrontabTask::prepareData();
                Timer::tick(60000, function () use ($serv) {
                    //CrontabTask
                    CrontabTask::prepareData();
                });
            });
        });
        $this->serv->addProcess($crontab_prepare_process);
        // 任务分发进程500ms每次
        $crontab_dispatch_process = new Process(function ($process) use ($serv) {
            swoole_set_process_name("swoole : crontab_dispatch_process {$process->pid}");
            run(function () use ($serv) {
                Timer::tick(500, function () use ($serv) {
                    $client_list = [];
                    $client_list_info = [];
                    $start_fd = 0;
                    $list = $serv->getClientList($start_fd, 100);
                    if($list){
                        foreach ($list as $fd) {
                            array_push($client_list, $fd);
                            array_push($client_list_info, $serv->getClientInfo($fd));
                        }
                        $start_fd = end($list);
                        $count = count($client_list);
                        if ($count > 0) {
                            while ($task = CrontabTaskExecute::pop()) {
                                $remaining = $task['id'] % $count;
                                $this->addSign($task);
                                $serv->send($client_list[$remaining], json_encode($task) . "\r\n");
                            }
                        }
                    }
                });
            });
        });
        $this->serv->addProcess($crontab_dispatch_process);

        // 任务失败发送警告60s执行一次
        $crontab_prepare_process = new Process(function ($process) use ($serv) {
            swoole_set_process_name("swoole : crontab_warning_process {$process->pid}");
            run(function () use ($serv) {
                Timer::tick(60000, function () use ($serv) {
                    CrontabTaskExecute::warningHandle();
                });
            });
        });
        $this->serv->addProcess($crontab_prepare_process);
        $this->serv->start();
    }

    /**
     * 使用父类的服务启动进行进程管理
     * @Author bing
     * @DateTime 2021-10-18 13:04:14
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @param [type] $serv
     * @return void
     */
    public function onStart($serv)
    {
        parent::onStart($serv);
    }

    /**
     * 客户端连接回调
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @param $serv Server Server对象
     * @param $fd int 是TCP客户端连接的标识符
     * @param $from_id int 投递任务的worker_id
     * @return void
     */
    public function onConnect($serv, $fd, $from_id)
    {
        echo "have connect!\n";
    }

  
    /**
     * 接收到客户端消息
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @param $serv Server Server对象
     * @param $fd int 是TCP客户端连接的标识符
     * @param $from_id int 投递任务的worker_id
     * @param $data string 接收到的数据
     *  * @return void
     */
    public function onReceive(Server $serv, $fd, $from_id, $data)
    {
        echo "Get Message From Client {$fd}:{$data}\n";
    }

    /**
     * 客户端关闭时
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @param $serv Server Server对象
     * @param $fd int 是TCP客户端连接的标识符
     * @param $from_id int 投递任务的worker_id
     */
    public function onClose($serv, $fd, $from_id)
    {
        echo "Client {$fd} close connection\n";
    }

    /**
     * 异步任务处理
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @param $serv Server Server对象
     * @param $task_id int 任务id
     * @param $from_id int 投递任务的worker_id
     * @param $data string 投递的数据
     */
    public function onTask(Server $serv, $task_id, $from_id, $data)
    {
        echo "onTask\n";
        $serv->finish($data);
    }

    /**
     * 异步任务执行完成回调
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @param $serv Server Server对象
     * @param $task_id int 任务id
     * @param $data string 任务返回的数据
     */
    public function onFinish(Server $serv, $task_id, $data)
    {
        echo "onFinish\n";
    }

    /**
     * worker进程启动时回调
     * @Author bing
     * @DateTime 2021-10-18 13:05:03
     * @param $serv Server Server对象
     * @param $worker_id
     */
    public function onWorkerStart($serv, $worker_id)
    {
        echo  "onWorkerStart\n";
        if ($worker_id >= $serv->setting['worker_num']) {
            swoole_set_process_name("swoole : new_task_worker");
        } else {
            swoole_set_process_name("swoole : worker");
        }
    }
    
    /**
     * 客户端消费定时任务
     * @Author bing
     * @DateTime 2021-10-21 10:29:46
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @return void
     */
    public function actionClientRun()
    {
        run(function () {
            $this->client = new Client(SWOOLE_SOCK_TCP);
            $this->client->set([
                'open_eof_split' => true,
                'package_eof' => "\r\n",
            ]);
            $is_docker_deploy = \Yii::$app->params['is_docker_deploy'] ?? false;
            $server = $is_docker_deploy ? 'crontab-server' : '127.0.0.1';

            if (!$this->client->connect($server, 9501, 0.5)){
                echo "connect failed. Error: {$this->client->errCode}".PHP_EOL;
            }
            while (true) {
                $data = $this->client->recv();
                if (strlen($data) > 0) {
                    $this->ExecuteCrontab($data);
                } else {
                    if ($data === '') {
                        // 全等于空 直接关闭连接
                        $this->client->close();
                        break;
                    } else {
                        if ($data === false) {
                            // 可以自行根据业务逻辑和错误码进行处理，例如：
                            // 如果超时时则不关闭连接，其他情况直接关闭连接
                            if ($this->client->errCode !== SOCKET_ETIMEDOUT) {
                                $this->client->close();
                                break;
                            }
                        } else {
                            $this->client->close();
                            break;
                        }
                    }
                }
                Co::sleep(0.3);
            }
        });
    }

    /**
     * 执行定时任务
     * @Author bing
     * @DateTime 2021-10-12 15:50:13
     * @copyright: Copyright (c) 2020 广东七件事集团
     * @param [type] $task
     * @return void
     */
    public function ExecuteCrontab($task){
        if ($this->checkSign($task)) {
            $task = json_decode($task,true);
            if (empty($task['execute_time']) || empty($task['command']) || empty($task['id']) || empty($task['type'])) return false;
            // 计算需要执行的时间
            $after_time_ms = $task['execute_time'] - time();
            if ($after_time_ms <= 0) $after_time_ms = 0.001;
            Timer::after($after_time_ms * 1000, function () use ($task) {
                // 开启一个协程处理任务
                go(function() use ($task){
                    $res = CrontabTaskExecute::updateProcess($task['id'], CrontabTaskExecute::PROCESS_PENDING, time());
                    $running_time = DateHelper::getMicrotime();
                    $date = date('Y-m-d H:i:s');
                    switch ($task['type']) {
                        case  CrontabTask::TYPE_SHELL:
                            $tmpFile = tempnam("/tmp/", "swoole_corntab_task_");
                            file_put_contents($tmpFile, $task['command']);
                            $command = "bash $tmpFile";
                            echo '['.$date.'] start exec '. $command . PHP_EOL;
                            exec($command,$output,$return_var);
                            $flag = $return_var === 0 ? true : false;
                            unlink($tmpFile);
                        break;
                        case  CrontabTask::TYPE_YII:
                            $yii_path = dirname(Yii::$app->basePath);
                            $command = $yii_path.'/yii '.$task['command'];
                            echo '['.$date.'] start exec '. $command . PHP_EOL;
                            exec($command,$output,$return_var);
                            $flag = $return_var === 0 ? true : false;
                        break;
                        case CrontabTask::TYPE_URL:
                            Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
                            Runtime::enableCoroutine(SWOOLE_HOOK_CURL);
                            echo '['.$date.'] start exec '.$task['command'].PHP_EOL;
                            $curl = new Curl();
                            $response = $curl->get($task['command']);
                            if ($curl->errorCode === null) {
                                $flag = true;
                                $output = $response;
                             } else {
                                $flag = false;
                                // 错误码参考：https://curl.haxx.se/libcurl/c/libcurl-errors.html
                                // 错误描述参考：http://php.net/manual/en/function.curl-strerror.php
                                $output = 'curl请求错误，错误码：'.$curl->errorCode.'错误描述：'.$curl->errorText;
                            }
                        break;
                    }
                    if($flag === true){
                        $running_time -= DateHelper::getMicrotime();
                        CrontabTaskExecute::updateProcess($task['id'],  CrontabTaskExecute::PROCESS_FINISH, null,json_encode($output, JSON_UNESCAPED_UNICODE), time(),abs($running_time/1000));
                    }else{
                        CrontabTaskExecute::updateProcess($task['id'], CrontabTaskExecute::PROCESS_FAIL,null,json_encode($output, JSON_UNESCAPED_UNICODE));
                    }
                });
            });
        }
    }



}