[ celery重试机制 ]


python中的celery使用

开启重试需要添加 bind=True参数 才能绑定到self来进行重试

@celery_app.task(bind=True, max_retries=3, default_retry_delay=5)
def send_email_when_check_eform(self, recipient_email, subject, message):
    password = 'G@9742ns'
    server_email = 'no-reply@gnstec.com.hk'
    from_addr = formataddr(("CEWS Reminder", server_email))
    to_addr = recipient_email

    msg = MIMEText(message, 'html')
    msg['From'] = from_addr
    msg['To'] = to_addr
    msg['Subject'] = subject

    smtp_server = 'smtp.office365.com'  
    smtp_port = 587  
    server = None

    for attempt in range(3):  # 最大尝试次数为 3 次
        try:
            server = smtplib.SMTP(smtp_server, smtp_port)
            server.starttls()
            server.ehlo()
            server.login(server_email, password)
            server.sendmail(from_addr, to_addr, msg.as_string())
            print('Email sent successfully')
            break
        except Exception as e:
            print('Email sending failed', str(e))
            if attempt < 2:  
                time.sleep(1)  
            else:
                self.retry(exc=e)
        finally:
            if server:
                server.quit()