celery配置国际化

### Celery 国际化配置 为了使 Celery 支持多语言环境下的任务执行，可以利用 Python 的 `gettext` 库来实现国际化功能。这涉及到设置翻译文件以及调整 Celery 配置以支持这些翻译。 #### 设置翻译目录结构 首先，在项目的根目录下创建一个名为 `locale` 的文件夹用于存储不同语言的消息目录： ``` project/ ├── locale/ │ ├── en_US/ │ │ └── LC_MESSAGES/ │ │ └── messages.mo │ └── zh_CN/ │ └── LC_MESSAGES/ │ └── messages.mo └── tasks.py ``` 每种语言对应一个子文件夹，其中包含编译后的 `.mo` 文件[^2]。 #### 修改 Celery 实例配置 在定义 Celery 应用程序时，需引入必要的包并将默认的语言设定为应用程序的首选项之一。同时指明消息目录的位置以便加载相应的翻译资源。 ```python from celery import Celery import os from django.utils.translation import gettext_lazy as _ os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') app = Celery('tasks') app.conf.update( accept_content=['json'], task_serializer='json', result_serializer='json', enable_utc=True, ) # 添加国际化配置选项 app.conf.update({ "task_default_locale": "zh_CN", # 默认区域设置 "accept_language": ["en_US", "zh_CN"], # 接受哪些地区的语言 }) def make_celery(app): """Create a new Celery object and tie together the Celery config to the app's config. Wrap all tasks in the context of the Flask application. :param app: Flask app instance """ ... # 加载自定义的任务类，确保任务运行于Flask上下文中 class ContextTask(celery.Task): def __call__(self, *args, **kwargs): with app.app_context(): return self.run(*args, **kwargs) celery_app.Task = ContextTask # 注册翻译函数到celery环境中 from flask_babel import Babel babel = Babel() babel.init_app(app) celery_app.loader.override_backends['django'] = 'flask_babel.BabelBackend' return celery_app ``` 上述代码片段展示了如何扩展标准的 Celery 初始化过程，加入了对多种语言的支持，并集成了 Flask-Babel 扩展来管理本地化的细节[^3]。 #### 使用翻译字符串 当编写需要国际化的任务逻辑时，应该采用 `_()` 或者更推荐的方式是使用来自 Django 工具链中的 `gettext_lazy()` 函数包裹待翻译的文字串。这样做可以让文字串延迟解析直到真正显示给用户的时候才确定具体使用的语言版本。 ```python @app.task(bind=True) def send_email(self, recipient, subject_template_name, email_template_name, context=None): try: subject = loader.render_to_string(subject_template_name, context or {}) body = render_to_string(email_template_name, context or {}) mail.send_mail(_(subject), _(body), settings.DEFAULT_FROM_EMAIL, [recipient]) logger.info(_('Email sent successfully')) except Exception as exc: raise self.retry(exc=exc, countdown=60*5) # 延迟重试机制 ``` 在这个例子中，假设有一个发送邮件的任务，所有的文本都经过了 `_()` 处理从而能够被正确地转换为目标用户的母语[^1]。