爱站程序员基地- 本实验通过Django 异步发送邮件作为示例
a.Celery实际上是生产者消费者模型来实现异步的:
b.实验所需组件：
1) Django
2) Celery
3) Redis
2.实验代码
a.启动redis-server
#./redis-server /etc/redis/redis.conf#cat /etc/redis/redis.confbind 192.168.60.72port 6379daemonize yesdbfilename dump.rdbdir /var/lib/redislogfile \"/var/log/redis/redis-server.log\"#ss -antlp | grep 6379LISTEN 0 128 192.168.60.72:6379 *:* users:((\"redis-server\",pid=77880,fd=6))
b.配置Django
#django-admin startproject send_mail_prj#cd send_mail_prj#django-admin startapp app01创建Celery相关文件:├─manage.py│├─app01│ │ admin.py│ │ apps.py│ │ models.py│ │ tests.py│ │ urls.py│ │ utils.py│ │ views.py│ ├─__init__.py│├─celery_tasks //创建如下文件│ │ config.py│ │ main.py│ │ __init__.py│ ││ ├─sms│ │ tasks.py│ └─__init__.py││└─send_mail_prj│ asgi.py│ settings.py│ urls.py│ wsgi.py│─ __init__.py
#celery_tasks.init.py为空
#celery_tasks.config.py
# Celery 配置文件broker_url = \'redis://192.168.60.72/10\'
#celery_tasks.main.py
# celery入口from celery import Celery# 初始化celery实例celery_app = Celery(\'send_mail\')# 加载celery配置celery_app.config_from_object(\'celery_tasks.config\')# 注册任务celery_app.autodiscover_tasks([\'celery_tasks.sms\'])
#celery_tasks.sms.init.py 为空
#celery_tasks.sms.tasks.py
from celery_tasks.main import celery_appfrom email.mime.text import MIMETextimport smtplib# 定义异步任务@celery_app.task(name=\'send_mail\')def sendEmail():# 设置服务器所需信息# 163邮箱服务器地址mail_host = \'smtp.163.com\'# 163用户名mail_user = \'yxxxx5\'# 密码(部分邮箱为授权码)mail_pass = \'KIMxxxxxxxxxxxZ\'# 邮件发送方邮箱地址sender = \'[email protected]\'# 邮件接受方邮箱地址，注意需要[]包裹，这意味着你可以写多个邮件地址群发receivers = [\'[email protected]\']# 设置email信息# 邮件内容设置message = MIMEText(\'dear frank, sdfsdfsdfasfdsdfsfwetrwre\', \'plain\', \'utf-8\')# 邮件主题message[\'Subject\'] = \'hi fxxxg sir sdfsfsdfasf\'# 发送方信息message[\'From\'] = sender# 接受方信息message[\'To\'] = receivers[0]# 登录并发送邮件try:smtpObj = smtplib.SMTP()# 连接到服务器smtpObj.connect(mail_host, 25)# 登录到服务器smtpObj.login(mail_user, mail_pass)# 发送smtpObj.sendmail(sender, receivers, message.as_string())# 退出smtpObj.quit()return (\'success\')except smtplib.SMTPException as e:print(\'error\', e) # 打印错误
3.Django写视图函数和配置URL
#app01.views.py
from django.http import HttpRequest, HttpResponse# 导入异步任务from celery_tasks.sms.tasks import sendEmail# Create your views here.def register(request):# 发送邮件sendEmail.delay()return HttpResponse(\'hello\')
#app01.urls.py
from django.conf.urls import urlfrom . import viewsurlpatterns = [url(r\'register/$\', views.register, name=\'register\'),]
#send_mail_prj.urls.py
from django.contrib import adminfrom django.urls import pathfrom django.conf.urls import url, includeurlpatterns = [path(\'admin/\', admin.site.urls),url(r\'^\', include(\'app01.urls\')),]
4.启动Celery
#cd send_mail_prj
#celery -A celery_tasks.main worker -l info
#celery -A celery_tasks.main worker -l info -P eventlet //windows下面需要添加-P eventlet
5.启动django
#cd send_mail_prj
#python manage.py runserver 0.0.0.0:80
6.访问127.0.0.1/register/
实验完成。
