任务调度是后端重要的组成部分，用于的场景非常广泛，比如生产系统中跑一些定时任务，涉及到上千台机器管理就很困难；Web应用中用户触发的操作比较耗时，这部分可以放到异步任务中处理；离线数据处理多个任务，并且任务间直接有依赖关系...
分布式任务调度通常有以下几类实现方式，一类是基于机器调度的方式，如CT(Contab Task)，百度内部是这种方式，定时执行指定机器上的具体指令，通常需要先把任务脚本发布到具体机器，提供给机器调度(worker)；第二类如Gearman/Celery，Client将任务发送给Job Server，Server根据后端负载情况，将任务投递到Worker；还有一类是编程框架层面提供的Schedule，功能相对局限，比如只能当前这台机器执行，当前机器挂了，任务也就挂了。
目前自己工作内容涉及到不少离线数据计算，这类任务是基于Python实现，公司内部现用调度系统对非Java支持不完善，而离线任务较多，其中任务间还有依赖关系，使用系统自带的Contab，不管从任务稳定性，还是任务间依赖关系的处理都不完善，最终选择部署Celery集群作为离线任务调度框架。Celery是一个专注于实时处理的任务队列，同时也支持任务调度，配套的任务监控也很完善，可以使用Flower作为可视化监控工具。
1、Celery任务架构
Celery有三个核心组件：Celery client: 用于发布后台作业，当与应用一起工作的时候，客户端与 Flask 应用一起运行。
Celery workers: 运行后台作业的进程。Celery 支持本地和远程的 workers，可以在本地服务器上启动一个单独的 worker，也可以在远程服务器上启动worker。
Broker: 消息代理，客户端通过消息队列和 workers 进行通信，Celery 支持多种方式来实现这些队列。最常用的代理就是 RabbitMQ 和 Redis。
2、编写任务实例
(1)安装celerypip install celery
(2)celery_con.pyfrom celery import Celery
import time
celery = Celery('tasks', broker='redis://localhost:6379/0')
(3)task.pyfrom celery_con import app
@app.task
def test(x, y):
time.sleep(5)
return x + y
@app.task
def scan(x,y):
time.sleep(1)
return x-y
(4)发布任务celery -A task worker -c 2
(5)触发任务from task import test,scan
res=test.delay(2,2)
print res.get()
3、任务依赖
任务间通常存在依赖关系，例如后续任务依赖前一个任务的结果，或者是依赖几个任务的完成，前几个任务有可能支持并行、也有可能串行，Celery提供了signature和chain，可以保证任务按照用户构想执行。比如：from celery import chain
from proj.tasks import add
# (4 + 4) * 8
chain(add.s(4, 4) | add.s(8))().get()
64
上面这种示例是后续任务依赖前一步的结果，有时候不需要前一步的结果，只需要前一步任务完成(add.s(4,4) | add.si(5,6)).get
4、异常处理@app.task(bind=True, default_retry_delay=300, max_retries=5)
def my_task_A():
try:
print("doing stuff here...")
except SomeNetworkException as e:
print("maybe do some clenup here....")
self.retry(e)
一般添加 default_retry_delay 重试等待时间和 max_retries 重试次数来限定，防止任务无限重试。
5、任务监控
作为一个完善的任务监控系统，需要有对任务的监控，Flower提供了对Celery任务的监控，使用起来也特别方便，可以通过UI对查看任务详细、终止任务、重试任务。
flower --port=5555 --broker=redis://localhost:6379/0
broker指定成Celery的broker即可
以上是Celery的简单用法，详细的可查看Celery官方文档。

