xyzshop/rdispatcher/coall.php

<?php
declare(strict_types=0);

define('APP_ID', 'coall');
define('MOBILE_SERVER',true);
define('USE_COROUTINE',true);
define('SUPPORT_PTHREAD',false);
define('COROUTINE_HOOK_TCP',true);
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 . '/algorithm.php');
require_once(BASE_HELPER_PATH . '/event_looper.php');
require_once(BASE_HELPER_PATH . '/queue/rdispatcher.php');
require_once(BASE_HELPER_PATH . '/refill/RefillFactory.php');
require_once(BASE_CORE_PATH . '/framework/libraries/CoRefPool.php');
require_once(BASE_CORE_PATH . '/framework/libraries/CoMysqliPool.php');
require_once(BASE_CORE_PATH . '/framework/libraries/CoPool.php');
require_once(BASE_CORE_PATH . '/framework/libraries/CoRedisPool.php');

require_once(BASE_PATH . '/processor.php');
require_once(BASE_PATH . '/proxy.php');

Co::set(['hook_flags' => SWOOLE_HOOK_ALL | SWOOLE_HOOK_NATIVE_CURL]);
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,$looper)
{
    $redis = new Redis();
    while (!$quit)
    {
        try
        {
            Log::record("subscribe_message start quit=" . strbool($quit),Log::DEBUG);
            if(!$redis->isConnected()) {
                $redis->close();
                $ret = $redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
                $redis->setOption(Redis::OPT_READ_TIMEOUT, 3600);
            }
            else {
                $ret = true;
            }

            if(!$ret) {
                Log::record("subscribe_message cannot connet redis.",Log::DEBUG);
                sleep(1);
                continue;
            }

            $fLoading = false;
            $ret = $redis->subscribe($channels,function ($redis,$channel,$msg) use(&$quit,$looper,&$fLoading)
            {
                Log::record("channel={$channel} msg={$msg}",Log::DEBUG);
                $content = json_decode($msg,true);
                $type = $content['type'];

                if($channel != 'refill') return;
                if($quit) return;

                if($type == 'channel' || $type == 'merchant') {
                    $fLoading = true;
                    $looper->pause();
                    Log::record("start load.....",Log::DEBUG);
                    refill\RefillFactory::instance()->load();
                    Log::record("fini load.....",Log::DEBUG);
                    $looper->resume();
                    $fLoading = false;
                }
                elseif($type == 'ratio')
                {
                    if($fLoading == false)
                    {
                        $ins = Cache::getInstance('cacheredis');
                        $val = $ins->get_org('channel_ratios');

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

                        refill\RefillFactory::instance()->UpdateRatio($ratios);
                    }
                }
                else {
                    Log::record("subscribe_message dont not handle mgs:{$channel}-{$type}",Log::DEBUG);
                }
            });
            Log::record("subscribe ret={$ret}",Log::DEBUG);
        }
        catch (Exception $ex)
        {
            Log::record("subscribe_message " . $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();
        set_error_handler('handle_error');

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

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

        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 = $signal_num, #{$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();
            CoRedisPool::instance()->stop();
            CoMysqliPool::instance()->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 #{$status['pid']} quit, code={$status['code']}, signal={$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 #{$status['pid']}, code={$status['code']}, signal={$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 #{$status['pid']}, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
            }
        }
        break;
    }
}
Log::record("Quit all",Log::DEBUG);