如何用RabbitMQ和Swoole实现一个异步任务系统

2022-04-15 246
目录
  • 系统介绍
  • 事件生产者
  • 任务调度器
  • 消费者
    • 正常任务
    • 延迟任务
  • 自定义调度器

    系统介绍

    如何用RabbitMQ和Swoole实现一个异步任务系统

    从图中可以看到,我们这个系统是一个基于事件的异步任务系统。就是说当一个事件产生时,生产者将事件抛给调度器,调度器负责查询事件下有哪些任务,然后将这些任务丢到相应的队列中,最后由消费者消费任务队列中的任务。

    在整个系统中主要分为三大部分

    1.事件生产者,即产生消息事件的一方。

    2.任务调度器(Scheduler),负责注册事件并调度任务。

    3.消费者(Worker),负责消费任务队列中的任务。

    事件生产者

    事件生产者很简单,在业务系统中直接调用即可,代码如下。

    <?php
     
    require_once DIR.'/../autoload.php';
     
    use Asynclib\Ebats\Event;
     
    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的流程图。

    如何用RabbitMQ和Swoole实现一个异步任务系统

    可以看到,在这里我们采用了两个交换器和两个队列,一个负责处理正常的任务即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 Asynclib\Ebats\Worker;
     
      
     
    //执行结果回调函数
     
    $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 Asynclib\Ebats\Task;
     
    use Asynclib\Core\Consumer;
     
    use Asynclib\Amq\ExchangeTypes;
     
    use Asynclib\Exception\ExceptionInterface;
     
      
     
    /**
     
     * 本示例演示了如何创建一个自定义调度器,开发者可以根据自身需求开发自己的任务调度器
     
     */
     
    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 \n";
     
            Task::create('demo', 'orderAsync', $msg);//创建任务,之后消息将作为参数由任务接管处理
     
        });
     
    }catch (ExceptionInterface $exc){
     
        echo $exc->getMessage();
     
    }

    这样,当接收到消息时就会产生一个orderAsync的任务,您只需要启动一个用来消费这个Topic的Worker即可。

    也许你会觉得这里直接写业务逻辑的代码就可以了,实际上也确实可以。当你可以忍受一个进程慢慢消费的时候是可以这样做的。但大多数情况下我们还是希望它能够尽快的消费掉,所以建议这里只负责创建任务,具体任务的业务逻辑由worker去执行。

    以上就是如何用RabbitMQ和Swoole实现一个异步任务系统的详细内容,更多关于用RabbitMQ和Swoole实现一个异步任务系统的资料请关注NICE源码其它相关文章!

    免责声明:
    1、本网站所有发布的源码、软件和资料均为收集各大资源网站整理而来;仅限用于学习和研究目的,您必须在下载后的24个小时之内,从您的电脑中彻底删除上述内容。 不得使用于非法商业用途,不得违反国家法律。否则后果自负!

    2、本站信息来自网络,版权争议与本站无关。一切关于该资源商业行为与www.niceym.com无关。
    如果您喜欢该程序,请支持正版源码、软件,购买注册,得到更好的正版服务。
    如有侵犯你版权的,请邮件与我们联系处理(邮箱:skknet@qq.com),本站将立即改正。

    NICE源码网 PHP编程 如何用RabbitMQ和Swoole实现一个异步任务系统 https://www.niceym.com/18078.html

    常见问题
    • 请您联系网站客服QQ,79024519,将第一时间处理您的问题,咨询请提供会员ID以及资源编号!
    查看详情
    • 本站发布的压缩包如遇需要解压密码请尝试本站域名:www.niceym.com (通常为小写,如不对可试试大写) 或者:niceym.com 如遇无法解压,请您提供资源编号发送给本站客服!
    查看详情