RabbitMQ 및 Swoole을 기반으로 한 완전한 비동기 작업 시스템

초기 단일 프로세스 소비 비동기 작업 시스템에서 redis를 사용하여 구현됨 Swoole의 다중 프로세스 소비 모델을 통해 우리의 비동기 작업 시스템이 마침내 한 단계 더 발전할 수 있게 되었습니다.

이전의 두 가지 간단한 시스템의 경험으로 인해 이번에는 RabbitMQ 기반의 비동기 작업 시스템이 더욱 개선되었습니다. , 다중 프로세스 소비, 예외 재시도 등

시스템 소개

소비자 측아키텍처그림

그림에서 볼 수 있듯이 우리 시스템은 이벤트 기반의 비동기 작업 시스템입니다. 즉, 이벤트가 발생하면 Producer가 해당 이벤트를 스케줄러에 전달하고, 스케줄러는 이벤트 아래에 어떤 작업이 있는지 쿼리한 다음 해당 작업을 해당 대기열에 넣고 마지막으로 소비자는 작업 대기열

시스템 전반에 걸쳐

1. 메시지 이벤트를 생성하는 당사자인 이벤트 프로듀서

2. 이벤트 등록 및 작업 예약을 담당하는 작업 스케줄러(Scheduler).
3. 소비자는 작업 대기열의 작업을 담당합니다.

이벤트 생산자는 비즈니스 시스템에서 직접 호출할 수 있습니다. .

작업 스케줄러

스케줄러는 주로 두 가지 작업을 수행합니다. 하나는 이벤트를 등록하는 것이고 다른 하나는 작업을 예약하는 것입니다.

등록 이벤트 코드는 다음과 같습니다.

<?php
require_once DIR.&#39;/../autoload.php&#39;;
use Asynclib\Ebats\Event;
try{
    $event = new Event(&#39;order_paied&#39;);  //定义事件
    $event->setOptions(['order_id' => 'FB138020392193312']); //事件产生的参数
    $event->publish();
}catch (Exception $exc){
    echo $exc->getMessage();
}

이렇게 하면 각각 하나의 작업으로 두 개의 이벤트가 등록됩니다.

구체적인 일정 코드는 매우 간단하므로 자세히 설명하지 않겠습니다.

Consumer

여기서 가장 중요한 부분이 바로 Worker의 흐름도를 살펴보겠습니다.

전체 소비 프로세스

여기서는 두 개의 스위치와 두 개의 대기열을 사용하는 것을 볼 수 있습니다. 하나는 일반 작업, 즉 ntask 처리를 담당하고 다른 하나는 다음을 담당합니다. 지연이 필요한 작업, 즉 dtask를 간단하게 설명합니다.

다음 작업의 라이프사이클

일반 작업

1. 태스크 교환 Exchange[ebats_core_ntask]

2. 교환 주제에 따라 해당 큐에 태스크를 분배합니다3. 하위 프로세스 ntask는 태스크가 성공적으로 획득되기를 기다리고 태스크를 실행합니다 4. 재시도가 필요하지 않은 경우 실행이 실패하고 RetryException이 발생합니다. TaskException

5이 발생합니다. 지연된 작업 교환 [ebats_core_dtask]

6. 작업 실행 정보를 상위 개발자에게 호출하여 저장 및 보기

지연된 작업

1. 작업이 성공적으로 획득될 때까지 기다렸다가 작업을 실행합니다
2. 실행이 실패하고 재시도가 필요한 경우 RetryException이 발생하며, 재시도가 필요하지 않은 경우에는 TaskException이 발생합니다.
3. -process dtask는 재시도 예외를 캡처하고 지연된 작업 교환 Exchange [ebats_core_dtask]에 작업을 전달합니다.
4. 저장하고 볼 수 있도록 작업 실행 정보를 상위 개발자에게 다시 호출합니다.

소비 코드

//注册事件
EventManager::register('order_create', 'closeOrder', 'demo', 10);//关闭未付款订单(延迟任务)
EventManager::register('order_paied', 'virtualShipping', 'demo'); //虚拟商品自动发货

커스텀 스케줄러

일반적으로 이벤트 기반 작업 시스템인데 직접 작업을 생성할 수 있나요? 대답은 '예'입니다.

사용자 정의 스케줄러를 생성하고, 스스로 스케줄링 로직을 구현하고, 마지막으로 작업을 생성하기만 하면 됩니다. 코드는 다음과 같습니다.

require_once DIR.'/../autoload.php';
require_once DIR.'/task/TaskDemoModel.php';
use Asynclib\Ebats\Worker;

//执行结果回调函数
$callback = function ($topic, $taskid, $taskname, $params, $timeuse, $message){

};
$worker = new Worker($callback);  //支持多进程消费默认为1
$worker->setQueue('demo');  //队列名和事件的topic一一对应
$worker->run();

이런 방식으로 메시지가 수신되면 orderAsync 작업이 생성됩니다. 이 주제를 사용하려면 작업자만 시작하면 됩니다.

여기서 비즈니스 로직 코드를 직접 작성하는 것만으로도 충분하다고 생각할 수도 있지만 실제로는 가능합니다. 천천히 소모되는 프로세스를 견딜 수 있을 때 이 작업을 수행할 수 있습니다. 그러나 대부분의 경우 여전히 가능한 한 빨리 소비되기를 원하므로 여기에서만 작업을 생성하고 특정 작업의 비즈니스 로직은 작업자에 의해 실행되는 것이 좋습니다.

위 내용은 RabbitMQ 및 Swoole을 기반으로 한 완전한 비동기 작업 시스템의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!