REAPI/app/task/controller/OrderTask.php

<?php

namespace app\task\controller;
set_time_limit(0);
ini_set('memory_limit','512M'); 
ini_set('default_socket_timeout', -1); //不超时
use app\channel\service\ChannelService;
use app\gateway\controller\api\Notify;
use app\gateway\service\RedisService;
use app\merchant\service\OrderService;
use app\openapi\controller\Core;
use dever\Log;


/**
 * 任务
 * Class Test
 * @package app\gateway\api
 */
class OrderTask extends Core
{
	# 是否检测数据
    protected $check = false;
    # redis连接
    protected $redis = false;

    public function cron()
    {
        while(1) {
            $this->query();
        }
    }

	# 对submit提交的数据进行处理
	public function query()
    {
//        var_dump(1);die;
        try {
            $settingUpdate = sysconf('settingUpdate');
            if ($settingUpdate == 1) {
                return true;
            }
            # 进程模式 1多进程 2单进程
            $queue = sysconf('settingRedisQueryOrder');


//            if ($queue == 1) {
//                # 获取当前执行的进程数量
//                $num = $this->getNum();
//                if ($num >= 500) {
//                    # 等会儿再执行
//                    sleep(30);
//                    return true;
//                }
//                $popen = true;
//            } else {
//                return true;
//            }

            $this->redis();
            $data = $this->redis->pop('query');

            $this->run($data, true);
            return true;
        } catch (\Exception $e) {
            $this->redis = false;
            return true;
        }
    }

    protected function redis()
    {
        if (!$this->redis) {
            $this->redis = RedisService::getInstance();
        }
    }

    protected function run($data, $popen = false)
    {

        if (isset($data['order_id']) && $data['order_id']) {
            if ($popen) {
//                $this->runOne


                $this->popen($data['order_id']);
            } else {
//                var_dump($data);die;
                $channelService = ChannelService::instance();
                $result = $channelService->call('query', $data['cid'], $data);
                Log::write('test', 'runss', $result);
//                Log::write('');
//                var_dump($result);die;
                if ($result['status'] != 4) {
                    Log::write('test', 'runs', $result);
                    $notifyClass = new Notify($this->app);
                    $notifyClass->queue_query($data['order_id'], $result);
                }
            }
        }
    }

    # 将数据推到子进程处理
    public function popen($order)
    {
        $command = 'php /www/wwwroot/reapi/public/index.php openapi/orderTask/runOne ' . $order . ' 1>/dev/null 2>&1 &';
//        var_dump($command);die;

        exec($command);
    }

    # 获取当前执行的子进程数量
    public function getNum()
    {
        $command = 'ps -ef | grep openapi/orderTask/runOne | grep -v grep | wc -l';
        $num = exec($command);
        return $num;
    }

    # 对一条数据进行处理
    public function runOne()
    {
//        var_dump(123);die;
        if (isset($_SERVER["argv"][2]) && $_SERVER["argv"][2]) {
            $order = $_SERVER["argv"][2];
        } else {
            $order = input('order');
        }

        Log::write('test', 'run', $order);

        if (!$order) {
            exit('error');
        }

        $nq = false;
//        var_dump($order);die;
        if (str_contains($order, '_nq')) {
            $order = str_replace('_nq', '', $order);
            $nq = true;
        }

        $orderService = OrderService::instance();
        $info = $orderService->get($order);
        if ($info && $info['status'] == 4 ) {
            if(isset($vo['num']) && $vo['num']>0) {
                $info['order_id'] = $vo['order_id'].'_'.$vo['num'];
            }
            if ($nq) {
                $info['request']['param']['s'] = 1;
            }
//            var_dump($info);die;
            $this->run($info);
        }
        exit('ok');
    }

    # 商户对账


    # 直接对数据库中的订单数据进行处理 如果队列失效，就使用这个方法进行操作，谨慎使用该方法，会和队列同时执行，已取消
    /*
    public function runAll()
    {
        $orderService = OrderService::instance();
        $data = $orderService->getData();

        if ($data) {
            foreach ($data as $k => $v) {
                if ($v['request']) {
                    $v['request'] = json_decode($v['request'], true);
                    $this->run($v['request']);
                }
            }
        }
    }
    */






}