python 异步定时任务

异步任务的搭建
flask与Django配置有差异，这里使用Django，

创建异步文件夹celery_task,和以下目录文件

__init__文件配置

from celery import Celery
from datetime import timedelta

import os
import django

# 调用项目环境
os.environ.setdefault("DJANGO_SETTINGS_MODULE","s1.settings")
django.setup()

# 异步任务

celery_app = Celery("mycelery")  # 创建一个异步任务
celery_app.config_from_object('celery_task.config')  # 读取config里的配置

# 定时任务

celery_app.autodiscover_tasks(['celery_task.times'])# 文件夹名.文件名

celery_app.conf.update(
    CELERYBEAT_SCHEDULE={
        'sum-task': {
            # 文件夹名.文件名.方法名
            'task': 'celery_task.times.my_con',
            # 多少秒执行一次（eg:5秒一执行）
            'schedule': timedelta(seconds=5),
            # 这是传入my_con方法中的参数
            'args': (5, 6)
        },

        'sum-task1': {
            'task': 'celery_task.times.my_print',
            'schedule': timedelta(seconds=3),
            'args': ()
        },
    }
)

config.py文件配置

# (注释部分仅为查询到的资料，未测试使用，仅供参考)


# 如果使用 redis 作为中间人——任务队列（消息队列）———作用：暂存任务的地方
# 需要这样配置:
# broker_url = 'redis://192.168.203.153:6379/3'
# 如果需要密码的话
# BROKER_URL='redis://:xxxxx@127.0.0.1:6379/2'

# 如果使用别的作为中间人, 例如使用 rabbitmq
# 则 rabbitmq 配置如下:
# broker_url= 'amqp://用户名:密码@ip地址:5672'

# 例如:
# meihao: 在rabbitq中创建的用户名, 注意: 远端链接时不能使用guest账户.
# 123456: 在rabbitq中用户名对应的密码
# ip部分: 指的是当前rabbitq所在的电脑ip
# 5672: 是规定的端口号
# broker_url = 'amqp://meihao:123456@172.16.238.128:5672'


broker_url = 'redis://127.0.0.1:6379/0'
# 结果队列的链接地址
result_backend = 'redis://127.0.0.1:6379/1'

# 时区
# CELERY_TIMEZONE = 'Asia/Shanghai'
timezone = 'Asia/Shanghai'

# 导入指定的任务模块
imports = (
    'celery_task.task'
)

task.py文件配置

# 自定义异步方法

import time
import random
import redis

from celery_task import celery_app
from ronglian_sms_sdk import SmsSDK
from rest_framework.response import Response
from django.core import mail

# 定义发送短信方法
accId="8aaf07087de13e4***********"
accToken="419de3ed6144***********"
appId ="8aaf07087de13************"

# @app.task 指定将这个函数的执行交给celery异步执行
@celery_app.task()
def send_mess(phone):
    '''生成短信验证码'''
    sdk = SmsSDK(accId, accToken, appId)
    tid = '1'  # 短信验证码模板
    # 验证码随机生成
    datas = (random.randint(000000, 999999),)
    print("生成的短信验证码>>>", datas)
    # 短信验证码的模板  手机号  验证码随机生成
    resp = sdk.sendMessage(tid, phone, datas)
    # print("验证码的内容>>>",resp)
    # 存入redis库中
    rdb = redis.Redis(host='localhost', port=6379, db=0)
    # setex (key,秒，验证码)
    if rdb.get('sms_%s' % phone):
        return Response({'msg': "短信已发送，不能重复发送", 'code': 500})
    rdb.setex("sms_%s" % phone, 300, datas[0])
    rdb.close()
    return resp


# 定义发送邮箱方法 
@celery_app.task(name='send_email1')
def send_email1(subject, message, from_email, to_email):
    time.sleep(10)
    mail.send_mail(subject, message, from_email, to_email)
    print('发邮件')

times.py文件配置

# 自定义定时任务，调用在__init__里配置，

#简单定义测试方法

from celery_task import celery_app
@celery_app.task()
def my_con(x,y):
    print(x,y)
    print('x和y相加')
    return x+y

@celery_app.task()
def my_print():    
    print('22222222')

调用命令

# 启动异步任务

# celery -A 任务名 worker -l info -P eventlet
# celery -A celery_task worker -l info -P eventlet

# 启动定时任务

# celery -A 任务名 beat -l info
# celery -A celery_task beat -l info

注意事项

# 注意：任务需要在终端启动后才可以使用
# 注意：要先启动异步任务才能启动定时任务

# 注意：在导入方法,并调用时，需要在方法名后加.delay,加上才是异步调用执行
# 例：调用发送手机短信验证码

# 调用发送手机短信验证码
from celery_task.task import send_message


phone = 13462336969
send_message.delay(phone)

python 异步定时任务

Python相关栏目本月热门文章