# 任务管理服务

版本修订

作者	时间	内容
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 任务执行的日志目录，按每天一个文件夹

1
2
3
4
5
6
7
8
9
10

# 配置：

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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

核心配置的选项

队列任务存储的介质redis/rabbitmq
临时文件的暂存介质memcache/filesystem
最终文件生成方式local/ftp
任务请求的域名真实地址(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

1
2
3
4
5
6
7
8
9
10
11

# 任务日志的清理：

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

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

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

# Step 1 增加任务白名单

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

截图一

数组中的Key(比如:autochk)是具体任务的标识
method是具体任务标识对应的处理方式(比如autochk对应的是wms/lib/autotask/task下的check.php)
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下的分钟、小时、天、周、月的函数里

← 日志服务运行环境要求 →