如何用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的流程图。

可以看到,在这里我们采用了两个交换器和两个队列,一个负责处理正常的任务即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实现一个异步任务系统的资料请关注我们其它相关文章!

时间: 2021-05-28

php异步多线程swoole用法实例

本文实例讲述了php异步多线程swoole用法.分享给大家供大家参考.具体分析如下: swoole重新定义PHP语言的高性能网络通信框架,提供了PHP语言的异步多线程服务,下面的实例就可以证实这一功能. 一般来说,Swoole提供了PHP语言的异步多线程服务器,异步TCP/UDP网络客户端,异步MySQL,数据库连接池,AsyncTask,消息队列,毫秒定时器,异步文件读写,异步DNS查询. swoole以前听过, 拿来做游戏服务器简直是神器-今天稍微的感受了一下,  在 ubuntu 下搭建了

PHP Swoole异步MySQL客户端实现方法示例

本文实例讲述了PHP Swoole异步MySQL客户端实现方法.分享给大家供大家参考,具体如下: 使用函数:swoole_mysql 使用版本:1.8.6及以上 把官方文档的例子改写成了面向对象的形式,示例如下: SwooleMysql.php: <?php class SwooleMysql { public $db = ""; public $server = []; public function __construct() { //实例化 $this->db = n

PHP swoole中使用task进程异步的处理耗时任务应用案例分析

本文实例讲述了PHP swoole中使用task进程异步的处理耗时任务.分享给大家供大家参考,具体如下: 我们知道,swoole中有两大进程,分别是 master 主进程和 manager 管理进程. 其中 master 主进程中会有一个主 reactor 线程和多个 reactor 线程,主要的作用就是用来维护TCP连接,处理网络IO,收发数据. 而 manager 管理进程,作用则是 fork 和管理 worker 和 task 进程. worker 进程的作用是接收 reactor 线程传

PHP Swoole异步Redis客户端实现方法示例

本文实例讲述了PHP Swoole异步Redis客户端实现方法.分享给大家供大家参考,具体如下: 使用版本:1.8.0及以上 使用条件: 1.开启async-redis php --ri swoole (如果没有开启,重新编译安装Swoole时加入 --enable-async-redis  相关文章:Linux下源码包安装Swoole及基本使用) 2.安装hiredis: CentOS yum install libhiredis-devel Ubuntu: apt install libhi

详解thinkphp5+swoole实现异步邮件群发(SMTP方式)

本文介绍了thinkphp5+swoole实现异步邮件群发(SMTP方式),分享给大家,具体如下: 1.环境说明 阿里云centos7 thinkphp5.0.11 swoole2.0.8 2.tp实现邮件发送 在项目下建立如下的文件目录: 其中SendMail.php是我们实际调用的发送邮件的文件.以下是主要代码: namespace app\library\utils\mail; use app\library\utils\mail\PhpMailer; use app\library\ut

php基于 swoole 实现的异步处理任务功能示例

本文实例讲述了php基于 swoole 实现的异步处理任务功能.分享给大家供大家参考,具体如下: 安装swoole: 下载官方swoole压缩包,解压进入目录 $ cd swoole # 进入 swoole 目录 $ phpize $ ./configure --with-php-config=/usr/local/php/bin/php-config #注意查看服务器上是否有这个路径 $ make && make install 修改php.in文件 vi /usr/local/php/

Swoole实现异步投递task任务案例详解

本文实例讲述了Swoole实现异步投递task任务案例.分享给大家供大家参考,具体如下: [使用场景] Swolle的task模块可以用来做一些异步的慢速任务.耗时场景.如webim中发广播,发送邮件等,把这些任务丢给task进程之后,worker进程可以继续处理新的数据请求,任务完成后会异步通知worker进程告诉它此任务已经完成.此外利用task还可以实现PHP的数据库连接池,异步队列等. [使用须知] 必须设置Task进程数: task_worker_num 投递一个异步任务到task_w

PHP扩展Swoole实现实时异步任务队列示例

本文实例讲述了PHP扩展Swoole实现实时异步任务队列.分享给大家供大家参考,具体如下: 假如要发100封邮件,for循环100遍,用户直接揭竿而起,什么破网站! 但实际上,我们很可能有超过1万的邮件.怎么处理这个延迟的问题? 答案就是用异步.把"发邮件"这个操作封装,然后后台异步地执行1万遍.这样的话,用户提交网页后,他所等待的时间只是"把发邮件任务请求推送进队列里"的时间.而我们的后台服务将在用户看不见的地方跑. 在实现"异步队列"这点上,

PHP swoole和redis异步任务实现方法分析

本文实例讲述了PHP swoole和redis异步任务实现方法.分享给大家供大家参考,具体如下: redis异步任务 interface.php <?php for($i=0;$i<100;$i++){ $msg = "zhezhao[".$i."]"; $redis = new Redis(); $redis->connect("127.0.0.1"); $redis->publish("test",

PHP Swoole异步读取、写入文件操作示例

本文实例讲述了PHP Swoole异步读取.写入文件操作.分享给大家供大家参考,具体如下: 异步读取文件:swoole_async_readfile 异步写入文件:swoole_async_writefile [示例] 读取文件 readfile.php: <?php $res = swoole_async_readfile(__DIR__."/1.txt", function($filename, $content) { echo "文件名:{$filename} 内

python读写文件操作示例程序

文件操作示例 复制代码 代码如下: #输入文件f = open(r'D:\Python27\pro\123.bak') #输出文件fw = open(r'D:\Python27\pro\123e.bak','w')#按行读出所有文本lines = f.readlines()num = -1for line in lines:    str = '@SES/%i/' %num    line = line.replace('@SES/1/',str)    num = num + 1    #写入

Node.js fs模块(文件模块)创建、删除目录(文件)读取写入文件流的方法

fs模块.js: /* 1. fs.stat 检测是文件还是目录(目录 文件是否存在) 2. fs.mkdir 创建目录 (创建之前先判断是否存在) 3. fs.writeFile 写入文件(文件不存在就创建,但不能创建目录) 4. fs.appendFile 写入追加文件 5.fs.readFile 读取文件 6.fs.readdir 读取目录 7.fs.rename 重命名 8. fs.rmdir 删除目录 9. fs.unlink 删除文件 */ var fs=require('fs');

Python实现的读取文件内容并写入其他文件操作示例

本文实例讲述了Python实现的读取文件内容并写入其他文件操作.分享给大家供大家参考,具体如下: 文件目录结构,如图: read_file.py是工作文件,file_test.py是读取文件源,write_test.py是写入目标文件. 文件A:file_test.py #coding=utf-8 for i in range(1, 10): print i 文件B:read_file.py # coding=utf-8 # 打开件A f = open('./file_test.py', 'rb

Python实现的读取/更改/写入xml文件操作示例

本文实例讲述了Python实现的读取/更改/写入xml文件操作.分享给大家供大家参考,具体如下: 原始文档内容(test.xml): <?xml version="1.0" encoding="UTF-8"?> <framework> <processers> <processer name="AProcesser" file="lib64/A.so" path="/tmp&

PHP实现读取文件夹及批量重命名文件操作示例

本文实例讲述了PHP实现读取文件夹及批量重命名文件操作.分享给大家供大家参考,具体如下: 以读取从网上下载的评书解压后的文件夹为例,并批量重命名 文件夹内容如下,现在使用php读取这个文件夹里面的文件,并批量重命名为"隋唐演义+序号.mp3"的格式: <?php $dir1 = './dir1/';//读取文件夹名称 (重命名为dir1避免中文报错,移动到与要执行的php文件同级目录) $filearr = scandir($dir1);//读取文件夹 print_r($file

Java RandomAccessFile基本文件操作示例

本文实例讲述了Java RandomAccessFile基本文件操作.分享给大家供大家参考,具体如下: 一 简介 RandomAccessFile:java提供对文件内容的访问,既可以读文件,也可以写文件. RandomAccessFile支持随机访问文件,可以访问文件的任意位置 1.java文件模型 在硬盘上的文件是byte byte byte存储的,是数据的集合 2.打开文件 有两种模式"rw"(读写) "r"(只读) RandomAccessFile raf

PHP5.6读写excel表格文件操作示例

本文实例讲述了PHP5.6读写excel表格文件操作.分享给大家供大家参考,具体如下: 测试环境:php5.6.24.这块没啥兼容问题. 需要更多栗子,请看PHPExcel的examples.还是蛮强大的. 读取excel文件: 第一步.下载开源的PHPExcel的类库文件,官方网站是http://www.codeplex.com/PHPExcel.里面也有很多示例包. 或者从本站下载:https://www.jb51.net/codes/194070.html 第二步.读取的基本代码示例: <