Celery实现分布式定时工作并开启监控(Celery-Beat、Celery-Once、flower)

原理：celery-beat作为任务调度，当达到定时工夫时，beat将工作id装载进rabbitmq队列中，worker在队列的另一端取出工作id，并匹配以后注册的工作。如果没有注册，那么会报错。除此之外，worker还会通过celery-once来尝试从redis中获取分布式锁，只有获取到锁的worker才会执行这个工作。worker执行胜利或者失败通过flower监控

 

1.环境筹备

pip install celerypip install flowerpip install celery_once

 

2.代码构造

代码构造如下：

工作函数sendDingTest.py（钉钉机器人）：

from dingtalkchatbot.chatbot import DingtalkChatbotfrom start import celery_appfrom datetime import datetimeimport socketfrom celery_once import QueueOnceWEB_HOOK_SPIDER = '在钉钉机器人页面获取web_hook'# 基于工作名及传递的参数值来确认是否是同一个工作@celery_app.task(base=QueueOnce, once={'graceful': True})def send_ding_test(arg1, arg2):    dingding = DingtalkChatbot(WEB_HOOK_SPIDER)    arg3 = arg1 + arg2    dingding.send_text(msg="城市数据定时工作测试-{},{} 执行时刻：{}".format(arg3, socket.gethostname(), datetime.now().strftime("%Y-%m-%d %H:%M:%S")))

配置文件config.py：

from celery.schedules import crontabclass celeryConfig(object):    # accept_content = ['json']  # 能够是set,list,tuple,pickle,yaml    # result_accept_content = ['json']    timezone = 'Asia/Shanghai'  # 中国只有两个时区，一个上海，一个乌鲁木齐    broker_url = "amqp://user:password@ip:port/vhost"    backend = ""    include = ['jobs.sendDingTest']  # worker启动时要导入的工作模块，须要在这里增加，以便worker可能找到咱们的工作    beat_schedule = {        'add-every-monday-morning':            {                'task': 'jobs.sendDingTest.send_ding_test', # 这里要写全门路，否则worker找不到                'schedule': crontab(minute="*/2"),            'args': (16, 16),            },    }    # celery-once配置    ONCE = {        'backend': 'celery_once.backends.Redis',        'settings': {            'url': 'redis://ip:port/database',            'default_timeout': 60 * 60 # 分布式锁的默认超时工夫        }    }

启动函数start.py：

from celery import Celeryimport os# windows平台须要设置，os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')celery_app = Celery()celery_app.config_from_object('config.celeryConfig')

 

3.启动定时工作与监控

# 留神：beat,worker,flower都能够不在同一台服务器上，散布开的话，须要在其余服务器上copy一份代码# 开启beat（start是文件名）celery -A start.celery_app beat# 开启workercelery -A start.celery_app worker -c 1 -l info# 开启flower（默认地址是localhost:5555）celery -A start flower

以下是flower页面，能够查看worker数量、音讯队列（须要rabbitmq开启rabbitmq_management）和工作执行后果等。

4.文档参考

celery-once文档

flower文档

celery官网文档