# 定时任务在文件中的写法
# -*- coding:utf-8 -*-
from Celery import app
from celery.schedules import crontab
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)# 通过beat组件周期性将任务发送给woker执行。在示例中 新建文件period_task.py 并添加任务到配置文件中
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):sender.add_periodic_task(10.0, add.s(1,3), name='1+3=') # 每10秒执行addsender.add_periodic_task(crontab(hour=16, minute=56, day_of_week=1),      #每周一下午四点五十六执行sayhaisayhi.s('wd'),name='say_hi')

celery -A haha worker -B -l info

CELERY_QUEUES = (
Queue("default",Exchange("default"),routing_key="default"),
Queue("for_add",Exchange("for_add"),routing_key="for_add"),
Queue("for_max",Exchange("for_max"),routing_key="for_max")
)
# 路由
CELERY_ROUTES = {'tasks.add':{"queue":"for_add","routing_key":"for_add"},
'tasks.max':{"queue":"for_max","routing_key":"for_max"}
}# 路由指明队列和routing_key
# routing_key认证所在的队列 用redis名称要一致（对于redis来说）

# coding=utf-8
from Celery import app
from celery.schedules import crontab
from celery.utils.log import get_task_logger
from celery import Task
logger = get_task_logger(__name__)class demotask(Task): # 这是三种运行的状态def on_success(self, retval, task_id, args, kwargs):   # 任务成功执行logger.info('task id:{} , arg:{} , successful !'.format(task_id,args))def on_failure(self, exc, task_id, args, kwargs, einfo):  #任务失败执行logger.info('task id:{} , arg:{} , failed ! error : {}' .format(task_id,args,exc))def on_retry(self, exc, task_id, args, kwargs, einfo):    #任务重试执行logger.info('task id:{} , arg:{} , retry !  info: {}'.format(task_id, args, exc))@app.task(base=demotask) # 需要继承这个基类
def func1(x,y):try:a=[]a[10] =1except Exception as e:logger.info(e)return x+y@app.task(base=demotask)
def func2(name):a=[]a[10] =1return 'hi {}'.format(name)@app.task(base=demotask)
def func3(a,b):return 'result = {} '.format(a+b)

@app.task
@decorator2
@decorator1
def add(x, y):return x + y

from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)@app.task
def add(x, y):logger.info('Adding {0} + {1}'.format(x, y))return x + y

pip install flower

celery flower --broker=redis://localhost:6379 --address=127.0.0.1 --port=5555或者flower -A proj --broker=redis://localhost:6379

# config.py
from __future__ import absolute_import
# broker
import datetime
BROKER_URL = 'redis://127.0.0.1:6379/0'
# backen
CELERY_RESULT_BACKEND = 'mongodb://127.0.0.1:27017'
# 导入任务，如tasks.py
CELERY_IMPORTS = ('task', )
# 列化任务载荷的默认的序列化方式
CELERY_TASK_SERIALIZER = 'json'
# 结果序列化方式
CELERY_RESULT_SERIALIZER = 'json'CELERY_ACCEPT_CONTENT = ['json']CELERY_TIMEZONE='Asia/Shanghai'    # 指定时区，不指定默认为 'UTC'
# CELERY_TIMEZONE='UTC'
# CELERY_ENABLE_UTC = True

# Celery.py
# coding=utf-8
from __future__ import absolute_import
from celery import Celeryapp = Celery("STQ")
# 加载配置模块
app.config_from_object('config')if __name__ == '__main__':app.start()

# coding=utf-8
from __future__ import absolute_import
import re
import requests
from lxml import etree
import urllib.request
from Celery import appheaders ={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36"}def get_html(url):r = urllib.request.Request(url,headers=headers)res = urllib.request.urlopen(r)return res.read().decode("utf-8")@app.task
def parse(html):ret = re.compile(r'<title>(.*?)</title>')title = re.findall(ret,html)return title

from __future__ import absolute_import
from tasks import get_html,parsewith open("url.txt","r") as f:for data in f.readlines():url = data.strip('n')html = get_html(url)result = parse.delay(html)

celery -A Celery worker -l info -P gevent -c 10

python run_task.py

CELERY_QUEUES = (
Queue("default",Exchange("default"),routing_key="default"),
Queue("for_add",Exchange("for_add"),routing_key="for_add"),
)

# 路由
CELERY_ROUTES = {'tasks.add':{"queue":"for_add","routing_key":"for_add"},
}

celery -A Celery worker -l info -n add -Q for_add -P gevent -c 10

 redis://ip:6379/0

celery -A Celery worker -l info

python run_haha.py