四、在Django中使用Celery


返回

4.1 Django视图

  • worker

    from django.shortcuts import HttpResponse
    
    from myCelery.mail.tasks import send_mail
    from datetime import datetime, timedelta
    
    
    def test(request):
        # 异步任务
        # res_mail = send_mail.delay('email')
        # print(res_mail.id)
        # 定时任务
        local_time = datetime.now()
        utc_time = datetime.utcfromtimestamp(local_time.timestamp())
        time_delay = timedelta(seconds=10)
        task_time = utc_time + time_delay
        res_mail = send_mail.apply_async(args=['mail', ], eta=task_time)
        print(res_mail.id)
        return HttpResponse('ok')
    
    

4.2 Celery包

  • 启动命令:celery -A myCelery.main worker --loglevel=info

  • __init__.py

  • 任务目录mail

    __init__.py

    tasks.py

    import time
    from myCelery.main import cel
    
    
    @cel.task
    def send_mail(info):
        print('sending mail...')
        time.sleep(3)
        print(f'send {info} ok!')
        return 'ok'
    
    
  • 配置文件config.py

    broker_url = 'redis://127.0.0.1:6379/15'  # broker
    result_backend = 'redis://127.0.0.1:6379/14'  # store
    
    
  • 入口main.py

    import os
    from celery import Celery
    
    """
    启动命令:celery -A myCelery.main worker --loglevel=info
    """
    
    # celery实例对象
    cel = Celery('djCelery')
    
    # 导入Django配置文件
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')
    
    # 加载Celery配置,并生效
    cel.config_from_object('myCelery.config')
    
    # 加载任务,默认找到 tasks.py
    cel.autodiscover_tasks(['myCelery.mail', ])
    
    
返回