四、在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', ])