php異步處理任務 使用redis消息隊列

安裝redis 和 php-redis擴展，參考網友博文

linux環境   https://blog.csdn.net/cailongbiaoyuli/article/details/83445906

windows環境  https://blog.csdn.net/q810391679/article/details/80812578

 

通過生產消費模式實現異步處理 ，把待運行任務添加到消息隊列， 定時從隊列取出消息處理任務；

定時處理任務的方法，

linux環境下，添加到corntab，每1分鍾運行一次

# crontab -e

*/1 * * * * /usr/local/bin/php /home/www/app.php?do=worker

windows環境下，添加到計划任務，計划任務運行bat文件，bat中調用php執行腳本

tmp.bat

D:\php56\php.exe -q E:\www\201906\app.php?do=worker

 

 

<?php
// 發布消息
// app.php?do=rpush&act=sendsms&mob=15530000000&msg=tmp_order_new
// 定時任務 消費消息
// app.php?do=worker
$conf = ['redis_host'=>'127.0.0.1', 'redis_port'=>6379, 'redis_auth'=>'6min']
$app = new appStd($conf);
$app->run();

class appStd
{
    private static $redis;
    
    public function __construct($conf){
        if(! (static::$redis instanceof Redis)) {
            self::$redis = new Redis();
            self::$redis->connect($conf['redis_host'], $conf['redis_port']);
            if(isset($conf['reids_authpass'])){
                self::$redis->auth($conf['redis_auth']);
            }
        }
    }

    public function worker($param){
        //ignore_user_abort(true);
        set_time_limit(30);
        while (true) {
            try{
                $cmd = self::$redis->get('cmd');
                if('stop' == $cmd) break;
                //取出一條消息
                $msg = self::$redis->lPop('msglist');
                if(!$msg) break;
                $wk = unserialize($msg);
                switch($wk['act']){
                    case 'sendsms':
                        $do = SMS::send($wk['data']);
                        break;
                    case 'report':
                        $do = YUN::report($wk['data']);
                        break;
                }
                //失敗任務放回隊列
                if(!$do){
                    self::$redis->rPush('msglist', $msg);
                }
            }catch(Exception $e){
                die($e->getMessage());
            }
        }
        die('success');
    }

    public function rpush($param){
        $msg = ['act'=>$param['act'], 'data'=>[]];
        self::$redis->rPush('msglist', serialize($msg));
        die('success');
    }
    
    public function run(){
        $param  = &$_REQUEST;
        $action = isset($param['do']) ? $param['do'] : '';
        if( !method_exists($this, $action)){
            $this->fail('disabled action');
        }
        $this->$action($param);
    }
}