xyzshop/rdispatcher/codispatcher.php

<?php
declare(strict_types=0);

define('APP_ID', 'rdispatcher');
define('MOBILE_SERVER',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 . '/refill/CoRefillFactory.php');
require_once(BASE_HELPER_PATH . '/algorithm.php');

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();

    go(function () {
        $redis = new Swoole\Coroutine\Redis;
        $ret = $redis->connect('host.docker.internal', 6379);
        $key = 'test_cache_data';
        $ret = wkcache($key, ["hello world"]);
        $data = rkcache($key);
    });

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

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

//swoole 和 pthreads 冲突
function corun()
{
    Co::set(['hook_flags' => SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_TCP]);
    Co\run(function() {
        Log::record("start run process",Log::DEBUG);
        work_proc();
    });
}
//corun();
work_proc();

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