使用django + celery + redis 异步发送邮件

2018/03/06 16:26
阅读数 27

参考:http://blog.csdn.net/Ricky110/article/details/77205291 

 

 

环境:

centos7  +  python3.6.1 + django2.0.1  +  celery4.1.0  +  redis3.2.10

 

yum  install -y redis

pip3 install redis,celery,django

开始:

创建django工程my_report

创建app celery_test, 如下所示 :

  • INSTALLED_APPS中注册app_celery

  • setting中celery配置

  • # Celery settings
    CELERY_BROKER_URL = 'redis://localhost:6379'
    #: Only add pickle to this list if your broker is secured
    CELERY_ACCEPT_CONTENT = ['json']
    CELERY_RESULT_BACKEND = 'redis://localhost:6379'
    CELERY_TASK_SERIALIZER = 'json'
    CELERY_ENABLE_UTC = True
    CELERY_TIMEZONE = 'Asia/Shanghai'

      

  • setting中mail配置
  • EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
    EMAIL_HOST = "smtp.mail.haoyisheng.com"
    EMAIL_HOST_PASSWORD = '******'
    DEFAULT_FROM_EMAIL = EMAIL_HOST_USER = "lijianwei@mail.haoyisheng.com"
    EMAIL_PORT = 25
    EMAIL_USE_TLS = True

      

  • app所在目录添加tasks.py文件(必须是该文件名), 用于处理任务
  • from __future__ import absolute_import, unicode_literals
    from celery import shared_task
    from django.core.mail import send_mail
    import logging
    
    logger = logging.getLogger(__name__)
    
    
    @shared_task
    def celery_send_email(subject, message, from_email, recipient_list, **kwrags):
        try:
            # 使用celery并发处理邮件发送的任务
            logger.info("\n开始发送邮件")
            send_mail(subject, message, from_email, recipient_list, **kwrags)
            logger.info("邮件发送成功")
            return 'success!'
        except Exception as e:
            logger.error("邮件发送失败: {}".format(e))

      

  • 配置目录my_report中添加celery.py文件
  • from __future__ import absolute_import, unicode_literals
    import os
    from celery import Celery
    
    
    # 为celery程序设置DJANGO_SETTINGS_MODULE环境变量
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_report.settings')
    
    app = Celery('celery_test')
    
    # 从Django的设置文件中导入CELERY设置
    app.config_from_object('django.conf:settings', namespace='CELERY')
    # 从所有已注册的app中加载任务模块
    app.autodiscover_tasks()
    
    
    @app.task(bind=True)
    def debug_task(self):
        print('Request: {0!r}'.format(self.request))

      

  • 配置目录my_report的__init__.py文件中添加如下
  • from __future__ import absolute_import, unicode_literals
    
    # 这将保证celery app总能在django应用启动时启动
    from .celery import app as celery_app
    __all__ = ['celery_app']

      

编写url映射和视图

from django.contrib import admin
from django.urls import path
from month_report import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('send_email/', views.add_task_to_celery, name='send_email'),
]

  

编写views:

from django.http import HttpResponse
from celery_test.tasks import celery_send_email

def add_task_to_celery(request):
celery_send_email.delay(u'邮件主题', 'test_mail_message', 'lijianwei@mail.haoyisheng.com',
['lijianwei@mail.haoyisheng.com'])
return HttpResponse('hello world')

  

  • 在manage.py同级目录执行如下命令, 启动celery的worker进程(主要用于消费或执行任务)

celery -A my_report  worker --loglevel=info

执行成功:

 

 

  • 从客户端请求
  • http://ip:port/send_email/
  • 然后会收到页面返回hello world, 并且终端显示时间处理结果为成功

然后验证邮箱有收到邮件,成功

温馨提示

  • 当前使用方法,如需要在tasks.py中新添加任务,新增后,则需要重启django, 并且需要从起celery worker进程, worker进程默认不能动态加载事件。
展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部