### 开发Python Django Celery Redis测试脚本 以下是基于 Python、Django、Celery 和 Redis 的完整测试脚本开发指南。此方案涵盖了项目的初始化、配置以及任务执行。 #### 1. 创建 Django 项目并设置环境 首先，创建一个新的 Django 项目 `myproject` 并进入其目录： ```bash django-admin startproject myproject cd myproject ``` 接着，在项目的根目录下创建一个名为 `celery.py` 的文件用于定义 Celery 应用程序[^1]。该文件的内容如下所示： ```python import os from celery import Celery from django.conf import settings os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') app = Celery('myproject') app.conf.timezone = 'Asia/Shanghai' app.conf.enable_utc = False class Config: BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' app.config_from_object(Config) @app.task(bind=True) def debug_task(self): print(f'Request: {self.request!r}') # 自动加载 tasks 文件中的任务函数 app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) ``` 在此部分中，通过 `BROKER_URL` 定义了消息队列使用的 Redis 地址和端口，并设置了默认的任务结果存储位置为同一实例下的不同数据库编号[^2]。 #### 2. 配置 Redis 确保本地已安装 Redis 或者可以通过网络访问到远程 Redis 实例。如果未启动服务，则需运行以下命令来激活 Redis： ```bash redis-server ``` 验证连接是否正常工作可以尝试 ping 命令： ```bash redis-cli ping PONG ``` #### 3. 添加 Celery Worker 启动脚本 为了方便管理 worker 进程，可以在项目根路径新增 shell 脚本来简化操作流程。例如命名为 `start_celery_worker.sh` 的 Bash 脚本内容如下： ```bash #!/bin/bash celery -A myproject worker --loglevel=info ``` 赋予可执行权限后再调用即可开启后台处理单元： ```bash chmod +x start_celery_worker.sh ./start_celery_worker.sh ``` #### 4. 编写异步任务逻辑 假设我们希望实现发送电子邮件功能作为演示案例之一。那么需要先引入必要的依赖库并通过 pip 工具完成安装过程： ```bash pip install redis==4.5.* celery==5.* ``` 随后于某个应用模块内部新建子目录结构如 apps/tasks/__init__.py 下放置实际业务代码片段。比如下面展示了一个简单的邮件通知方法样例： ```python from celery import shared_task from time import sleep @shared_task(name="send_email_notification") def send_email(email_address, subject, body): """Simulate sending an email.""" try: sleep(5) # Simulating network delay or processing time. message = f'Sent "{subject}" to {email_address}. Body was:\n{body}' with open("/tmp/email_log.txt", "a") as file_handler: file_handler.write(message + "\n") return {"status": "success", "message": message} except Exception as exc: error_msg = str(exc) raise ValueError(error_msg) ``` 最后记得更新 INSTALLED_APPS 设置列表加入对应的应用名称以便能够被自动扫描发现注册好的 task 函数资源[^2]。 --- #### 注意事项 当处于调试模式时可能会遇到内存泄漏警告提示信息[^3]。因此建议仅在生产环境中关闭 DEBUG 参数开关以优化性能表现。 --- ### 测试整个流程 现在可以从终端触发刚才编写的自定义任务看看效果如何吧！只需简单几行 python 控制台交互语句就能轻松搞定啦～ ```python >>> from apps.tasks.send_emails import send_email >>> result = send_email.delay("test@example.com", "Hello!", "This is a test.") >>> print(result.get()) {'status': 'success', ... } ```