一、概述
1、自动化部署能简化开发过程的代码管理，让开发人员把更多的时间专注于业务实现，
简化繁琐的上线流程和操作步骤，做到项目的快速打包和部署，减少人为操作中的失误。
2、分布式任务调度平台，XXL-JOB：美团点评员工许雪里开发，
其核心设计目标是开发迅速、学习简单、轻量级、易扩展。
现已开放源代码并接入多家公司线上产品线，开箱即用。
源码地址：https://github.com/xuxueli/xxl-job
二、本文简介
主要是利用成熟的CI部署工具Jenkins，结合开源的分布式任务调度框架xxl-job作为demo工程，快速搭建自己的持续集成项目，其他项目可以类似构建。
三、Jenkins部署和配置
1、安装
1.1用特定版本的安装包安装
安装包下载页：   https://pkg.jenkins.io/redhat-stable/
(注意：会有一些安装版本和jdk版本匹配的问题，一般CentOS 6.5 安装版本jenkins-2.7.4-1.1.noarch.rpm是OK的)
1.2、安装最新版本：
按照文档来就没问题： https://jenkins.io/doc/book/getting-started/installing/#unix-linux
1.3、启动和停止
sudo service jenkins start/stop/restart
sudo chkconfig jenkins on
2、全局配置
2.1 全局配置：点击系统管理，进入Global Tool Configuration
2.2 配置常用环境变量
注意的点：所有配置需要配置在全局环境变量里面，让系统所有用户都能访问到，比如说maven，不能配置在特定的用户目录下，否则后面jenkins用户就无法执行mvn命令，而报错。
jdk：
git:
maven:
3、 项目配置
3.1、新建一个项目
点击新建，新建一个自由风格的软件项目
3.2、项目源码配置(使用git管理):
首先需要安装插件：
系统管理-》管理插件，在可选插件里找到这个，让后直接安装就可以
3.3、配置远程SSH
在系统管理-》系统设置里，找到Publish over SSH
4、构建配置详细：
1、 General配置项目名称和简介
2、源码管理配置git项目地址
3、构建触发器配置远程构建脚本
4、构建环境配置项目构建前需要执行的一些脚本
5、构建，配置构建工具，比如maven打包，war包的上传，执行tomcat服务启动脚本
6、构建后操作发布，构建后的邮件发送等操作
4.1、简介和配置
4.2、 配置项目源码地址和访问密码
4.3、 配置Credentials
4.4、构建环境
4.5、构建 配置构建命令 (这里的maven是前面全局环境变量中配置好的)
添加构建后操作(需要前面安装好远程ssh插件才行)
4.6、配置构建后操作
其中:
Transfer SetSource files：表示要上传的本地的war包及路径，可到工作空间去看
Remove prefix:表示要上传时要去除的文件夹，即只上传war包
remote driectory:即表示执行时的路径，相当于把war包上传到这里了
exec commad:要执行的命令，这里可以重命名jar包，启动tomcat服务器
7、构建后操作：
8、回到配置主页，进行构建
9、进入构建控制台查看构建日志，如果有错误，可以直接查看到
四、xxl-job的介绍
项目地址:
https://github.com/xuxueli/xxl-job
概述：
XXL-JOB是一个轻量级分布式任务调度框架，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。
现已开放源代码并接入多家公司线上产品线，开箱即用。
特性：
- 1、简单：支持通过Web页面对任务进行CRUD操作，操作简单，一分钟上手；
- 2、动态：支持动态修改任务状态、暂停/恢复任务，以及终止运行中任务，即时生效；
- 3、调度中心HA(中心式)：调度采用中心式设计，“调度中心”基于集群Quartz实现，可保证调度中心HA；
- 4、执行器HA(分布式)：任务分布式执行，任务"执行器"支持集群部署，可保证任务执行HA；
- 5、任务Failover：执行器集群部署时，任务路由策略选择"故障转移"情况下调度失败时将会平滑切换执行器进行Failover；
- 6、一致性：“调度中心”通过DB锁保证集群分布式调度的一致性, 一次任务调度只会触发一次执行；
- 7、自定义任务参数：支持在线配置调度任务入参，即时生效；
- 8、调度线程池：调度系统多线程触发调度运行，确保调度精确执行，不被堵塞；
- 9、弹性扩容缩容：一旦有新执行器机器上线或者下线，下次调度时将会重新分配任务；
- 10、邮件报警：任务失败时支持邮件报警，支持配置多邮件地址群发报警邮件；
- 11、状态监控：支持实时监控任务进度；
- 12、Rolling执行日志：支持在线查看调度结果，并且支持以Rolling方式实时查看执行器输出的完整的执行日志；
- 13、GLUE：提供Web IDE，支持在线开发任务逻辑代码，动态发布，实时编译生效，省略部署上线的过程。支持30个版本的历史版本回溯。
- 14、数据加密：调度中心和执行器之间的通讯进行数据加密，提升调度信息安全性；
- 15、任务依赖：支持配置子任务依赖，当父任务执行结束且执行成功后将会主动触发一次子任务的执行, 多个子任务用逗号分隔；
- 16、推送maven中央仓库: 将会把最新稳定版推送到maven中央仓库, 方便用户接入和使用;
- 17、任务注册: 执行器会周期性自动注册任务, 调度中心将会自动发现注册的任务并触发执行。同时，也支持手动录入执行器地址；
- 18、路由策略：执行器集群部署时提供丰富的路由策略，包括：第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移；
- 19、运行报表：支持实时查看运行数据，如任务数量、调度次数、执行器数量等；以及调度报表，如调度日期分布图，调度成功分布图等；
- 20、脚本任务：支持以GLUE模式开发和运行脚本任务，包括Shell、Python等类型脚本;
- 21、阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行(默认)、丢弃后续调度、覆盖之前调度；
- 22、失败处理策略；调度失败时的处理策略，策略包括：失败告警(默认)、失败重试；

