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|SWOOLE_HOOK_SLEEP]);
Co::set(['socket_connect_timeout' => 60,'socket_read_timeout' => 900,'socket_write_timeout' => 900,'socket_timeout' => 900]);

if (empty($_SERVER['argv'][1])) exit('parameter error');
$process_count = intval($_SERVER['argv'][1]);

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

function strbool($value) {
    return $value ? 'true' : 'false';
}

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 subscribe_message(&$quit, &$redis, $channels)
{
    $redis = new Swoole\Coroutine\Redis();
    while (!$quit)
    {
        try
        {
            Log::record("subscribe_message start quit=" . strbool($quit),Log::DEBUG);
            if(!$redis->connected) {
                $ret = $redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
            }
            else {
                $ret = true;
            }

            if(!$ret) {
                Log::record("subscribe_message cannot connet redis.",Log::DEBUG);
                $redis->close();
            }
            elseif($redis->subscribe($channels))
            {
                while ($msg = $redis->recv())
                {
                    [$sub_type, $channel, $content] = $msg;
                    $content = json_decode($content,true);
                    $type = $content['type'];

                    if($channel != 'refill') continue;
                    if($quit) break;

                    if($type == 'channel' || $type == 'merchant') {
                        refill\RefillFactory::instance()->load();
                        refill\transfer::instance()->load();
                    }
                    elseif($type == 'ratio') {
                        $ins = Cache::getInstance('cacheredis');
                        $val = $ins->get_org('channel_ratios');

                        if(empty($val)) continue;
                        $val = json_decode($val,true);
                        if(empty($val)) continue;
                        $ratios = $val['ratios'];
                        if(empty($ratios)) continue;

                        refill\RefillFactory::instance()->UpdateRatio($ratios);
                    }
                    elseif($type == 'mch_counts') {
                        $ins = Cache::getInstance('cacheredis');
                        $content = $ins->get_org('merchant_refill_counts');

                        if(empty($content)) continue;
                        $counts = json_decode($content,true);
                        if(empty($counts)) continue;

                        $gross = $counts['gross'];
                        $detail = $counts['detail'];

                        refill\RefillFactory::instance()->UpdateMchRatios($gross,$detail);
                    }
                    elseif($type == 'channels_speed') {
                        $ins = Cache::getInstance('cacheredis');
                        $content = $ins->get_org('channels_speed');

                        if(empty($content)) continue;
                        $speeds = json_decode($content,true);
                        if(empty($speeds)) continue;

                        refill\RefillFactory::instance()->UpdateSpeeds($speeds);
                    }
                    else {
                        //Log::record("subscribe_message dont not handle mgs:{$sub_type}-{$channel}-{$type}",Log::DEBUG);
                    }
                }
            }
            else {
                Log::record("subscribe_message subscribe error",Log::ERR);
            }
        }
        catch (Exception $ex)
        {
            Log::record($ex->getMessage(),Log::ERR);
        }
    }

    Log::record("subscribe_message quit =" . strbool($quit),Log::DEBUG);
    $quit = true;
}

$workers = [];
for ($i = 0; $i < $process_count;$i++)
{
    $process = new Swoole\Process(function(Swoole\Process $worker)
    {
        Base::run_util();
        refill\RefillFactory::instance();
        refill\transfer::instance();

        $sub_quit = false;
        $sub_redis = null;
        $looper = new processor(false);

        set_error_handler('handle_error');
        register_shutdown_function(function () use ($looper)
        {
            $error = error_get_last();
            if(!empty($error) && $error['type'] != E_NOTICE) {
                $msg = "register_shutdown_function type:{$error['type']}\n message:{$error['message']}\n file={$error['file']} line={$error['line']}";
                Log::record("$msg",Log::ERR);
            }
        });

        go(function () use (&$sub_quit,&$sub_redis) {
            subscribe_message($sub_quit,$sub_redis,['refill']);
        });

        go(function () use ($looper) {
            $looper->run();
        });

        Swoole\Process::signal(SIGTERM, function($signal_num) use ($worker,&$sub_quit,$sub_redis,$looper)
        {
            Log::record("signal call SIGTERM begin signum={$signal_num}, pid={$worker->pid}",Log::DEBUG);
            set_error_handler(null);
            try {
                $sub_quit = true;
                $sub_redis->close();
            }
            catch(Exception $ex) {
                Log::record($ex->getMessage(),Log::DEBUG);
            }

            $looper->stop();
        });
    }, 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))
    {
        $quit_pid = $status['pid'];
        Log::record("Sub process pid={$status['pid']} quit, code={$status['code']}, signum={$status['signal']}",Log::DEBUG);
        foreach ($workers as $pid => $worker)
        {
            if($pid != $quit_pid) {
                Swoole\Process::kill($pid, SIGTERM);
            }
        }

        foreach ($workers as $pid => $worker)
        {
            if($pid != $quit_pid && ($status = Swoole\Process::wait(true))) {
                Log::record("Graceful Recycled pid={$status['pid']}, code={$status['code']}, signum={$status['signal']}",Log::DEBUG);
            }
        }
        break;
    }
    else
    {
        foreach ($workers as $pid => $worker) {
            Swoole\Process::kill($pid, SIGTERM);
        }

        foreach ($workers as $pid => $worker)
        {
            if($status = Swoole\Process::wait(true)) {
                Log::record("Graceful Recycled pid={$status['pid']}, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
            }
        }
        break;
    }
}
Log::record("Quit all",Log::DEBUG);