xyzshop/rdispatcher/dispatcher.php

<?php
declare(strict_types=0);

define('APP_ID', 'rdispatcher');
define('MOBILE_SERVER',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');

Log::record("1",Log::DEBUG);
require_once(BASE_HELPER_PATH . '/event_looper.php');
require_once(BASE_HELPER_PATH . '/queue/rdispatcher.php');
Log::record("2",Log::DEBUG);
Log::record(BASE_HELPER_RAPI_PATH,Log::DEBUG);

require_once(BASE_HELPER_PATH . '/refill/RefillFactory.php');
require_once(BASE_HELPER_PATH . '/algorithm.php');
Log::record("3",Log::DEBUG);

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

//这个可以
//Co::set(['hook_flags' => SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_TCP ^ SWOOLE_HOOK_UDP ^
//    SWOOLE_HOOK_FILE ^ SWOOLE_HOOK_UNIX ^
//    SWOOLE_HOOK_STREAM_FUNCTION ^
//    SWOOLE_HOOK_BLOCKING_FUNCTION ^
//    SWOOLE_HOOK_PROC ^
//    SWOOLE_HOOK_SLEEP ^
//    SWOOLE_HOOK_TLS ^
//    SWOOLE_HOOK_SSL
//]);

//这样会导致，curl 协程部分不执行，收不到数据
//Co::set(['hook_flags' => SWOOLE_HOOK_NATIVE_CURL | SWOOLE_HOOK_SOCKETS]);

//这样会导致，redis任何阻塞函数不返回
//Co::set(['hook_flags' => SWOOLE_HOOK_ALL]);

//目前没发现问题
Co::set(['hook_flags' => SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_TCP]);

if (empty($_SERVER['argv'][1])) exit('parameter error');
$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 work_proc()
{
    Base::run_util();
    set_error_handler('handle_error');
    $looper = new processor();
    $looper->run();
}

//swoole 和 pthreads 冲突
function corun()
{
    Log::record("start run process",Log::DEBUG);
    Co\run(function() {
        work_proc();
    });
}

event\util::fork_workerex('corun',$count);