任务管理服务

版本修订

作者 时间 内容
kami long long ago 初版
kami 2016-04-28 任务管理优化、接管原来的后台伪队列、定时任务。如果要扩展定时单次或者循环任务可以在timer任务类型里增加
kami 2016-06-30 针对使用redis作为队列服务部署在某些服务器上负载飙升的问题解决,在配置项中增加wait_time参数,识别队列为空时短暂休眠

基于3.5.0+版本新增的任务管理服务(TASKMGR APP)

任务管理的设计和研发主要是为了解决OMS中一些定时任务或者标化的业务处理任务的执行效率、性能以及并发处理问题。

我们大致将OMS中的任务分别五类:

为了方便管理,我们对软件的安装做了一些简单的约定

1.定时任务timer

原先OMS后台的一些队列任务,按分钟、小时、天、月触发的任务以及报表等

2.一般标化处理任务task

管理员操作触发的任务,批量发货、校验

3.导出任务export

数据导出后台任务

4.初始化定时任务init

将定时任务加入队列,定时任务依赖这个初始化任务,比如每半小时出发一个任务或者固定时间点14点执行一个任务,就会在这里初始化丢进队列

5.请求响应及回调队列任务rpc

实现将OMS相关的请求回调及响应内容走队列,并做流速可控

注:由于定时任务没有相应的参数而和时间有关,所以任务生成到队列中需要借助taskmgr任务管理中的另外一个init(初始化定时任务)任务去实现。

Taskmgr APP的核心程序文件及路径:

Taskmgr/taskDaemon.php 任务启动脚本
Taskmgr/lib/controller 不同类型任务的命名
Taksmgr/lib/task 具体线程任务,按任务类型区分文件夹task、export、init、rpc、timer
Taskmgr/lib/rpc 任务请求接收入口及签名算法
Taskmgr/lib/connecter 任务的存储介质
Taskmgr/lib/cache 导出分片任务的临时存储介质
Taskmgr/lib/storage 导出任务最终存储介质
Taskmgr/lib/third php socket直连amqp方式,不走amqp php C扩展
Taskmgr/lib/thread 线程操作的lib封装
Taskmgr/logs 任务执行的日志目录,按每天一个文件夹

配置:

error_reporting(E_ALL & ~E_NOTICE & ~E_DEPRECATED);

//定义数据提供者队列
//define('__CONNECTER_MODE','rabbitmq');

//define('__RABBITMQ_INTERFACE__','pecl');

//mq配置定义
/*
$GLOBALS['__RABBITMQ_CONFIG'] = array(

    'HOST'      => '127.0.0.1',
    'PORT'      => '5672',
    'USER'      => 'task',
    'PASSWD'    => 'task123',
    'VHOST'     => 'erp_task',
    'ROUTER'    => 'erp.task.%s.*',
    'QUEUE_PREFIX' => 'ERP',
);
*/

define('__CONNECTER_MODE','redis');

$GLOBALS['__REDIS_CONFIG'] = array(
    'HOST'      => '127.0.0.1',
    'PORT'      => '6379',
    //'PASSWD'    => 'erp3310',
    'QUEUE_PREFIX' => 'ERP',
    'WAIT_TIME' => 1,
);

//缓存存储介质提供者
//define('__CACHE_MODE','memcache');

//define('__MEMCACHE_CONFIG','127.0.0.1:11211,127.0.0.1:11212');

define('__CACHE_MODE','filesystem');

//文件存储介质提供者
//define('__STORAGE_MODE','ftp');

/*
$GLOBALS['__STORAGE_CONFIG'] = array(
    'HOST'    => '127.0.0.1',
    'PORT'    => '21',
    'USER'    => 'test',
    'PASSWD'  => 'test',
    'TIMEOUT' => '30',
    'PASV'    => false,
);
*/

define('__STORAGE_MODE','local');

//设置为真实的域名
define('DOMAIN', '127.0.0.1');

//定义内部任务请求的token
define('REQ_TOKEN', 'SzjJ7VF9w2Dy3yt4K9vcV3LEwFc5pX3cXHu');

核心配置的选项

  1. 队列任务存储的介质redis/rabbitmq

  2. 临时文件的暂存介质memcache/filesystem

  3. 最终文件生成方式local/ftp

  4. 任务请求的域名真实地址(DOMAIN)以及请求时候的验签名token(REQ_TOKEN)

PS:

针对队列的配置除了账号基本信息外,我们还可以自定义队列名的前缀

2和3选项主要是针对导出任务的信息存储

任务的启动与暂停:

默认情况下,我们只要在crontab里部署check.sh守护进行即可启动所有类型的任务,命令行如下:

* * * * * /bin/bash /data/httpd/56/app/taskmgr/check.sh

当然我们也可以针对单个任务类型进行启用与暂停,这里需要注意当我们修改了程序代码后,需要重启任务不然是无法直接看到效果的。

/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php start timer
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php start task
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php start export
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php start init
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php start rpc

/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php stop timer
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php stop task
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php stop export
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php stop init
/usr/local/php-pthreads/bin/php /data/taskmgr/taskDaemon.php stop rpc

任务日志的清理:

也是通过部署crontab任务,具体触发时间自己定义凌晨跑即可,清理两周前的日志

0 1 * * * /bin/bash /data/httpd/56/app/taskmgr/cleanlogs.sh

如何在一个类型任务中添加一种处理任务,以下我们用批量校验为例:

Step 1 增加任务白名单

找到taskmgr/lib/whitelist.php , 在对应的任务列表里添加相应的代码:

截图一

  1. 数组中的Key(比如:autochk)是具体任务的标识
  2. method是具体任务标识对应的处理方式(比如autochk对应的是wms/lib/autotask/task下的check.php)
  3. threadNum是具体当前队列任务处理的最大启用线程数,一般就按默认的设置为准(高能预警,非战斗人员请勿随意修改该参数)

Step 2 添加任务的线程处理程序

Taskmgr/lib/task/task/autochk.php 常规的业务处理继承abstract.php 大致程序处理逻辑为:

在线程中连接mq队列,堵塞模式(类似死循环不停的取要处理的任务)取数据,打http请求。

当返回成功的时候,告诉mq队列 ack任务已处理(失败的话里面加了三次重试,但有的任务可能不需要重试,这里要注意)。

当你在白名单中添加了一个具体任务类型的任务后,重启这个任务。该任务就变成运行状态。

PS:后续重试机制会改成按单个任务可配置,未完待续。。。。

Step 3 添加具体的处理任务处理Lib

批量校验的任务处理是wms/lib/autotask/task/check.php,这个就是具体任务具体怎么处理了。

方法名固定为process,params为传入参数,error_msg为处理时遇到的错误可以返回,处理方法最终执行成功的话,需要返回true,失败则为false

Step 4 如何添加一个任务到队列

Taskmgr提供了一个添加任务的lib方法,在taskmgr/lib/interface/connecter.php

截图二

url和task_type是必要参数,url是目标客户OMS的自动任务接收的openapi域名地址,task_type就是任务的标示,也就是上面白名单中的数组key,其他参数就是具体任务处理需要用到的应用级参数。

注:由于taskmgr新版本已经接管了原来的ecos框架misctask任务,所以建议二次开发的时候,不要再把定时任务放在misctask下的分钟、小时、天、周、月的函数里


沪ICP备05002918号

© 2003-2017 ShopEx,Inc.All rights reserved.