从最开始的使用redis实现的单进程消费的异步任务系统到加入swoole的多进程消费模式，现在，我们的异步任务系统终于又能迈进一步。
因为有了前面两个简单系统的经验，这回基于RabbitMQ的异步任务系统设计的的更加完善，包括多进程消费，异常重试等。
系统介绍
消费端架构图
从图中可以看到，我们这个系统是一个基于事件的异步任务系统。就是说当一个事件产生时，生产者将事件抛给调度器，调度器负责查询事件下有哪些任务，然后将这些任务丢到相应的队列中，最后由消费者消费任务队列中的任务。
在整个系统中主要分为三大部分
1.事件生产者，即产生消息事件的一方。
2.任务调度器(Scheduler)，负责注册事件并调度任务。
3.消费者(Worker)，负责消费任务队列中的任务。
事件生产者
事件生产者很简单，在业务系统中直接调用即可，代码如下。
<?php
require_once DIR.'/../autoload.php';
use AsynclibEbatsEvent;
try{
$event = new Event('order_paied'); //定义事件
$event->setOptions(['order_id' => 'FB138020392193312']); //事件产生的参数
$event->publish();
}catch (Exception $exc){
echo $exc->getMessage();
}
任务调度器
调度器主要做两件事，一是注册事件，另一个是调度任务。
注册事件代码如下：
//注册事件
EventManager::register('order_create', 'closeOrder', 'demo', 10);//关闭未付款订单(延迟任务)
EventManager::register('order_paied', 'virtualShipping', 'demo'); //虚拟商品自动发货
这样就注册了两个事件，事件下各有一个任务。
具体调度部分代码很简单，就不多赘述，有兴趣的可以去看代码。
消费者
重头戏来了，一个异步任务系统最重要的就是消费端了，现在让我们来看下Worker的流程图。
一个完整的消费进程
可以看到，在这里我们采用了两个交换器和两个队列，一个负责处理正常的任务即ntask，另一个负责处理需要延迟执行的任务即dtask。简单描述下一个任务的生命周期。
正常任务
1、task产生，进入正常任务的交换器Exchange[ebats_core_ntask]
2、交换器根据topic将任务分发到对应的队列中
3、子进程ntask阻塞等待成功获取到task，并执行该任务
4、执行失败，需要重试时抛出RetryException，不需要重试时抛出TaskException
5、子进程ntask捕获到重试异常将任务抛给延迟任务的交换器Exchange[ebats_core_dtask]
6、将任务执行信息回调给上层开发者以便保存查看
延迟任务
1、子进程dtask阻塞等待成功获取到task，并执行该任务
2、执行失败，需要重试时抛出RetryException，不需要重试时抛出TaskException
3、子进程dtask捕获到重试异常将任务抛给延迟任务的交换器Exchange[ebats_core_dtask]
4、将任务执行信息回调给上层开发者以便保存查看
消费者代码如下：
require_once DIR.'/../autoload.php';
require_once DIR.'/task/TaskDemoModel.php';
use AsynclibEbatsWorker;
//执行结果回调函数
$callback = function ($topic, $taskid, $taskname, $params, $timeuse, $message){
};
$worker = new Worker($callback); //支持多进程消费默认为1
$worker->setQueue('demo'); //队列名和事件的topic一一对应
$worker->run();
自定义调度器
一般来说这是一个基于事件的任务系统，那么能不能直接产生任务呢。答案是肯定的。
只需要创建一个自定义调度器，由您自行实现调度逻辑，最终生成一个任务即可。代码如下：
<?php
require_once DIR.'/../autoload.php';
use AsynclibEbatsTask;
use AsynclibCoreConsumer;
use AsynclibAmqExchangeTypes;
use AsynclibExceptionExceptionInterface;
/**
* 本示例演示了如何创建一个自定义调度器,开发者可以根据自身需求开发自己的任务调度器
*/
try{
$worker = new Consumer();
$worker->setExchange('order_fanout', ExchangeTypes::TOPIC);
$worker->setQueue('shzf_order_paied', ['*.*.WAIT_SELLER_SEND_GOODS']);
$worker->run(function($key, $msg){
$order_data = json_encode($msg);
echo " [$key] $order_data
";
Task::create('demo', 'orderAsync', $msg);//创建任务,之后消息将作为参数由任务接管处理
});
}catch (ExceptionInterface $exc){
echo $exc->getMessage();
}
这样，当接收到消息时就会产生一个orderAsync的任务，您只需要启动一个用来消费这个Topic的Worker即可。
也许你会觉得这里直接写业务逻辑的代码就可以了，实际上也确实可以。当你可以忍受一个进程慢慢消费的时候是可以这样做的。但大多数情况下我们还是希望它能够尽快的消费掉，所以建议这里只负责创建任务，具体任务的业务逻辑由worker去执行。
以上就是基于RabbitMQ和Swoole实现的一个完整的异步任务系统的详细内容
视频教学：九年架构师讲解php+swoole+rabbitMQ实现异步任务多进程消费
更多内容请访问
怎么从一名码农成为架构师的必看知识点：目录大全（持续更新）50W年薪挑战！
我的官方群点击此处加入群聊【PHP/web/高级学习交流群】,一起学习，相互讨论。
群内已经有管理将知识体系整理好（源码，学习视频等资料），欢迎加群免费领取。
这套精品PHP教程绝不是市场上的那些妖艳贱货可比，作为web开发的佼佼者PHP并不逊色其他语言，加上Swoole后更加是如虎添翼！进军通信 、物联网行业开发百度地图、百度订单中心、虎牙、战旗TV等！寒冬裁员期过后正是各大企业扩大招人的时期，现在市场初级程序员泛滥，进阶中高级程序员绝对是各大企业急需的人才，这套学习教程适合那些1-5年以内的PHP开发者正处于瓶颈期，想要突破自己进阶中高级、架构师！名额有限，先到先得！
腾讯T3-T4标准精品PHP架构师教程目录大全，只要你看完保证薪资上升一个台阶（持续更新）
部分资料截图：
还有限时精品福利：
★腾讯高级PHP工程师笔试题目
★亿级PV高并发场景订单的处理
★laravel开发天猫商城组件服务
★战旗TV视频直播的架构项目实战
扫描下面二维码领取
对PHP后端技术，对PHP架构技术感兴趣的朋友，我的官方群点击此处，一起学习，相互讨论。
群内已经有管理将知识体系整理好（源码，学习视频等资料），欢迎加群免费领取。
本课程深度对标腾讯T3-T4标准，贴身打造学习计划为web开发人员进阶中高级、架构师提升技术，为自己增值涨薪！加入BAT特训营还可以获得内推大厂名额以及GO语言学习权限！！！