Swoole 및 Redis 기반 동시 대기열 처리 시스템

Question

PHP는 멀티스레딩을 지원하지 않기 때문에 완전한 시스템으로서 비동기적으로 완료해야 하는 작업이 많습니다. 이러한 비동기 작업을 완료하기 위해 Redis 대기열 작업 시스템을 구축했습니다.

우리 모두 알고 있듯이 메시지 대기열 처리 시스템은 주로 소비자와 생산자의 두 부분으로 나뉩니다.

저희 시스템에서는 메인 시스템이 생산자 역할을 하고 작업 시스템이 소비자 역할을 합니다.

구체적인 작업 흐름은 다음과 같습니다. 1. 기본 시스템은 처리해야 하는 작업 이름 + 작업 매개변수를 대기열에 푸시합니다. 2. 작업 시스템은 작업 대기열을 실시간으로 팝합니다. 작업이 나타나면 하위 프로세스를 포크하고 하위 프로세스는 특정 작업 논리를 완료합니다.

/**
 * 启动守护进程
 */
public function runAction() {
    Tools::log_message('ERROR', 'daemon/run' . ' | action: restart', 'daemon-');
    while (true) {
        $this->fork_process();
    }
    exit;
}
 
/**
 * 创建子进程
 */
private function fork_process() {
    $ppid = getmypid();
    $pid = pcntl_fork();
    if ($pid == 0) {//子进程
        $pid = posix_getpid();
        //echo "* Process {$pid} was created \n\n";
        $this->mq_process();
        exit;
    } else {//主进程
        $pid = pcntl_wait($status, WUNTRACED); //取得子进程结束状态
        if (pcntl_wifexited($status)) {
            //echo "\n\n* Sub process: {$pid} exited with {$status}";
            //Tools::log_message('INFO', 'daemon/run succ' . '|status:' . $status . '|pid:' . $ppid . '|childpid:' . $pid );
        } else {
            Tools::log_message('ERROR', 'daemon/run fail' . '|status:' . $status . '|pid:' . $ppid . '|childpid:' . $pid, 'daemon-');
        }
    }
}
 
/**
 * 业务任务队列处理
 */
private function mq_process() {
    $data_pop = $this->masterRedis->rPop($this->redis_list_key);
    $data = json_decode($data_pop, 1);
    if (!$data) {
        return FALSE;
    }
    $worker = '_task_' . $data['worker'];
    $class_name = isset($data['class']) ? $data['class'] : 'TaskproModel';
    $params = $data['params'];
    $class = new $class_name();
    $class->$worker($params);
    return TRUE;
}