基于python实现的三方组件----Celery

一.基于python实现的三方组件----Celery

1.作用

用于异步周期任务的处理

2.Celery的组成

(1)任务 app

(2)记录任务的缓存（通常用redis或rabbitMQ）

    任务记录 -broker

    任务返回记录-backend

(3)Worker 员工

    主动执行任务

    主动反馈结果

3.celery简单实例

s1.py

from celery import Celery

import time

#使用redis连接url格式 ：redis://:password@hostname:port/db_number

my_task=Celery("my_task",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379") #括号中task代表你对任务在内部的称呼

@my_task.task

def my_func1(a,b):

    time.sleep(10)

    return a+b

@my_task.task

def my_func2():

    time.sleep(10)

    return 2

@my_task.task

def my_func3():

    time.sleep(10)

    return 3

命令行运行

Linux:Linux - celery worker -A s1 -l INFO

Windows:celery worker -A s1 -l INFO -P eventlet

#Windows下需要下载eventlet模块模块，否则celery4的版本不支持windows

#l：日志输出

#c:数量

s2.py

from s1 import my_func1,my_func2,my_func3

pid=my_func1.delay(10,20)

print(pid)

pid=my_func2.delay()

print(pid)

pid=my_func3.delay()

print(pid)

s3.py

from celery.result import AsyncResult

from s1 import my_task

#运行s2.py得到的pid

res=AsyncResult(id='2b36d20f-da07-42fe-b203-1e56fbaafd5e',app=my_task)

if res.successful():

    print(res.get())

else:

    print("任务正在进行中")

4.爬虫简单应用

在caiji.py中

from flask import Flask,request as requ,jsonify,render_template,send_file

import pymongo

import json

import time

import  urllib

import requests

import re

from urllib import request

import uuid

from celery import Celery

import time

#使用redis连接url格式 ：redis://:password@hostname:port/db_number

my_task=Celery("my_task",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379") #括号中task代表你对任务在内部的称呼

#获取各种分类的歌曲列表

@my_task.task

def getcontent():

        # content=requ.form.get("content")

        # print(content)

        headers={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.92 Safari/537.36"}

        url="https://www.ximalaya.com/ertong/ertongwenxue"

        request=urllib.request.Request(url,headers=headers)

        response=urllib.request.urlopen(request)

        response_text=response.read().decode("utf-8")

        title_id_list=re.findall('"album-title line-2 lg.+?" title="(.+?)" href="/ertong/(\d+?)/">',response_text)

        anthor_list=re.findall('"album-author.+?" title="(.+?)" href',response_text)

        response_list=[]

        i = 0

        for i in range(len(title_id_list)):

            response_dict={}

            response_dict={

                "title":title_id_list[i][0],

                "id":title_id_list[i][1],

                "author":anthor_list[i]

            }

            response_list.append(response_dict)

        # print("返回",response_list)

        return response_list

#获取music的二进制文件

@my_task.task

def getmusic(id):

    print(id)

    url="http://m.ximalaya.com/ertong/"+id+"/"

    response=requests.get(url)

    response.encoding="utf-8"

    path=re.findall('"isCopyright":.+?"src":"(.+?)","albumId"',response.text)[0]

    print("res",path)

    d_data = requests.get(path)

    get_str=str(uuid.uuid4())

    print(get_str)

    name="./music/"+get_str + ".mp3"

    with open(name,"wb") as f:

        f.write(d_data.content)

    return send_file(name)

# getcontent()

在results.py中

from caiji import getcontent,getmusic

res1=getcontent.delay()

print(res1)

for i in res1.get():

    res2 = getmusic.delay(i["id"])

    print(res2)

5.定时任务(十秒钟后执行函数)

在s4.py中

from celery import Celery

import time

my_task=Celery("task",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379") #括号中代表你对任务在内部的称呼



@my_task.task

def my_func1(a,b):

    return 1

在s5.py中

import datetime

import time

from s4 import my_func1



tp = time.time()

utc_time = datetime.datetime.utcfromtimestamp(tp)

add_time = datetime.timedelta(seconds=10)

utc_time = utc_time + add_time

res = my_func1.apply_async(args=(2,3),eta=utc_time)

print(res)

6.周期任务

task_one.py

from celery import Celery

import time



my_task=Celery("my_task",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379")

@my_task.task

def my_func1():

    time.sleep(10)

    return "十秒钟执行的"

task_two.py

import time

from task_one import my_task

@my_task.task

def my_func2():

    time.sleep(5)

    return "五秒钟执行的"

s6.py

from celery import Celery

from celery.schedules import crontab

celery_task = Celery("task",

                     broker="redis://127.0.0.1:6379",

                     backend="redis://127.0.0.1:6379",

                     include=["task_one","task_two"])

#我要要对beat任务生产做一个配置,这个配置的意思就是每10秒执行一次Celery_task.task_one任务参数是(10,10)

celery_task.conf.beat_schedule={

    "each10s_task":{

        "task":"task_one.my_func1",

        "schedule":10, # 每10秒钟执行一次

        # "args":(10,20)

    },

    "each5s_task": {

        "task": "task_two.my_func2",

        "schedule":5, # 每5秒

        # "args": (50, 60)

    },

}

# celery beat -A Celery_task

# celery worker -A Celery_task -l INFO -P eventlet

celery beat -A Celery_task

7.celery项目目录

在selery.py中

from celery import Celery

my_task = Celery("task",

                 broker="redis://127.0.0.1:6379",

                 backend="redis://127.0.0.1:6379",

                 include=["Celery_task.task_one","Celery_task.task_two"])

在task_one.py中

from Celery_task.celery import my_task

@my_task.task

def func1():

    return 1

@my_task.task

def func3():

    return 3

在task_two.py中

from Celery_task.celery import my_task

@my_task.task

def func2():

    return 2

celery worker -A Celery_task -l INFO -P eventlet

巴特西