ubuntu服务器下使用 supervisor 和 celery
supervisor 的卸载过程:
sudo apt purge supervisor whereis supervisord
如果有用 pip 安装的,用pip uninstall supervisor 再卸载一遍
查找 supervisord 在哪,然后删除所有信息
root@jdu4e00u53f7:~# whereis supervisord supervisord: /usr/local/bin/supervisord rm -rf /usr/local/bin/supervisord
重新安装过程:
安装: pip3 install supervisor 或者 apt-get install supervisor
新建文件: mkdir /etc/supervisor
运行代码: echo_supervisord_conf > /etc/supervisor/supervisord.conf
目录 etc/supervisor 下有主配置文件 supervisord.conf 和 conf.d 文件夹。可以直接使用 vim 编辑,在 supervisord.conf 文件底部插入执行的内容;
也可以在 conf.d 文件夹里新建 filename.conf 的文件,把执行的内容写在该文件下同样能被加载到,不过需要在 supervisord.conf 的配置底部,添加 files = /etc/supervisor/conf.d/*.conf ,也就是包含加载该路径下的配置文件。
1 [program:workername] # 进程的名字,随意起 2 command=celery -A celery_name worker -l info # celery运行的命令 3 directory=/mnt/your_project_dir # celery任务文件所在的目录; 4 user=root 5 numprocs=1 6 # 设置log的路径 7 stdout_logfile=/var/log/supervisor/mailboxworker.log 8 stderr_logfile=/var/log/supervisor/mailboxworker.log 9 autostart=true 10 autorestart=true 11 startsecs=10 12 stopwaitsecs = 600 13 priority=15
然后运行命令 supervisord 启动。
查看状态: supervisorctl tail celery_name
停止: supervisorctl stop celery_name
启动: supervisorctl start celery_name
重启: supervisorctl restart celery_name
1 supervisorctl stop all 停止全部进程 2 supervisorctl reload 载入最新的配置文件,停止原有进程并按新的配置启动、管理所有进程 3 supervisorctl update 根据最新的配置文件,启动新配置或有改动的进程,配置没有改动的进程不会受影响而重启
也可到 /var/log/supervisor/ 目录下查看日志文件。可根据内容查看 celery 是否正常运行。
遇到的报错问题:
1、执行 supervisord 运行时报错:
-bash: /usr/bin/supervisord: No such file or directory supervisord
把 superviosr 卸载重装就好了。
2、运行 supervisorctl status 报错:
一开始用的 pip 安装,运行报以下错误,然后卸载后改用 apt-get 安装解决;但报了第二个错误,然后在不卸载的情况下再次用 pip 安装了一遍就好了。。。
unix:///tmp/supervisor.sock no such file
-bash: /usr/local/bin/supervisorctl: No such file or directory
3、运行状态发现报错,提示推出太多次失败等其它问题 BACKOFF 或者 FATAL 状态, Exited too quickly (process log may have details)
改 celery 的配置 startsecs=0 ,然后重新运行 supervisorctl reload 加载配置解决。参考路径
4、如果有其它错误,根据配置中写的日志报错位置,查看报错日志,一般报错位置: /var/log/supervisor
网上其他的一些 celery 配置参考:
1 [program:update_ip] ;项目名称 2 directory = /home/xxxx/works/ip_update/ip_update_on_server_no_1/ ; 程序的启动目录 3 command = python /home/xxxx/works/ip_update/ip_update_on_server_no_1/update_ip_internal.py ; 启动命令,可以看出与手动在命令行启动的命令是一样 4 autostart = true ; 在 supervisord 启动的时候也自动启动 5 startsecs = 5 ; 启动 5 秒后没有异常退出,就当作已经正常启动了 6 autorestart = true ; 程序异常退出后自动重启 7 startretries = 3 ; 启动失败自动重试次数,默认是 3 8 user = shimeng ; 用哪个用户启动 9 redirect_stderr = true ; 把 stderr 重定向到 stdout,默认 false 10 stdout_logfile_maxbytes = 50MB ; stdout 日志文件大小,默认 50MB 11 stdout_logfile_backups = 20 ; stdout 日志文件备份数 12 ; stdout 日志文件,需要注意当指定目录不存在时无法正常启动,所以需要手动创建目录(supervisord 会自动创建日志文件) 13 stdout_logfile = /home/xxxx/works/ip_update/ip_update_on_server_no_1/supervisor.log 14 loglevel=info 15 16 [supervisorctl] 17 serverurl=unix:///tmp/supervisor.sock ; use a unix:// URL for a unix socket 18 19 [unix_http_server] 20 file=/tmp/supervisor.sock ; (the path to the socket file) 21 chmod=0777 ; socket file mode (default 0700) 22 ;chown=nobody:nogroup ; socket file uid:gid owner 23 ;username=shimeng ; (default is no username (open server)) 24 ;password=123 ; (default is no password (open server)) 25 26 [inet_http_server] ; inet (TCP) server disabled by default 27 port=127.0.0.1:9001 ; (ip_address:port specifier, *:port for all iface) 28 username=shimeng ; (default is no username (open server)) 29 password=123
1 配置详解: 2 a) 在supervisord.conf文件中,分号“;”后面的内容表示注释 3 b) [group:组名],设置一个服务分组,programs后面跟组内所有服务的名字,以分号分格。 4 c) [program:服务名],下面是这个服务的具体设置: 5 Command:启用Tornado服务文件的命令,也就是我们手动启动的命令。 6 Directory:服务文件所在的目录 7 User:启用服务的用户 8 Autorestart:是否自动重启服务 9 stdout_logfile:服务的产生的日起文件 10 loglevel:日志级别
以下内容无关
个人项目内容,resource/send_email.py
1 from error_handler import error_handler 2 from util import id_generator 3 from wrapper import universal_resource_wrapper 4 from model import EmailSmtpModel 5 from celery_tasks import send_task 6 7 8 class SendEmail(Resource): 9 @universal_resource_wrapper(required=['title', 'content', 'receiverEmails', 'sendEmail'], 10 optional=['annexPathList', 'draftId']) 11 @cross_origin(allow_headers=['Content-Type']) 12 def post(self): 13 data = request.get_json() 14 smtp_data = EmailSmtpModel.find_by_email(data['sendEmail']) 15 if len(data['receiverEmails']) > 100: 16 return error_handler('单次群发邮箱数不要超过100个') 17 data['annexList'] = [] 18 if 'annexPathList' in data and data['annexPathList'] != '': 19 for k in data['annexPathList']: 20 data['annexList'].append({'annexName': k['fileName'], 'annexPath': k['filePath'], 21 'annexUrl': k['fileUrl']}) 22 del data['annexPathList'] 23 if len(data['title']) <= 300: 24 title_template = data['title'] 25 else: 26 raise ValueError('邮件标题太长,应小于等于300字符!') 27 if len(data['content']) <= 100000: 28 body_template = data['content'] 29 else: 30 raise ValueError('邮件内容太长,应小于等于100000字!') 31 # 开启异步任务 32 send_task.apply_async( 33 args=[smtp_data['clientId'], smtp_data.sendEmail, smtp_data.smtpServer, smtp_data.smtpPort, 34 smtp_data.authorizationCode, smtp_data.sendUsername, data, title_template, body_template]) 35 # task.wait() 36 resp = jsonify({ 37 'msg': '邮件默认后台发送', 38 'status': True 39 }) 40 # return jsonify({}), 202, {'Location': url_for('gettask', task_id=task.id)} 41 return resp
celery_tasks/celery.py 的内容:
1 import time 2 from celery import Celery 3 from flask import Flask 4 from config import MONGO_DB, MONGO_URI, REDIS_SERVER_IP, REDIS_SERVER_PORT 5 from model import UserModel, send_email_with_custom_smtp, DraftBoxMailModel, EmailContentModel, db 6 from util import id_generator 7 8 9 def make_celery(app): 10 celery = Celery( 11 app.import_name, 12 backend=app.config['CELERY_RESULT_BACKEND'], 13 broker=app.config['CELERY_BROKER_URL'] 14 ) 15 # celery.conf.update(app.config) 16 17 class ContextTask(celery.Task): 18 def __call__(self, *args, **kwargs): 19 with app.app_context(): 20 return self.run(*args, **kwargs) 21 22 celery.Task = ContextTask 23 return celery 24 25 26 current_app = Flask(__name__) 27 28 current_app.config.update( 29 CELERY_BROKER_URL='redis://{}:{}'.format(REDIS_SERVER_IP, REDIS_SERVER_PORT), 30 CELERY_RESULT_BACKEND='redis://{}:{}'.format(REDIS_SERVER_IP, REDIS_SERVER_PORT) 31 ) 32 celery = make_celery(current_app) 33 current_app.config['MONGODB_SETTINGS'] = [ 34 { 35 'alias': 'MAILDB', 36 'db': MONGO_DB, 37 'host': MONGO_URI, 38 "connect": False 39 }, 40 { 41 'alias': 'TESTDB', 42 'db': MONGO_DB, 43 'host': MONGO_URI, 44 "connect": False 45 } 46 ] 47 db.init_app(current_app) 48 # # 异步任务 49 # @celery.task(bind=True) 50 # def long_task(self): 51 # total = 50 52 # for i in range(total): 53 # # 自定义状态 state 54 # self.update_state(state=u'处理中', meta={'current': i, 'total': total}) 55 # time.sleep(3) 56 # return {'current': 100, 'total': 100, 'result': u'完成'} 57 58 59 @celery.task 60 def send_task(clientId, sendEmail, smtpServer, smtpPort, authorizationCode, sendUsername, data, title_template, 61 body_template): 62 usr = UserModel.find_by_client_id(clientId) 63 from datetime import datetime 64 print('开始', datetime.now()) 65 for email_addr in data['receiverEmails']: 66 status, msg = send_email_with_custom_smtp( 67 sender_email=sendEmail, 68 receiver_email=email_addr, 69 server=smtpServer, 70 port=smtpPort, 71 usr=sendEmail, 72 password=authorizationCode, 73 subject=title_template, 74 text=body_template, 75 html=body_template, 76 annex_path=data['annexList'] 77 ) 78 if status: 79 if 'draftId' in data: 80 draft_mail = DraftBoxMailModel.find_by_draft_id(data['draftId']) 81 draft_mail.sentStatus = True 82 draft_mail.save() 83 else: 84 raise ValueError(msg) 85 time.sleep(60) 86 print('结束:',datetime.now()) 87 # 保存发送的记录 88 usr.send_records(fromEmail=sendEmail, toEmailList=data['receiverEmails'], title=title_template, 89 content=body_template, created_at=time.time(), 90 created_by=usr['uid'], status=True, annex_list=data['annexList']) 91 mail_data = DraftBoxMailModel.find_email(title_template, body_template, sendEmail, 92 data['receiverEmails'], data['annexList']) 93 if 'Reply:' in title_template: 94 email_data = EmailContentModel( 95 eid=id_generator(template='uuid'), 96 clientId=usr['clientId'], 97 subject=title_template, 98 content=body_template, 99 belongEmail=sendEmail, 100 fromEmail=sendEmail, 101 fromName=sendUsername, 102 emailTime=time.time(), 103 toEmail=data['receiverEmails'][0], 104 annexList=data['annexList'], 105 createdAt=time.time(), 106 createdBy=usr['uid'] 107 ) 108 email_data.save() 109 if mail_data is None: 110 draft_mail_data = DraftBoxMailModel( 111 clientId=usr['clientId'], 112 draftId=id_generator('uuid'), 113 subject=title_template, 114 content=body_template, 115 belongEmail=sendEmail, 116 fromEmail=sendEmail, 117 fromName=sendUsername, 118 toEmailList=data['receiverEmails'], 119 createdAt=time.time(), 120 createdBy=usr['uid'], 121 sentStatus=True, 122 annexList=data['annexList'] 123 ) 124 draft_mail_data.save()