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');
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 . '/refill/RefillFactory.php');
require_once(BASE_HELPER_PATH . '/message/msgstates.php');
require_once(BASE_HELPER_PATH . '/message/msgutil.php');
require_once(BASE_HELPER_PATH . '/message/subscriber.php');

require_once(BASE_PATH . '/processor.php');
require_once(BASE_PATH . '/proxy.php');
require_once(BASE_PATH . '/yzpolicy.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 work_proc()
{
    Base::run_util();

    $trdid = Thread::getCurrentThreadId();
    Log::record("thread id = {$trdid}",Log::INFO);
//
//    $gMessageStates = new MsgStates();
//    StatesHelper::init();
//    $listener = new message\subscriber($gMessageStates);
//    $listener->start();

    $looper = new processor();
    $looper->run();
}

function corun()
{
    Log::record("start run process",Log::DEBUG);
    Co\run(function() {
        work_proc();
    });

//    $process = new swoole_process(function(swoole_process $process)
//    {
//        try {
//            Log::record("start run process",Log::DEBUG);
//            Co\run(function() {
//                work_proc();
//            });
//        } catch (Exception $ex) {
//            Log::record("",Log::ERR);
//        }
//    });
//
//    $pid = $process->start();
//    return $pid;
}

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