异步芹菜任务开销

Posted 2023-02-25

【中文标题】异步芹菜任务开销

【英文标题】：Async celery tasks overhead

【发布时间】：2021-03-08 02:36:49

【问题描述】：

我正在使用此代码为数千封电子邮件发送通知。似乎它在运行 for 循环之后在 Web 服务器（gunicorn）中产生了一些开销。发送电子邮件时一切正常。

users = User.query.all()

for user in users:
    send_email_celery(user.email, gettext(u'New email'), 'users/email/new_data', #plus some parameters)

所以我正在考虑使用密件抄送，但后来我意识到我不能，因为每个用户都有一个唯一的 uuid 退订并且密件抄送在邮件客户端中有一些限制。

那么，使用批量邮件处理此类操作的正确方法是什么？

@celery.task
def send_async_email_celery(msg):
    mail.send(msg)
 
def send_email_celery(to, subject, template, **kwargs):
    countdown = kwargs.get('countdown', 600)
    app = current_app._get_current_object()
    msg = Message(subject, sender=app.config['MAIL_SENDER'], recipients=[to])
    msg.html = render_template(template + '.html', **kwargs)
    send_async_email_celery.apply_async(args=[msg], countdown=countdown)

---

gunicorn --error-logfile gunicorn-error.log --timeout 600 --max-requests 500 --max-requests-jitter 50 --workers 5 app:app -b localhost:8080

---

编辑 1：使用 top 进行一些调试后，cpu 未满载。是在几秒钟内暂时停止 gunicorn 的其他东西。

编辑 2：根据 2ps 的答案进行更改后，现在循环很快，但在循环执行几秒钟后，Web 服务器仍被锁定。

编辑3：尝试将rabbitmq更改为reddis，同样的问题。忽略结果，同样的问题。

编辑 4：将用户循环移动到 @celery 路由后，问题仍然存在

编辑 5：问题已通过使用 gevent worker 解决。

【问题讨论】：

也将循环推迟到 celery，否则您会锁定线程以为每个用户创建 1 个 celery 任务，并且在此期间网络服务器无法处理任何请求。

@AnthonyPerot 循环后的服务器响应现在几乎是瞬时的。然而，几秒钟后，gunicorn 会停止，这是最初的问题。我不知道如何诚实地调试这个

一口气拉入所有用户看起来不太好。在服务器端这样做也看起来不太好。所以你应该有一个芹菜任务，你应该从你的网络服务器启动，然后你应该让任务启动其他发送电子邮件任务。此外，您应该在从数据库中拉出所有用户时使用分页。 simpleisbetterthancomplex.com/tutorial/2016/08/03/…。接下来，当您只使用电子邮件时，为什么要提取所有字段？您应该减少用户查询的占用空间

【参考方案1】：

html 电子邮件渲染是开销——如果您想让调用者更快地完成此操作，请将模板渲染移动到 celery 任务中。

@celery.task
def send_async_email_celery(subject, from_, to_, template, **kwargs):
    msg = Message(subject, sender=from_, recipients=[to_])
    msg.html = render_template(template + '.html', **kwargs)
    mail.send(msg)
 
def send_email_celery(to, subject, template, **kwargs):
    countdown = kwargs.get('countdown', 600)
    app = current_app._get_current_object()
    from_ = app.config['MAIL_SENDER']
    send_async_email_celery.apply_async(args=[subject, from_, to, template], kwargs=kwargs, countdown=countdown)

【讨论】：

经过数小时的重构，现在循环速度很快，但我遇到了同样的问题。