Celery是一个分布式的任务调度系统
任务队列是一种在线程或机器间分发任务的机制
消息队列的输入是工作的一个单元，成为任务，独立的职称(Worker)进程持续监视队列中是否有需要处理的新任务
Celery用于消息通信，通常使用中间人(Broker)在客户端和职程间斡旋。这个过程从客户端向队列添加消息开始，之后中间人把消息派发给职程
Celery系统可包含多个职程和中间人，一次获得高可用性和横向扩展能力
Celery需要一个发送和接受消息的传输者。RabbitMQ和 Redis中间人的消息传输支持所有特性，但也提供大量其他实验性方案的支持，包括用 SQLite进行本地开发
Celery可以单机运行，也可以在多台机器上运行，甚至可以跨越数据中心运行
比如编写tasks.py
from celery import Celery
import time
celery = Celery('tasks', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0')
@celery.task
def add(a, b):
# 模拟一个耗时很长的任务    time.sleep(10)
return a + b
@celery.task
def subtract(a, b):
# 模拟一个异常    A_Exception()
return a - b
先启动redis服务器，再启动Celery的worker
$ redis-server
$ pip3 install celery
$ pip3 install celery-with-redis
$ celery -A tasks worker --loglevel=info
启动新的shell，然后再运行Celery的Client
>>> from tasks import add
>>> from tasks import subtract
>>>
>>> result = subtract.delay(2, 1)
>>> result.ready() # 获取Worker 是否执行完任务
True
>>>
>>> result.get()  # 如果Worker执行出现异常，这样直接把Worker执行的异常输出出来
Traceback (most recent call last):
File "", line 1, in 
File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/result.py", line 162, in get
self.maybe_reraise()
File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/result.py", line 271, in maybe_reraise
raise self.result
NameError: name 'A_Exception' is not defined
>>>
>>> result.get(propagate=False) # 指定propagate 参数来覆盖抛异常行为
NameError("name 'A_Exception' is not defined",)
>>>
>>> result.traceback # 如果任务抛出了一个异常，你也可以获取原始的回溯信息
'Traceback (most recent call last):\n  File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/app/trace.py", line 240, in trace_task\n    R = retval = fun(*args, **kwargs)\n  File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/app/trace.py", line 438, in __protected_call__\n    return self.run(*args, **kwargs)\n  File "/Users/xumenger/Desktop/code/Python/tasks.py", line 16, in subtract\n    A_Exception()\nNameError: name \'A_Exception\' is not defined\n'
>>>
>>>
>>>
>>> result = add.delay(4, 4)
>>> result.ready()
False
>>> result.get(timeout=1)
Traceback (most recent call last):
File "", line 1, in 
File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/result.py", line 169, in get
no_ack=no_ack,
File "/Users/xumenger/Desktop/code/Laboratory3/_LAB3/lib/python3.6/site-packages/celery/backends/base.py", line 238, in wait_for
raise TimeoutError('The operation timed out.')
celery.exceptions.TimeoutError: The operation timed out.
>>>
>>> result.get() # 程序一直【阻塞】，直到等到Worker执行完获取结果！
8
>>>
>>> result.backend # 获取结果后端信息
在Celery的worker运行环境中有这样的输出
针对以上的小例子的运行效果，其实可以对应的提出这些问题：
Worker是怎么从Broker中获取信息的，拉？推(订阅)？
Client、Broker、Worker之间传递了什么信息、数据的格式是什么样的？
…………
另外，Celery有很多优点
高可用性：倘若连接丢失或失败，职程和客户端会自动重试，并且一些中间人通过 主/主 或 主/从 方式复制来提高可用性
快速：单个 Celery进程每分钟可处理数以百万计的任务，而保持往返延迟在亚毫秒级(使用 RabbitMQ、py-librabbitmq 和优化过的设置)
灵活：Celery几乎所有部分都可以扩展或单独使用。可以自制连接池、 序列化、压缩模式、日志、调度器、消费者、生产者、自动扩展、 中间人传输或更多
Client和Worker共用代码？
上面的例子看到Worker 和Client 都共用了tasks.py 代码，为什么呢？或者说这是必须的吗？那么就实验看一下
假设现在tasks.py 的路径是/Users/xumenger/Python。Worker还是不变的逻辑，还是用tasks.py 代码启动
celery -A tasks worker --loglevel=info
控制Worker方面的变量不变，改变Client 的变量来进行测试
情况一：使用同目录下的不同文件
比如我在相同的目录下创建一个demo.py 文件，将tasks.py的 代码完全拷贝过去，然后Client 端的运行效果是这样的(直接在获取结果的地方阻塞住了)
再去看Worker的运行情况
值得注意看一下传递的消息的内容是这样的格式(下面展示的是格式化之后的效果)
{
'task': 'demo.add',
'id': '434fcc2e-4084-4595-8c5c-8ac6f4c7ef3e',
'args': (2, 3),
'kwargs': {},
'retries': 0,
'eta': None,
'expires': None,
'utc': True,
'callbacks': None,
'errbacks': None,
'timelimit': (None, None),
'taskset': None,
'chord': None
}
情况二：不同目录，文件名相同，代码除了接口相同外，都不同
在/Users/xumenger/Python 下创建目录dir，在dir 下面创建tasks.py，编码内容为
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0')
# 函数只有接口与原来一致，代码逻辑直接删除了@app.task
def add(a, b):
pass
然后这次Client 可以成功运行
大概谈一下我的理解！
Celery 是异步调用任务的框架，更是远程调用的框架，远程调用的重点是什么
Client 和Worker 的接口要约定一致(约定好函数名称、一样的函数参数类型、一样返回值类型)
代码逻辑是在Worker 端执行，所以Client 端的代码逻辑是非必要的！
其他内容
最开始展示的例子之所以Client 和Worker 共用一份代码，其实也是有好处的，这样接口就一定是一样的，当然你也可以分开写，只要像上面的例子中保证接口的一致性，但是说实话，既然是人来开发的，难免会出现写错的情况，比如函数名写错了等等，反而容易招致不必要的麻烦
Celery配置中心化
Celery 提供了一些配置项供开发者使用
# 配置序列化任务载荷的默认的序列化方式celery.conf.CELERY_TASK_SERIALIZER = 'json'
# 一次性设置多个选项celery.conf.update(
CELERY_TASK_SERIALIZER='json',
CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content    CELERY_RESULT_SERIALIZER='json',
CELERY_TIMEZONE='Europe/Oslo',
CELERY_ENABLE_UTC=True,
)
对于大型项目，采用独立配置模块更为有效，事实上你会为硬编码周期任务间隔和任务路由等配置项而感到沮丧，因为中心化保存配置更为合适
尤其是对于库而言，这使得用户控制任务行为成为可能，你也可以想象系统管理员在与搭配系统故障时对配置做出简单修改
可以调用config_from_object()来让 Celery 实例加载配置模块：
celery.config_from_object('celeryconfig')
配置模块名字为celeryconfig.py，必须可以从当前目录或Python路径加载。内容格式为：
BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'amqp://'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'Europe/Oslo'
CELERY_ENABLE_UTC = True
生产环境中，可以在一台配置中心的服务器上为Celery、Nginx等所有的组件维护它们所有的配置信息，在Celery、Nginx工作的机器上专门开一个监控进程去定时从配置中心获取最新的配置信息更新到本地配置文件，如果发现配置文件更新了，那么监控进程就去重启对应的Celery、Nginx等进程。当然这是一种比较直观、粗暴的实现方式
将配置集中到一个地方是很好的解耦方式，也很强大，比如下面这个例子展示了如何把“脏活”路由到专门的队列
CELERY_ROUTES = {
'tasks.add': 'add_queue',
'tasks.subtract': 'subtract_queue'
}
或者你可以限制任务的速率，这样每分钟只允许处理10个该类型的任务
CELERY_ANNOTATIONS = {
'tasks.add': {'rate_limit': '10/m'}
}

