xyzshop/rdispatcher/codispatcher.php

<?php
declare(strict_types=0);

define('APP_ID', 'cordispatcher');
define('MOBILE_SERVER',true);
define('USE_COROUTINE',true);
define('SUPPORT_PTHREAD',false);


define('BASE_ROOT_PATH',str_replace('/rdispatcher','',dirname(__FILE__)));
define('BASE_PATH',BASE_ROOT_PATH . '/rdispatcher');

require_once(BASE_ROOT_PATH . '/global.php');
require_once(BASE_ROOT_PATH . '/fooder.php');

require_once(BASE_HELPER_PATH . '/event_looper.php');
require_once(BASE_HELPER_PATH . '/queue/rdispatcher.php');
require_once(BASE_HELPER_PATH . '/algorithm.php');
require_once(BASE_HELPER_PATH . '/refill/RefillFactory.php');
require_once(BASE_PATH . '/processor.php');
require_once(BASE_PATH . '/proxy.php');

Co::set(['hook_flags' => SWOOLE_HOOK_NATIVE_CURL]);
if (empty($_SERVER['argv'][1])) exit('parameter error');
$process_count = intval($_SERVER['argv'][1]);

function all_channels() {
    return ['refill'];
}

function handle_error($level, $message, $file, $line)
{
    if($level == E_NOTICE) return;
    $trace = "handle_error: level={$level},msg={$message} file={$file},line={$line}\n";
    $backtrace = debug_backtrace();
    foreach ($backtrace as $item) {
        $trace .= "{$item['file']}\t{$item['line']}\t{$item['function']}\n";
    }

    Log::record($trace,Log::ERR);
}

function sub_message(&$waiting_quit,&$sub_redis,$channels)
{
    $sub_redis = new Swoole\Coroutine\Redis();
    while (!$waiting_quit)
    {
        try
        {
            if(!$sub_redis->connected) {
                $ret = $sub_redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
                Log::record("sub_message redis connected = {$sub_redis->connected}",Log::DEBUG);
            }
            else {
                $ret = true;
            }

            if(!$ret) {
                Log::record("sub_message cannot connet redis.",Log::DEBUG);
                $sub_redis->close();
                Swoole\Coroutine::sleep(1);
            }
            elseif($sub_redis->subscribe($channels))
            {
                while ($msg = $sub_redis->recv())
                {
                    [$sub_type, $channel, $content] = $msg;
                    $content = unserialize($content);
                    $type = $content['type'];

                    if($channel != 'refill') continue;
                    if($type == 'channel' || $type == 'merchant') {
                        Log::record("sub_message recv mgs:{$sub_type}-{$channel}-{$type}",Log::DEBUG);
                        refill\RefillFactory::instance()->load();
                    }
                }
                Log::record("sub_message redis recv timeout",Log::DEBUG);
            }
            else {
                Log::record("sub_message subscribe error",Log::ERR);
            }
        }
        catch (Exception $ex)
        {
            Log::record($ex->getMessage(),Log::ERR);
        }
    }

    $waiting_quit = true;
    Log::record("sub_message quit",Log::DEBUG);
}

function ratio_update(&$waiting_quit)
{
    //每分钟的第0秒 将上一分钟的数据，写入HD5File。
    //每分钟的第2秒 统计之前的数据
    //每分钟的第3秒触发更新 成功率统计数据
    while (!$waiting_quit)
    {
        for ($i = 0; $i < 61; $i++) {
            $cur_time = time();
            $time_sec = $cur_time;
            $next_min = $time_sec - $time_sec % 60;

            if(($next_min > $cur_min && $time_sec % 60 == 3) || $waiting_quit) {
                $cur_min = $next_min;
                break;
            } else {
                Swoole\Coroutine::sleep(0.1);
            }
        }
        if(!$waiting_quit) {
            refill\RefillFactory::instance()->UpdateRatio();
        }
    }

    $waiting_quit = true;
    Log::record("ratio_update quit",Log::DEBUG);
}

$workers = [];
for ($i = 0; $i < $process_count;$i++)
{
    $process = new Swoole\Process(function(Swoole\Process $worker)
    {
        Base::run_util();
        set_error_handler('handle_error');
        $waiting_quit = false;
        $sub_redis = null;
        go(function () use (&$waiting_quit,&$sub_redis) {
            sub_message($waiting_quit,$sub_redis,['refill']);
        });

        $looper = new processor(true);
        Swoole\Process::signal(SIGTERM, function($signal_num) use ($worker,&$waiting_quit,$sub_redis,$looper)
        {
            Log::record("signal call SIGTERM begin = $signal_num, #{$worker->pid}",Log::DEBUG);

            set_error_handler(null);
            try {
                $waiting_quit = true;
                $sub_redis->close();
            }
            catch(Exception $ex) {
                Log::record($ex->getMessage(),Log::DEBUG);
            }

            $looper->stop();
            do {
                $res = Swoole\Coroutine::stats();
                $num = $res['coroutine_num'];
                if($num > 1) {
                    Swoole\Coroutine::sleep(0.1);
                }
                Log::record("coroutine_num = {$num}",Log::DEBUG);
            } while($num > 1);
        });

        $looper->run();

    }, false, false, true);

    $pid = $process->start();
    $workers[$pid] = $process;
}


Log::record("main process start wait sub process....",Log::DEBUG);
while (true)
{
    if($status = Swoole\Process::wait(true)) {
        Log::record("Sub process #{$status['pid']} quit, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
    }
    else
    {
        foreach ($workers as $pid => $worker)
        {
            Swoole\Process::kill($pid,SIGTERM);
            if($status = Swoole\Process::wait(true)) {
                Log::record("Graceful Recycled #{$status['pid']}, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
            }
        }
        break;
    }
}
Log::record("Quit all",Log::DEBUG);