第如何用RabbitMQ和Swoole實現(xiàn)一個異步任務系統(tǒng)從圖中可以看到，我們這個系統(tǒng)是一個基于事件的異步任務系統(tǒng)。就是說當一個事件產(chǎn)生時，生產(chǎn)者將事件拋給調(diào)度器，調(diào)度器負責查詢事件下有哪些任務，然后將這些任務丟到相應的隊列中，最后由消費者消費任務隊列中的任務。

在整個系統(tǒng)中主要分為三大部分

1.事件生產(chǎn)者，即產(chǎn)生消息事件的一方。

2.任務調(diào)度器(Scheduler)，負責注冊事件并調(diào)度任務。

3.消費者(Worker)，負責消費任務隊列中的任務。

事件生產(chǎn)者

事件生產(chǎn)者很簡單，在業(yè)務系統(tǒng)中直接調(diào)用即可，代碼如下。

require_onceDIR.'/../autoload.php';

useAsynclib\Ebats\Event;

$event=newEvent('order_paied');//定義事件

$event-setOptions(['order_id'='FB138020392193312']);//事件產(chǎn)生的參數(shù)

$event-publish();

}catch(Exception$exc){

echo$exc-getMessage();

}

任務調(diào)度器

調(diào)度器主要做兩件事，一是注冊事件，另一個是調(diào)度任務。

注冊事件代碼如下：

//注冊事件

EventManager::register('order_create','closeOrder','demo',10);//關閉未付款訂單(延遲任務)

EventManager::register('order_paied','virtualShipping','demo');//虛擬商品自動發(fā)貨

這樣就注冊了兩個事件，事件下各有一個任務。

具體調(diào)度部分代碼很簡單，就不多贅述，有興趣的可以去看代碼。

重頭戲來了，一個異步任務系統(tǒng)最重要的就是消費端了，現(xiàn)在讓我們來看下Worker的流程圖。

可以看到，在這里我們采用了兩個交換器和兩個隊列，一個負責處理正常的任務即ntask，另一個負責處理需要延遲執(zhí)行的任務即dtask。簡單描述下一個任務的生命周期。

1、task產(chǎn)生，進入正常任務的交換器Exchange[ebats_core_ntask]

2、交換器根據(jù)topic將任務分發(fā)到對應的隊列中

3、子進程ntask阻塞等待成功獲取到task，并執(zhí)行該任務

4、執(zhí)行失敗，需要重試時拋出RetryException，不需要重試時拋出TaskException

5、子進程ntask捕獲到重試異常將任務拋給延遲任務的交換器Exchange[ebats_core_dtask]

6、將任務執(zhí)行信息回調(diào)給上層開發(fā)者以便保存查看

1、子進程dtask阻塞等待成功獲取到task，并執(zhí)行該任務

2、執(zhí)行失敗，需要重試時拋出RetryException，不需要重試時拋出TaskException

3、子進程dtask捕獲到重試異常將任務拋給延遲任務的交換器Exchange[ebats_core_dtask]

4、將任務執(zhí)行信息回調(diào)給上層開發(fā)者以便保存查看

消費者代碼如下：

require_onceDIR.'/../autoload.p