{"moduleinfo":{"title":"精心打造的功能"},"optioninfo":{"dynamic":"true","static":"true"},"li":[{"spm":"","img":"https://img.alicdn.com/tfs/TB1jdmyXAL0gK0jSZFxXXXWHVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1jdmyXAL0gK0jSZFxXXXWHVXa-108-108.png","h1":"定时调度","p":"支持 cron、fixed_rate、second_delay 定时调度，能精确到秒级别。","more":[]},{"h1":"可视化任务编排","p":"通过可视化的工作流，拖拖拽拽即可进行任务依赖关系编排。","img":"https://img.alicdn.com/tfs/TB1hhixXxz1gK0jSZSgXXavwpXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1hhixXxz1gK0jSZSgXXavwpXa-108-108.png"},{"h1":" 多种任务类型","p":"支持 Java、Shell、Python、Go 和 HTTP 等多种任务。","img":"https://img.alicdn.com/tfs/TB1aHaxXEz1gK0jSZLeXXb9kVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1aHaxXEz1gK0jSZLeXXb9kVXa-108-108.png"},{"h1":"分布式计算","p":"提供单机、广播、Map、MapReduce 和分片等多种分布式编程模型，简单几行代码即可通过您自己的机器自建分布式引擎，进行大数据跑批。","img":"https://img.alicdn.com/tfs/TB13TuxXrY1gK0jSZTEXXXDQVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB13TuxXrY1gK0jSZTEXXXDQVXa-108-108.png"},{"h1":"报警监控","p":"全面的监控指标，丰富及时的报警方式，便于运维人员快速定位和解决线上问题。","img":"https://img.alicdn.com/tfs/TB1qIeyXAH0gK0jSZFNXXXMqXXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1qIeyXAH0gK0jSZFNXXXMqXXa-108-108.png"}]}
{"$env":{"JSON":{}},"$page":{"env":"production"},"$context":{"moduleinfo":{"title":"精心打造的功能"},"optioninfo":{"dynamic":"true","static":"true"},"li":[{"spm":"","img":"https://img.alicdn.com/tfs/TB1jdmyXAL0gK0jSZFxXXXWHVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1jdmyXAL0gK0jSZFxXXXWHVXa-108-108.png","h1":"定时调度","p":"支持 cron、fixed_rate、second_delay 定时调度，能精确到秒级别。","more":[]},{"h1":"可视化任务编排","p":"通过可视化的工作流，拖拖拽拽即可进行任务依赖关系编排。","img":"https://img.alicdn.com/tfs/TB1hhixXxz1gK0jSZSgXXavwpXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1hhixXxz1gK0jSZSgXXavwpXa-108-108.png"},{"h1":" 多种任务类型","p":"支持 Java、Shell、Python、Go 和 HTTP 等多种任务。","img":"https://img.alicdn.com/tfs/TB1aHaxXEz1gK0jSZLeXXb9kVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1aHaxXEz1gK0jSZLeXXb9kVXa-108-108.png"},{"h1":"分布式计算","p":"提供单机、广播、Map、MapReduce 和分片等多种分布式编程模型，简单几行代码即可通过您自己的机器自建分布式引擎，进行大数据跑批。","img":"https://img.alicdn.com/tfs/TB13TuxXrY1gK0jSZTEXXXDQVXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB13TuxXrY1gK0jSZTEXXXDQVXa-108-108.png"},{"h1":"报警监控","p":"全面的监控指标，丰富及时的报警方式，便于运维人员快速定位和解决线上问题。","img":"https://img.alicdn.com/tfs/TB1qIeyXAH0gK0jSZFNXXXMqXXa-108-108.png","imgAction":"https://img.alicdn.com/tfs/TB1qIeyXAH0gK0jSZFNXXXMqXXa-108-108.png"}]}}
精心打造的功能
定时调度
支持 cron、fixed_rate、second_delay 定时调度，能精确到秒级别。
可视化任务编排
通过可视化的工作流，拖拖拽拽即可进行任务依赖关系编排。
多种任务类型
支持 Java、Shell、Python、Go 和 HTTP 等多种任务。
分布式计算
提供单机、广播、Map、MapReduce 和分片等多种分布式编程模型，简单几行代码即可通过您自己的机器自建分布式引擎，进行大数据跑批。
报警监控
全面的监控指标，丰富及时的报警方式，便于运维人员快速定位和解决线上问题。