RMBG-2.0实战教程使用CeleryRedis构建异步抠图任务队列系统你是不是也遇到过这样的场景用户上传了一张图片点击“抠图”按钮后页面就卡在那里转圈圈等了几十秒甚至更久才出结果。或者当多个用户同时上传图片时服务器直接“罢工”响应超时。传统的同步处理方式就像一家只有一个厨师的餐厅客人点完菜必须站在厨房门口等厨师做完一道才能做下一道。效率低下体验糟糕。今天我们就来解决这个问题。我将手把手带你基于强大的RMBG-2.0抠图模型构建一个异步抠图任务队列系统。这个系统就像一家现代化的中央厨房用户下单提交图片后立刻收到一个“取餐号”任务ID然后就可以去干别的事了。后厨Celery Worker会按顺序处理订单处理完成后用户凭“取餐号”来领取结果下载透明背景图片。我们将使用Celery作为异步任务队列框架Redis作为消息代理和结果存储Flask作为Web框架。学完这篇教程你将掌握如何将计算密集型的AI模型服务改造成高并发、高可用的异步任务系统。1. 为什么需要异步任务队列在深入代码之前我们先搞清楚“为什么”。理解了痛点解决方案才更有价值。想象一下RMBG-2.0抠图的过程加载模型、读取图片、预处理、模型推理、后处理、保存结果。对于一张1024x1024的图片即使在GPU上这个过程也可能需要1-3秒。如果是CPU时间会更长。同步处理的三大痛点请求阻塞用户浏览器会一直等待服务器响应期间无法进行任何其他操作体验极差。并发能力弱一个Worker比如Gunicorn的一个Worker进程在同一时间只能处理一个请求。多个请求需要排队导致响应时间线性增加。服务不可靠如果处理过程中发生错误如下载的图片损坏整个HTTP请求会失败用户得不到任何反馈。异步任务队列带来的三大好处即时响应Web服务器接收任务后立即返回一个任务ID用户无需等待。高并发可以启动多个Celery Worker进程并行处理大量任务吞吐量大幅提升。任务可追踪每个任务都有唯一ID用户可以随时查询任务状态等待中、处理中、成功、失败。错误隔离单个任务失败不会影响Web服务器和其他任务系统更健壮。我们的目标架构很简单Flask接收请求将抠图任务扔进Redis队列Celery Worker从队列取出任务执行结果存回Redis用户通过任务ID查询结果。2. 环境搭建与项目初始化工欲善其事必先利其器。我们先来搭建好整个开发环境。2.1 创建项目并安装依赖首先创建一个新的项目目录并建立虚拟环境。# 创建项目目录 mkdir rmbg-async-system cd rmbg-async-system # 创建虚拟环境Python 3.8 python -m venv venv # 激活虚拟环境 # Linux/Mac source venv/bin/activate # Windows venv\Scripts\activate # 安装核心依赖 pip install flask celery redis pillow opencv-python numpy requests # 安装PyTorch根据你的CUDA版本选择以CUDA 11.8为例 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118接下来创建我们的项目文件结构。一个清晰的结构能让后续开发维护更轻松。rmbg-async-system/ ├── app.py # Flask主应用 ├── celery_app.py # Celery应用配置 ├── tasks.py # 具体的任务函数核心抠图逻辑 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── static/ # 静态文件CSS, JS │ └── style.css ├── templates/ # HTML模板 │ └── index.html └── logs/ # 日志目录手动创建创建requirements.txt文件方便他人部署Flask2.3.3 celery5.3.4 redis5.0.1 Pillow10.1.0 opencv-python4.8.1.78 numpy1.24.4 requests2.31.0 torch2.1.0cu118 torchvision0.16.0cu1182.2 配置RedisCelery需要一个“消息代理”来传递任务。我们选择Redis因为它同时还能作为结果存储非常方便。如果你还没有安装Redis可以快速通过Docker启动一个docker run -d --name redis-celery -p 6379:6379 redis:alpine或者你也可以直接安装Redis服务。确保Redis服务在localhost:6379运行。3. 核心代码实现三步构建异步系统环境准备好了现在我们开始写代码。整个过程分为三步配置Celery、定义任务、创建Web接口。3.1 第一步配置Celery应用 (celery_app.py)Celery应用是我们的任务调度中心。它知道如何连接Redis如何序列化任务以及任务的路由规则。# celery_app.py from celery import Celery import os # 从环境变量或配置文件中读取Redis地址默认使用本地 redis_url os.getenv(REDIS_URL, redis://localhost:6379/0) # 创建Celery应用实例 # 第一个参数是应用名第二个是消息代理的URL celery_app Celery(rmbg_tasks, brokerredis_url, backendredis_url) # 配置Celery celery_app.conf.update( # 任务序列化方式 task_serializerjson, # 结果序列化方式 result_serializerjson, # 接受的内容类型 accept_content[json], # 时区设置 timezoneAsia/Shanghai, # 是否启用UTC时间 enable_utcTrue, # 任务结果过期时间秒设为1小时 result_expires3600, # Worker从Broker预取的任务数量设为1确保公平调度 worker_prefetch_multiplier1 ) # 自动发现任务模块 # Celery会自动从tasks.py中导入以celery_app.task装饰的函数 celery_app.autodiscover_tasks([tasks])关键点解释broker消息队列Celery从这里取任务。我们用的是Redis。backend结果存储Celery把任务执行结果存回这里。我们也用Redis方便。worker_prefetch_multiplier1这个设置很重要。默认情况下Worker会预取多个任务到内存可能导致一个Worker占着多个任务而其他Worker空闲。设为1确保更公平的任务分发。3.2 第二步实现抠图任务 (tasks.py)这是最核心的部分包含了RMBG-2.0模型的加载和推理逻辑。我们将它包装成一个Celery任务。# tasks.py import os import cv2 import numpy as np from PIL import Image import torch from torchvision import transforms from celery_app import celery_app import time import logging # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 假设模型权重路径请根据你的实际路径修改 MODEL_PATH /root/ai-models/AI-ModelScope/RMBG-2___0/ MODEL_FILE os.path.join(MODEL_PATH, model.pth) # 预处理变换将图像缩放并归一化 preprocess_transform transforms.Compose([ transforms.Resize((1024, 1024)), transforms.ToTensor(), transforms.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]), ]) # 后处理将模型输出的mask转换为二值图像 def postprocess_mask(mask_tensor): 将模型输出的张量转换为二值掩码图像 mask_np mask_tensor.squeeze().cpu().numpy() # 使用0.5作为阈值大于0.5的视为前景 mask_binary (mask_np 0.5).astype(np.uint8) * 255 return mask_binary # 核心抠图函数 def remove_background_with_rmbg(image_path): 使用RMBG-2.0模型移除图片背景 Args: image_path: 输入图片的路径 Returns: output_path: 输出的透明背景PNG图片路径 # 1. 加载图像 original_image Image.open(image_path).convert(RGB) original_size original_image.size # 保存原始尺寸 # 2. 预处理 input_tensor preprocess_transform(original_image) # 添加batch维度: [C, H, W] - [1, C, H, W] input_batch input_tensor.unsqueeze(0) # 3. 加载模型这里简化实际需要加载RMBG-2.0模型 # 注意由于RMBG-2.0模型文件较大且需要特定加载方式 # 此处为伪代码你需要根据实际模型文件替换 logger.warning(f注意此处为模型推理伪代码请替换为真实的RMBG-2.0模型加载和推理逻辑) logger.info(f模拟加载模型权重: {MODEL_FILE}) # --- 模拟推理过程开始 --- # 为了教程的完整性我们模拟一个推理过程 # 实际使用时请替换为以下步骤 # model torch.load(MODEL_FILE, map_locationdevice) # model.eval() # with torch.no_grad(): # output model(input_batch) time.sleep(2) # 模拟模型推理耗时 # 生成一个模拟的mask实际应从模型输出获取 # 这里简单创建一个中心椭圆的mask来模拟抠图结果 h, w 1024, 1024 mask np.zeros((h, w), dtypenp.uint8) cv2.ellipse(mask, (w//2, h//2), (w//3, h//4), 0, 0, 360, 255, -1) # --- 模拟推理过程结束 --- # 4. 后处理将mask缩放到原始图像尺寸 mask_resized cv2.resize(mask, original_size, interpolationcv2.INTER_NEAREST) # 5. 将原始图像转换为RGBA并应用mask作为Alpha通道 original_np np.array(original_image) # 如果原始图像是RGB转换为RGBA if original_np.shape[2] 3: rgba cv2.cvtColor(original_np, cv2.COLOR_RGB2RGBA) else: rgba original_np # 将mask作为Alpha通道 rgba[:, :, 3] mask_resized # 6. 保存结果 output_filename os.path.basename(image_path).split(.)[0] _nobg.png output_path os.path.join(static, results, output_filename) # 确保输出目录存在 os.makedirs(os.path.dirname(output_path), exist_okTrue) # 保存为PNG支持透明通道 result_image Image.fromarray(rgba) result_image.save(output_path, PNG) logger.info(f抠图完成结果保存至: {output_path}) return output_path # 定义Celery任务 celery_app.task(bindTrue, nametasks.remove_background) def remove_background_task(self, image_urlNone, image_pathNone): Celery任务移除图片背景 支持通过URL或本地路径传入图片 task_id self.request.id logger.info(f开始处理任务 {task_id}) try: # 这里可以添加更复杂的逻辑如下载网络图片 if image_url: # 示例从URL下载图片 import requests from io import BytesIO response requests.get(image_url) img_data BytesIO(response.content) # 保存到临时文件 temp_path ftemp_{task_id}.jpg with open(temp_path, wb) as f: f.write(img_data.read()) input_path temp_path else: # 使用本地路径 input_path image_path if not input_path or not os.path.exists(input_path): raise ValueError(未提供有效的图片路径或URL) # 调用抠图函数 result_path remove_background_with_rmbg(input_path) # 清理临时文件如果存在 if image_url and os.path.exists(ftemp_{task_id}.jpg): os.remove(ftemp_{task_id}.jpg) logger.info(f任务 {task_id} 处理成功) return { status: SUCCESS, result_path: result_path, task_id: task_id } except Exception as e: logger.error(f任务 {task_id} 处理失败: {str(e)}) # 任务失败时返回错误信息 return { status: FAILURE, error: str(e), task_id: task_id }代码要点说明celery_app.task装饰器这是将普通函数变为Celery任务的关键。bindTrue参数允许我们在任务函数内访问任务实例self从而获取task_id等信息。任务幂等性一个好的任务应该是幂等的即多次执行相同输入会产生相同结果。我们的抠图函数符合这个要求。错误处理任务内部有完整的try-except块确保单个任务失败不会导致Worker崩溃。错误信息会随结果一起返回。模拟推理由于RMBG-2.0模型文件较大且需要特定加载方式我用模拟过程代替。你需要根据实际的模型文件替换推理部分。3.3 第三步创建Web接口 (app.py)Web层负责接收用户请求触发异步任务并提供任务状态查询和结果下载接口。# app.py from flask import Flask, render_template, request, jsonify, send_file from celery_app import celery_app from tasks import remove_background_task import os import uuid import logging app Flask(__name__) app.config[UPLOAD_FOLDER] static/uploads app.config[MAX_CONTENT_LENGTH] 16 * 1024 * 1024 # 限制上传16MB # 确保上传和结果目录存在 os.makedirs(app.config[UPLOAD_FOLDER], exist_okTrue) os.makedirs(static/results, exist_okTrue) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app.route(/) def index(): 渲染主页 return render_template(index.html) app.route(/api/upload, methods[POST]) def upload_image(): 上传图片并创建抠图任务 if image not in request.files: return jsonify({error: 没有选择文件}), 400 file request.files[image] if file.filename : return jsonify({error: 没有选择文件}), 400 # 生成唯一文件名 file_ext os.path.splitext(file.filename)[1] or .jpg filename f{uuid.uuid4().hex}{file_ext} filepath os.path.join(app.config[UPLOAD_FOLDER], filename) # 保存文件 file.save(filepath) logger.info(f图片已保存: {filepath}) # 创建异步任务 # 这里将本地文件路径传递给任务 task remove_background_task.delay(image_pathfilepath) return jsonify({ message: 任务已提交, task_id: task.id, filename: filename }), 202 # 202 Accepted 表示请求已被接受处理 app.route(/api/task/task_id, methods[GET]) def get_task_status(task_id): 查询任务状态 task_result remove_background_task.AsyncResult(task_id) response { task_id: task_id, status: task_result.status } if task_result.status SUCCESS: # 任务成功返回结果路径 result task_result.result if isinstance(result, dict) and result.get(status) SUCCESS: response[result] result else: response[result] result elif task_result.status FAILURE: # 任务失败返回错误信息 response[error] str(task_result.result) return jsonify(response) app.route(/api/download/filename, methods[GET]) def download_result(filename): 下载处理结果 # 安全检查防止路径遍历 safe_filename os.path.basename(filename) filepath os.path.join(static, results, safe_filename) if not os.path.exists(filepath): return jsonify({error: 文件不存在}), 404 return send_file(filepath, as_attachmentTrue) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)接口设计解析POST /api/upload接收图片上传立即返回task_id状态码202而不是等待处理完成。GET /api/task/task_id前端可以轮询这个接口获取任务状态PENDING、STARTED、SUCCESS、FAILURE。GET /api/download/filename任务成功后前端通过这个接口下载结果图片。状态码202这是HTTP协议中“已接受”的状态码明确告诉客户端请求已被接受处理但尚未完成。3.4 前端页面 (templates/index.html)一个简单的前端页面用于上传图片、查看任务状态和下载结果。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleRMBG-2.0 异步抠图系统/title link relstylesheet href{{ url_for(static, filenamestyle.css) }} /head body div classcontainer header h1 RMBG-2.0 异步抠图系统/h1 p classsubtitle上传图片获取透明背景。任务异步处理无需等待。/p /header main div classupload-section h21. 上传图片/h2 div classupload-area iddropArea input typefile idfileInput acceptimage/* hidden div classupload-placeholder span点击选择或拖拽图片到此区域/span br small支持 JPG, PNG 格式最大 16MB/small /div button classbtn onclickdocument.getElementById(fileInput).click()选择图片/button /div div classpreview idpreviewContainer styledisplay: none; h3原图预览/h3 img idpreviewImage src alt预览 /div /div div classtask-section idtaskSection styledisplay: none; h22. 任务状态/h2 div classtask-info pstrong任务ID:/strong span idtaskId-/span/p pstrong状态:/strong span idtaskStatus classstatus-pending等待中/span/p div classprogress-container div classprogress-bar idprogressBar/div /div p idstatusMessage任务已提交正在排队.../p /div div classresult-section idresultSection styledisplay: none; h3处理结果/h3 div classresult-preview img idresultImage src alt抠图结果 /div a iddownloadLink classbtn btn-success download下载透明背景图片/a /div /div div classinstructions h2如何使用/h2 ol li上传一张需要抠图的图片人物、产品、物品等/li li系统会立即返回一个任务ID无需等待/li li页面会自动轮询任务状态显示处理进度/li li处理完成后可以预览并下载透明背景的PNG图片/li /ol p classtip 提示你可以同时上传多张图片系统会依次处理。/p /div /main /div script let currentTaskId null; let pollInterval null; // 文件选择处理 document.getElementById(fileInput).addEventListener(change, handleFileSelect); // 拖拽处理 const dropArea document.getElementById(dropArea); dropArea.addEventListener(dragover, (e) { e.preventDefault(); dropArea.classList.add(dragover); }); dropArea.addEventListener(dragleave, () { dropArea.classList.remove(dragover); }); dropArea.addEventListener(drop, (e) { e.preventDefault(); dropArea.classList.remove(dragover); if (e.dataTransfer.files.length) { handleFile(e.dataTransfer.files[0]); } }); function handleFileSelect(e) { if (e.target.files.length) { handleFile(e.target.files[0]); } } function handleFile(file) { // 验证文件类型和大小 if (!file.type.match(image.*)) { alert(请选择图片文件); return; } if (file.size 16 * 1024 * 1024) { alert(文件大小不能超过16MB); return; } // 显示预览 const reader new FileReader(); reader.onload function(e) { const preview document.getElementById(previewImage); preview.src e.target.result; document.getElementById(previewContainer).style.display block; }; reader.readAsDataURL(file); // 上传文件 uploadFile(file); } function uploadFile(file) { const formData new FormData(); formData.append(image, file); fetch(/api/upload, { method: POST, body: formData }) .then(response response.json()) .then(data { if (data.task_id) { currentTaskId data.task_id; document.getElementById(taskId).textContent data.task_id; document.getElementById(taskSection).style.display block; startPolling(data.task_id); } else { alert(上传失败: (data.error || 未知错误)); } }) .catch(error { console.error(上传错误:, error); alert(上传失败请重试); }); } function startPolling(taskId) { // 清除之前的轮询 if (pollInterval) clearInterval(pollInterval); // 开始轮询 pollInterval setInterval(() { checkTaskStatus(taskId); }, 2000); // 每2秒查询一次 // 立即查询一次 checkTaskStatus(taskId); } function checkTaskStatus(taskId) { fetch(/api/task/${taskId}) .then(response response.json()) .then(data { updateTaskStatus(data); }) .catch(error { console.error(查询任务状态失败:, error); }); } function updateTaskStatus(data) { const statusElement document.getElementById(taskStatus); const progressBar document.getElementById(progressBar); const messageElement document.getElementById(statusMessage); switch(data.status) { case PENDING: statusElement.textContent 等待中; statusElement.className status-pending; progressBar.style.width 10%; messageElement.textContent 任务已提交等待Worker处理...; break; case STARTED: statusElement.textContent 处理中; statusElement.className status-processing; progressBar.style.width 50%; messageElement.textContent 正在使用RMBG-2.0模型进行抠图处理...; break; case SUCCESS: statusElement.textContent 成功; statusElement.className status-success; progressBar.style.width 100%; messageElement.textContent 抠图完成; // 显示结果 showResult(data.result); // 停止轮询 clearInterval(pollInterval); break; case FAILURE: statusElement.textContent 失败; statusElement.className status-failure; progressBar.style.width 0%; messageElement.textContent 处理失败: (data.error || 未知错误); // 停止轮询 clearInterval(pollInterval); break; default: statusElement.textContent data.status; messageElement.textContent 未知状态; } } function showResult(result) { const resultSection document.getElementById(resultSection); const resultImage document.getElementById(resultImage); const downloadLink document.getElementById(downloadLink); if (result result.result_path) { // 显示结果图片 resultImage.src result.result_path ?t new Date().getTime(); // 加时间戳避免缓存 resultSection.style.display block; // 设置下载链接 const filename result.result_path.split(/).pop(); downloadLink.href /api/download/${filename}; downloadLink.download filename; // 滚动到结果区域 resultSection.scrollIntoView({ behavior: smooth }); } } /script /body /html3.5 简单样式 (static/style.css)/* static/style.css */ * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Oxygen, Ubuntu, sans-serif; line-height: 1.6; color: #333; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh; } .container { max-width: 1200px; margin: 0 auto; padding: 20px; } header { text-align: center; margin-bottom: 40px; color: white; } header h1 { font-size: 2.5rem; margin-bottom: 10px; } .subtitle { font-size: 1.1rem; opacity: 0.9; } main { background: white; border-radius: 15px; padding: 30px; box-shadow: 0 20px 60px rgba(0, 0, 0, 0.3); } .upload-section, .task-section, .instructions { margin-bottom: 40px; } h2 { color: #4a5568; margin-bottom: 20px; padding-bottom: 10px; border-bottom: 2px solid #e2e8f0; } .upload-area { border: 3px dashed #cbd5e0; border-radius: 10px; padding: 60px 20px; text-align: center; background: #f7fafc; transition: all 0.3s ease; margin-bottom: 20px; } .upload-area.dragover { border-color: #667eea; background: #ebf4ff; } .upload-placeholder { margin-bottom: 20px; color: #718096; } .btn { background: #667eea; color: white; border: none; padding: 12px 30px; border-radius: 8px; font-size: 1rem; cursor: pointer; transition: background 0.3s ease; } .btn:hover { background: #5a67d8; } .btn-success { background: #48bb78; } .btn-success:hover { background: #38a169; } .preview, .result-preview { margin-top: 20px; text-align: center; } .preview img, .result-preview img { max-width: 100%; max-height: 400px; border-radius: 8px; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } .task-info { background: #f7fafc; padding: 20px; border-radius: 10px; margin-bottom: 20px; } .task-info p { margin-bottom: 10px; } .status-pending { color: #d69e2e; font-weight: bold; } .status-processing { color: #3182ce; font-weight: bold; } .status-success { color: #38a169; font-weight: bold; } .status-failure { color: #e53e3e; font-weight: bold; } .progress-container { height: 8px; background: #e2e8f0; border-radius: 4px; margin: 20px 0; overflow: hidden; } .progress-bar { height: 100%; background: linear-gradient(90deg, #667eea, #764ba2); width: 0%; transition: width 0.3s ease; } .instructions ol { margin-left: 20px; margin-bottom: 15px; } .instructions li { margin-bottom: 8px; } .tip { background: #ebf8ff; border-left: 4px solid #4299e1; padding: 15px; border-radius: 0 8px 8px 0; margin-top: 20px; } media (max-width: 768px) { .container { padding: 10px; } header h1 { font-size: 2rem; } main { padding: 20px; } }4. 部署与运行让系统动起来代码写完了现在让我们启动整个系统。4.1 启动Redis确保Redis服务正在运行# 如果使用Docker docker start redis-celery # 或者直接运行redis-server redis-server4.2 启动Celery Worker打开一个新的终端窗口激活虚拟环境然后启动Celery Worker# 激活虚拟环境 source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 启动Celery Worker # -A 指定Celery应用模块 # --loglevelinfo 显示详细信息 # --concurrency 指定Worker并发数根据CPU核心数调整 celery -A celery_app.celery_app worker --loglevelinfo --concurrency4你会看到类似这样的输出表示Worker已启动并准备接收任务[config] . app: rmbg_tasks:0x7f8b1c0b4a90 . transport: redis://localhost:6379/0 . results: redis://localhost:6379/0 . concurrency: 4 (prefork) . task events: OFF (enable -E to monitor tasks in real time) [queues] . celery exchangecelery(direct) keycelery4.3 启动Flask Web服务再打开一个终端窗口启动Flask应用# 激活虚拟环境 source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 启动Flask应用 python app.pyFlask服务将在http://localhost:5000启动。4.4 测试系统打开浏览器访问http://localhost:5000上传一张图片你会立即收到一个任务ID观察任务状态的变化等待中 → 处理中 → 成功处理完成后下载透明背景的图片同时处理多个任务打开多个浏览器标签页同时上传多张图片。你会发现所有任务都被快速接收然后Celery Worker会依次处理它们而Web服务器不会阻塞。5. 生产环境部署建议我们现在的系统适合开发环境要用于生产环境还需要一些优化。5.1 使用Gunicorn运行FlaskFlask自带的开发服务器不适合生产环境。使用Gunicornpip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 app:app-w 4启动4个Worker进程-b 0.0.0.0:5000绑定到所有网络接口的5000端口5.2 使用Supervisor管理进程在生产环境中我们需要确保服务在崩溃后能自动重启。使用Supervisor来管理Flask和Celery进程。创建配置文件/etc/supervisor/conf.d/rmbg-async.conf[program:rmbg-web] command/path/to/venv/bin/gunicorn -w 4 -b 0.0.0.0:5000 app:app directory/path/to/rmbg-async-system userwww-data autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/rmbg-web.log [program:rmbg-celery] command/path/to/venv/bin/celery -A celery_app.celery_app worker --loglevelinfo --concurrency4 directory/path/to/rmbg-async-system userwww-data autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/rmbg-celery.log environmentPATH/path/to/venv/bin,PYTHONPATH/path/to/rmbg-async-system然后启动Supervisorsudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start all5.3 使用Nginx作为反向代理Nginx可以处理静态文件、负载均衡和SSL加密。配置示例/etc/nginx/sites-available/rmbg-asyncserver { listen 80; server_name your-domain.com; location / { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } location /static { alias /path/to/rmbg-async-system/static; expires 30d; } }5.4 监控与日志Celery监控可以使用Flower来监控Celery集群pip install flower celery -A celery_app.celery_app flower --port5555然后访问http://localhost:5555查看任务监控面板。日志配置为Celery和Flask配置更详细的日志# 在celery_app.py中添加 celery_app.conf.update( worker_log_format[%(asctime)s: %(levelname)s/%(processName)s] %(message)s, worker_task_log_format[%(asctime)s: %(levelname)s/%(processName)s] [%(task_name)s(%(task_id)s)] %(message)s, )6. 总结通过这篇教程我们成功构建了一个基于RMBG-2.0的异步抠图任务队列系统。让我们回顾一下关键收获系统架构的优势高响应性用户上传图片后立即得到响应无需等待处理完成高并发性通过增加Celery Worker数量可以线性提升处理能力高可靠性单个任务失败不会影响整体系统任务状态可追踪易于扩展可以轻松扩展为分布式系统多个Worker在不同机器上运行核心技术要点Celery异步任务队列的核心负责任务的调度和执行Redis作为消息代理和结果存储简单高效任务状态机Celery内置的任务状态PENDING、STARTED、SUCCESS、FAILURE前端轮询通过定期查询任务状态实现进度显示实际应用建议根据负载调整Worker数量CPU密集型任务如图像处理的Worker数量建议设为CPU核心数1设置任务超时为长时间运行的任务设置超时时间避免Worker被卡住使用任务重试机制对于可能因临时问题失败的任务可以配置自动重试结果存储优化对于大量任务可以考虑使用数据库或对象存储替代Redis存储结果这个系统不仅适用于抠图任务任何计算密集型、耗时的操作都可以套用这个模式比如视频处理、文档转换、大数据分析等。异步任务队列是构建高并发Web应用的必备技能希望这篇教程能帮助你掌握它。现在你的抠图服务再也不会因为用户同时上传多张图片而崩溃了。去构建更强大的应用吧获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
RMBG-2.0实战教程：使用Celery+Redis构建异步抠图任务队列系统
RMBG-2.0实战教程使用CeleryRedis构建异步抠图任务队列系统你是不是也遇到过这样的场景用户上传了一张图片点击“抠图”按钮后页面就卡在那里转圈圈等了几十秒甚至更久才出结果。或者当多个用户同时上传图片时服务器直接“罢工”响应超时。传统的同步处理方式就像一家只有一个厨师的餐厅客人点完菜必须站在厨房门口等厨师做完一道才能做下一道。效率低下体验糟糕。今天我们就来解决这个问题。我将手把手带你基于强大的RMBG-2.0抠图模型构建一个异步抠图任务队列系统。这个系统就像一家现代化的中央厨房用户下单提交图片后立刻收到一个“取餐号”任务ID然后就可以去干别的事了。后厨Celery Worker会按顺序处理订单处理完成后用户凭“取餐号”来领取结果下载透明背景图片。我们将使用Celery作为异步任务队列框架Redis作为消息代理和结果存储Flask作为Web框架。学完这篇教程你将掌握如何将计算密集型的AI模型服务改造成高并发、高可用的异步任务系统。1. 为什么需要异步任务队列在深入代码之前我们先搞清楚“为什么”。理解了痛点解决方案才更有价值。想象一下RMBG-2.0抠图的过程加载模型、读取图片、预处理、模型推理、后处理、保存结果。对于一张1024x1024的图片即使在GPU上这个过程也可能需要1-3秒。如果是CPU时间会更长。同步处理的三大痛点请求阻塞用户浏览器会一直等待服务器响应期间无法进行任何其他操作体验极差。并发能力弱一个Worker比如Gunicorn的一个Worker进程在同一时间只能处理一个请求。多个请求需要排队导致响应时间线性增加。服务不可靠如果处理过程中发生错误如下载的图片损坏整个HTTP请求会失败用户得不到任何反馈。异步任务队列带来的三大好处即时响应Web服务器接收任务后立即返回一个任务ID用户无需等待。高并发可以启动多个Celery Worker进程并行处理大量任务吞吐量大幅提升。任务可追踪每个任务都有唯一ID用户可以随时查询任务状态等待中、处理中、成功、失败。错误隔离单个任务失败不会影响Web服务器和其他任务系统更健壮。我们的目标架构很简单Flask接收请求将抠图任务扔进Redis队列Celery Worker从队列取出任务执行结果存回Redis用户通过任务ID查询结果。2. 环境搭建与项目初始化工欲善其事必先利其器。我们先来搭建好整个开发环境。2.1 创建项目并安装依赖首先创建一个新的项目目录并建立虚拟环境。# 创建项目目录 mkdir rmbg-async-system cd rmbg-async-system # 创建虚拟环境Python 3.8 python -m venv venv # 激活虚拟环境 # Linux/Mac source venv/bin/activate # Windows venv\Scripts\activate # 安装核心依赖 pip install flask celery redis pillow opencv-python numpy requests # 安装PyTorch根据你的CUDA版本选择以CUDA 11.8为例 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118接下来创建我们的项目文件结构。一个清晰的结构能让后续开发维护更轻松。rmbg-async-system/ ├── app.py # Flask主应用 ├── celery_app.py # Celery应用配置 ├── tasks.py # 具体的任务函数核心抠图逻辑 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── static/ # 静态文件CSS, JS │ └── style.css ├── templates/ # HTML模板 │ └── index.html └── logs/ # 日志目录手动创建创建requirements.txt文件方便他人部署Flask2.3.3 celery5.3.4 redis5.0.1 Pillow10.1.0 opencv-python4.8.1.78 numpy1.24.4 requests2.31.0 torch2.1.0cu118 torchvision0.16.0cu1182.2 配置RedisCelery需要一个“消息代理”来传递任务。我们选择Redis因为它同时还能作为结果存储非常方便。如果你还没有安装Redis可以快速通过Docker启动一个docker run -d --name redis-celery -p 6379:6379 redis:alpine或者你也可以直接安装Redis服务。确保Redis服务在localhost:6379运行。3. 核心代码实现三步构建异步系统环境准备好了现在我们开始写代码。整个过程分为三步配置Celery、定义任务、创建Web接口。3.1 第一步配置Celery应用 (celery_app.py)Celery应用是我们的任务调度中心。它知道如何连接Redis如何序列化任务以及任务的路由规则。# celery_app.py from celery import Celery import os # 从环境变量或配置文件中读取Redis地址默认使用本地 redis_url os.getenv(REDIS_URL, redis://localhost:6379/0) # 创建Celery应用实例 # 第一个参数是应用名第二个是消息代理的URL celery_app Celery(rmbg_tasks, brokerredis_url, backendredis_url) # 配置Celery celery_app.conf.update( # 任务序列化方式 task_serializerjson, # 结果序列化方式 result_serializerjson, # 接受的内容类型 accept_content[json], # 时区设置 timezoneAsia/Shanghai, # 是否启用UTC时间 enable_utcTrue, # 任务结果过期时间秒设为1小时 result_expires3600, # Worker从Broker预取的任务数量设为1确保公平调度 worker_prefetch_multiplier1 ) # 自动发现任务模块 # Celery会自动从tasks.py中导入以celery_app.task装饰的函数 celery_app.autodiscover_tasks([tasks])关键点解释broker消息队列Celery从这里取任务。我们用的是Redis。backend结果存储Celery把任务执行结果存回这里。我们也用Redis方便。worker_prefetch_multiplier1这个设置很重要。默认情况下Worker会预取多个任务到内存可能导致一个Worker占着多个任务而其他Worker空闲。设为1确保更公平的任务分发。3.2 第二步实现抠图任务 (tasks.py)这是最核心的部分包含了RMBG-2.0模型的加载和推理逻辑。我们将它包装成一个Celery任务。# tasks.py import os import cv2 import numpy as np from PIL import Image import torch from torchvision import transforms from celery_app import celery_app import time import logging # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 假设模型权重路径请根据你的实际路径修改 MODEL_PATH /root/ai-models/AI-ModelScope/RMBG-2___0/ MODEL_FILE os.path.join(MODEL_PATH, model.pth) # 预处理变换将图像缩放并归一化 preprocess_transform transforms.Compose([ transforms.Resize((1024, 1024)), transforms.ToTensor(), transforms.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]), ]) # 后处理将模型输出的mask转换为二值图像 def postprocess_mask(mask_tensor): 将模型输出的张量转换为二值掩码图像 mask_np mask_tensor.squeeze().cpu().numpy() # 使用0.5作为阈值大于0.5的视为前景 mask_binary (mask_np 0.5).astype(np.uint8) * 255 return mask_binary # 核心抠图函数 def remove_background_with_rmbg(image_path): 使用RMBG-2.0模型移除图片背景 Args: image_path: 输入图片的路径 Returns: output_path: 输出的透明背景PNG图片路径 # 1. 加载图像 original_image Image.open(image_path).convert(RGB) original_size original_image.size # 保存原始尺寸 # 2. 预处理 input_tensor preprocess_transform(original_image) # 添加batch维度: [C, H, W] - [1, C, H, W] input_batch input_tensor.unsqueeze(0) # 3. 加载模型这里简化实际需要加载RMBG-2.0模型 # 注意由于RMBG-2.0模型文件较大且需要特定加载方式 # 此处为伪代码你需要根据实际模型文件替换 logger.warning(f注意此处为模型推理伪代码请替换为真实的RMBG-2.0模型加载和推理逻辑) logger.info(f模拟加载模型权重: {MODEL_FILE}) # --- 模拟推理过程开始 --- # 为了教程的完整性我们模拟一个推理过程 # 实际使用时请替换为以下步骤 # model torch.load(MODEL_FILE, map_locationdevice) # model.eval() # with torch.no_grad(): # output model(input_batch) time.sleep(2) # 模拟模型推理耗时 # 生成一个模拟的mask实际应从模型输出获取 # 这里简单创建一个中心椭圆的mask来模拟抠图结果 h, w 1024, 1024 mask np.zeros((h, w), dtypenp.uint8) cv2.ellipse(mask, (w//2, h//2), (w//3, h//4), 0, 0, 360, 255, -1) # --- 模拟推理过程结束 --- # 4. 后处理将mask缩放到原始图像尺寸 mask_resized cv2.resize(mask, original_size, interpolationcv2.INTER_NEAREST) # 5. 将原始图像转换为RGBA并应用mask作为Alpha通道 original_np np.array(original_image) # 如果原始图像是RGB转换为RGBA if original_np.shape[2] 3: rgba cv2.cvtColor(original_np, cv2.COLOR_RGB2RGBA) else: rgba original_np # 将mask作为Alpha通道 rgba[:, :, 3] mask_resized # 6. 保存结果 output_filename os.path.basename(image_path).split(.)[0] _nobg.png output_path os.path.join(static, results, output_filename) # 确保输出目录存在 os.makedirs(os.path.dirname(output_path), exist_okTrue) # 保存为PNG支持透明通道 result_image Image.fromarray(rgba) result_image.save(output_path, PNG) logger.info(f抠图完成结果保存至: {output_path}) return output_path # 定义Celery任务 celery_app.task(bindTrue, nametasks.remove_background) def remove_background_task(self, image_urlNone, image_pathNone): Celery任务移除图片背景 支持通过URL或本地路径传入图片 task_id self.request.id logger.info(f开始处理任务 {task_id}) try: # 这里可以添加更复杂的逻辑如下载网络图片 if image_url: # 示例从URL下载图片 import requests from io import BytesIO response requests.get(image_url) img_data BytesIO(response.content) # 保存到临时文件 temp_path ftemp_{task_id}.jpg with open(temp_path, wb) as f: f.write(img_data.read()) input_path temp_path else: # 使用本地路径 input_path image_path if not input_path or not os.path.exists(input_path): raise ValueError(未提供有效的图片路径或URL) # 调用抠图函数 result_path remove_background_with_rmbg(input_path) # 清理临时文件如果存在 if image_url and os.path.exists(ftemp_{task_id}.jpg): os.remove(ftemp_{task_id}.jpg) logger.info(f任务 {task_id} 处理成功) return { status: SUCCESS, result_path: result_path, task_id: task_id } except Exception as e: logger.error(f任务 {task_id} 处理失败: {str(e)}) # 任务失败时返回错误信息 return { status: FAILURE, error: str(e), task_id: task_id }代码要点说明celery_app.task装饰器这是将普通函数变为Celery任务的关键。bindTrue参数允许我们在任务函数内访问任务实例self从而获取task_id等信息。任务幂等性一个好的任务应该是幂等的即多次执行相同输入会产生相同结果。我们的抠图函数符合这个要求。错误处理任务内部有完整的try-except块确保单个任务失败不会导致Worker崩溃。错误信息会随结果一起返回。模拟推理由于RMBG-2.0模型文件较大且需要特定加载方式我用模拟过程代替。你需要根据实际的模型文件替换推理部分。3.3 第三步创建Web接口 (app.py)Web层负责接收用户请求触发异步任务并提供任务状态查询和结果下载接口。# app.py from flask import Flask, render_template, request, jsonify, send_file from celery_app import celery_app from tasks import remove_background_task import os import uuid import logging app Flask(__name__) app.config[UPLOAD_FOLDER] static/uploads app.config[MAX_CONTENT_LENGTH] 16 * 1024 * 1024 # 限制上传16MB # 确保上传和结果目录存在 os.makedirs(app.config[UPLOAD_FOLDER], exist_okTrue) os.makedirs(static/results, exist_okTrue) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app.route(/) def index(): 渲染主页 return render_template(index.html) app.route(/api/upload, methods[POST]) def upload_image(): 上传图片并创建抠图任务 if image not in request.files: return jsonify({error: 没有选择文件}), 400 file request.files[image] if file.filename : return jsonify({error: 没有选择文件}), 400 # 生成唯一文件名 file_ext os.path.splitext(file.filename)[1] or .jpg filename f{uuid.uuid4().hex}{file_ext} filepath os.path.join(app.config[UPLOAD_FOLDER], filename) # 保存文件 file.save(filepath) logger.info(f图片已保存: {filepath}) # 创建异步任务 # 这里将本地文件路径传递给任务 task remove_background_task.delay(image_pathfilepath) return jsonify({ message: 任务已提交, task_id: task.id, filename: filename }), 202 # 202 Accepted 表示请求已被接受处理 app.route(/api/task/task_id, methods[GET]) def get_task_status(task_id): 查询任务状态 task_result remove_background_task.AsyncResult(task_id) response { task_id: task_id, status: task_result.status } if task_result.status SUCCESS: # 任务成功返回结果路径 result task_result.result if isinstance(result, dict) and result.get(status) SUCCESS: response[result] result else: response[result] result elif task_result.status FAILURE: # 任务失败返回错误信息 response[error] str(task_result.result) return jsonify(response) app.route(/api/download/filename, methods[GET]) def download_result(filename): 下载处理结果 # 安全检查防止路径遍历 safe_filename os.path.basename(filename) filepath os.path.join(static, results, safe_filename) if not os.path.exists(filepath): return jsonify({error: 文件不存在}), 404 return send_file(filepath, as_attachmentTrue) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)接口设计解析POST /api/upload接收图片上传立即返回task_id状态码202而不是等待处理完成。GET /api/task/task_id前端可以轮询这个接口获取任务状态PENDING、STARTED、SUCCESS、FAILURE。GET /api/download/filename任务成功后前端通过这个接口下载结果图片。状态码202这是HTTP协议中“已接受”的状态码明确告诉客户端请求已被接受处理但尚未完成。3.4 前端页面 (templates/index.html)一个简单的前端页面用于上传图片、查看任务状态和下载结果。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleRMBG-2.0 异步抠图系统/title link relstylesheet href{{ url_for(static, filenamestyle.css) }} /head body div classcontainer header h1 RMBG-2.0 异步抠图系统/h1 p classsubtitle上传图片获取透明背景。任务异步处理无需等待。/p /header main div classupload-section h21. 上传图片/h2 div classupload-area iddropArea input typefile idfileInput acceptimage/* hidden div classupload-placeholder span点击选择或拖拽图片到此区域/span br small支持 JPG, PNG 格式最大 16MB/small /div button classbtn onclickdocument.getElementById(fileInput).click()选择图片/button /div div classpreview idpreviewContainer styledisplay: none; h3原图预览/h3 img idpreviewImage src alt预览 /div /div div classtask-section idtaskSection styledisplay: none; h22. 任务状态/h2 div classtask-info pstrong任务ID:/strong span idtaskId-/span/p pstrong状态:/strong span idtaskStatus classstatus-pending等待中/span/p div classprogress-container div classprogress-bar idprogressBar/div /div p idstatusMessage任务已提交正在排队.../p /div div classresult-section idresultSection styledisplay: none; h3处理结果/h3 div classresult-preview img idresultImage src alt抠图结果 /div a iddownloadLink classbtn btn-success download下载透明背景图片/a /div /div div classinstructions h2如何使用/h2 ol li上传一张需要抠图的图片人物、产品、物品等/li li系统会立即返回一个任务ID无需等待/li li页面会自动轮询任务状态显示处理进度/li li处理完成后可以预览并下载透明背景的PNG图片/li /ol p classtip 提示你可以同时上传多张图片系统会依次处理。/p /div /main /div script let currentTaskId null; let pollInterval null; // 文件选择处理 document.getElementById(fileInput).addEventListener(change, handleFileSelect); // 拖拽处理 const dropArea document.getElementById(dropArea); dropArea.addEventListener(dragover, (e) { e.preventDefault(); dropArea.classList.add(dragover); }); dropArea.addEventListener(dragleave, () { dropArea.classList.remove(dragover); }); dropArea.addEventListener(drop, (e) { e.preventDefault(); dropArea.classList.remove(dragover); if (e.dataTransfer.files.length) { handleFile(e.dataTransfer.files[0]); } }); function handleFileSelect(e) { if (e.target.files.length) { handleFile(e.target.files[0]); } } function handleFile(file) { // 验证文件类型和大小 if (!file.type.match(image.*)) { alert(请选择图片文件); return; } if (file.size 16 * 1024 * 1024) { alert(文件大小不能超过16MB); return; } // 显示预览 const reader new FileReader(); reader.onload function(e) { const preview document.getElementById(previewImage); preview.src e.target.result; document.getElementById(previewContainer).style.display block; }; reader.readAsDataURL(file); // 上传文件 uploadFile(file); } function uploadFile(file) { const formData new FormData(); formData.append(image, file); fetch(/api/upload, { method: POST, body: formData }) .then(response response.json()) .then(data { if (data.task_id) { currentTaskId data.task_id; document.getElementById(taskId).textContent data.task_id; document.getElementById(taskSection).style.display block; startPolling(data.task_id); } else { alert(上传失败: (data.error || 未知错误)); } }) .catch(error { console.error(上传错误:, error); alert(上传失败请重试); }); } function startPolling(taskId) { // 清除之前的轮询 if (pollInterval) clearInterval(pollInterval); // 开始轮询 pollInterval setInterval(() { checkTaskStatus(taskId); }, 2000); // 每2秒查询一次 // 立即查询一次 checkTaskStatus(taskId); } function checkTaskStatus(taskId) { fetch(/api/task/${taskId}) .then(response response.json()) .then(data { updateTaskStatus(data); }) .catch(error { console.error(查询任务状态失败:, error); }); } function updateTaskStatus(data) { const statusElement document.getElementById(taskStatus); const progressBar document.getElementById(progressBar); const messageElement document.getElementById(statusMessage); switch(data.status) { case PENDING: statusElement.textContent 等待中; statusElement.className status-pending; progressBar.style.width 10%; messageElement.textContent 任务已提交等待Worker处理...; break; case STARTED: statusElement.textContent 处理中; statusElement.className status-processing; progressBar.style.width 50%; messageElement.textContent 正在使用RMBG-2.0模型进行抠图处理...; break; case SUCCESS: statusElement.textContent 成功; statusElement.className status-success; progressBar.style.width 100%; messageElement.textContent 抠图完成; // 显示结果 showResult(data.result); // 停止轮询 clearInterval(pollInterval); break; case FAILURE: statusElement.textContent 失败; statusElement.className status-failure; progressBar.style.width 0%; messageElement.textContent 处理失败: (data.error || 未知错误); // 停止轮询 clearInterval(pollInterval); break; default: statusElement.textContent data.status; messageElement.textContent 未知状态; } } function showResult(result) { const resultSection document.getElementById(resultSection); const resultImage document.getElementById(resultImage); const downloadLink document.getElementById(downloadLink); if (result result.result_path) { // 显示结果图片 resultImage.src result.result_path ?t new Date().getTime(); // 加时间戳避免缓存 resultSection.style.display block; // 设置下载链接 const filename result.result_path.split(/).pop(); downloadLink.href /api/download/${filename}; downloadLink.download filename; // 滚动到结果区域 resultSection.scrollIntoView({ behavior: smooth }); } } /script /body /html3.5 简单样式 (static/style.css)/* static/style.css */ * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Oxygen, Ubuntu, sans-serif; line-height: 1.6; color: #333; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh; } .container { max-width: 1200px; margin: 0 auto; padding: 20px; } header { text-align: center; margin-bottom: 40px; color: white; } header h1 { font-size: 2.5rem; margin-bottom: 10px; } .subtitle { font-size: 1.1rem; opacity: 0.9; } main { background: white; border-radius: 15px; padding: 30px; box-shadow: 0 20px 60px rgba(0, 0, 0, 0.3); } .upload-section, .task-section, .instructions { margin-bottom: 40px; } h2 { color: #4a5568; margin-bottom: 20px; padding-bottom: 10px; border-bottom: 2px solid #e2e8f0; } .upload-area { border: 3px dashed #cbd5e0; border-radius: 10px; padding: 60px 20px; text-align: center; background: #f7fafc; transition: all 0.3s ease; margin-bottom: 20px; } .upload-area.dragover { border-color: #667eea; background: #ebf4ff; } .upload-placeholder { margin-bottom: 20px; color: #718096; } .btn { background: #667eea; color: white; border: none; padding: 12px 30px; border-radius: 8px; font-size: 1rem; cursor: pointer; transition: background 0.3s ease; } .btn:hover { background: #5a67d8; } .btn-success { background: #48bb78; } .btn-success:hover { background: #38a169; } .preview, .result-preview { margin-top: 20px; text-align: center; } .preview img, .result-preview img { max-width: 100%; max-height: 400px; border-radius: 8px; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } .task-info { background: #f7fafc; padding: 20px; border-radius: 10px; margin-bottom: 20px; } .task-info p { margin-bottom: 10px; } .status-pending { color: #d69e2e; font-weight: bold; } .status-processing { color: #3182ce; font-weight: bold; } .status-success { color: #38a169; font-weight: bold; } .status-failure { color: #e53e3e; font-weight: bold; } .progress-container { height: 8px; background: #e2e8f0; border-radius: 4px; margin: 20px 0; overflow: hidden; } .progress-bar { height: 100%; background: linear-gradient(90deg, #667eea, #764ba2); width: 0%; transition: width 0.3s ease; } .instructions ol { margin-left: 20px; margin-bottom: 15px; } .instructions li { margin-bottom: 8px; } .tip { background: #ebf8ff; border-left: 4px solid #4299e1; padding: 15px; border-radius: 0 8px 8px 0; margin-top: 20px; } media (max-width: 768px) { .container { padding: 10px; } header h1 { font-size: 2rem; } main { padding: 20px; } }4. 部署与运行让系统动起来代码写完了现在让我们启动整个系统。4.1 启动Redis确保Redis服务正在运行# 如果使用Docker docker start redis-celery # 或者直接运行redis-server redis-server4.2 启动Celery Worker打开一个新的终端窗口激活虚拟环境然后启动Celery Worker# 激活虚拟环境 source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 启动Celery Worker # -A 指定Celery应用模块 # --loglevelinfo 显示详细信息 # --concurrency 指定Worker并发数根据CPU核心数调整 celery -A celery_app.celery_app worker --loglevelinfo --concurrency4你会看到类似这样的输出表示Worker已启动并准备接收任务[config] . app: rmbg_tasks:0x7f8b1c0b4a90 . transport: redis://localhost:6379/0 . results: redis://localhost:6379/0 . concurrency: 4 (prefork) . task events: OFF (enable -E to monitor tasks in real time) [queues] . celery exchangecelery(direct) keycelery4.3 启动Flask Web服务再打开一个终端窗口启动Flask应用# 激活虚拟环境 source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 启动Flask应用 python app.pyFlask服务将在http://localhost:5000启动。4.4 测试系统打开浏览器访问http://localhost:5000上传一张图片你会立即收到一个任务ID观察任务状态的变化等待中 → 处理中 → 成功处理完成后下载透明背景的图片同时处理多个任务打开多个浏览器标签页同时上传多张图片。你会发现所有任务都被快速接收然后Celery Worker会依次处理它们而Web服务器不会阻塞。5. 生产环境部署建议我们现在的系统适合开发环境要用于生产环境还需要一些优化。5.1 使用Gunicorn运行FlaskFlask自带的开发服务器不适合生产环境。使用Gunicornpip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 app:app-w 4启动4个Worker进程-b 0.0.0.0:5000绑定到所有网络接口的5000端口5.2 使用Supervisor管理进程在生产环境中我们需要确保服务在崩溃后能自动重启。使用Supervisor来管理Flask和Celery进程。创建配置文件/etc/supervisor/conf.d/rmbg-async.conf[program:rmbg-web] command/path/to/venv/bin/gunicorn -w 4 -b 0.0.0.0:5000 app:app directory/path/to/rmbg-async-system userwww-data autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/rmbg-web.log [program:rmbg-celery] command/path/to/venv/bin/celery -A celery_app.celery_app worker --loglevelinfo --concurrency4 directory/path/to/rmbg-async-system userwww-data autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/rmbg-celery.log environmentPATH/path/to/venv/bin,PYTHONPATH/path/to/rmbg-async-system然后启动Supervisorsudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start all5.3 使用Nginx作为反向代理Nginx可以处理静态文件、负载均衡和SSL加密。配置示例/etc/nginx/sites-available/rmbg-asyncserver { listen 80; server_name your-domain.com; location / { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } location /static { alias /path/to/rmbg-async-system/static; expires 30d; } }5.4 监控与日志Celery监控可以使用Flower来监控Celery集群pip install flower celery -A celery_app.celery_app flower --port5555然后访问http://localhost:5555查看任务监控面板。日志配置为Celery和Flask配置更详细的日志# 在celery_app.py中添加 celery_app.conf.update( worker_log_format[%(asctime)s: %(levelname)s/%(processName)s] %(message)s, worker_task_log_format[%(asctime)s: %(levelname)s/%(processName)s] [%(task_name)s(%(task_id)s)] %(message)s, )6. 总结通过这篇教程我们成功构建了一个基于RMBG-2.0的异步抠图任务队列系统。让我们回顾一下关键收获系统架构的优势高响应性用户上传图片后立即得到响应无需等待处理完成高并发性通过增加Celery Worker数量可以线性提升处理能力高可靠性单个任务失败不会影响整体系统任务状态可追踪易于扩展可以轻松扩展为分布式系统多个Worker在不同机器上运行核心技术要点Celery异步任务队列的核心负责任务的调度和执行Redis作为消息代理和结果存储简单高效任务状态机Celery内置的任务状态PENDING、STARTED、SUCCESS、FAILURE前端轮询通过定期查询任务状态实现进度显示实际应用建议根据负载调整Worker数量CPU密集型任务如图像处理的Worker数量建议设为CPU核心数1设置任务超时为长时间运行的任务设置超时时间避免Worker被卡住使用任务重试机制对于可能因临时问题失败的任务可以配置自动重试结果存储优化对于大量任务可以考虑使用数据库或对象存储替代Redis存储结果这个系统不仅适用于抠图任务任何计算密集型、耗时的操作都可以套用这个模式比如视频处理、文档转换、大数据分析等。异步任务队列是构建高并发Web应用的必备技能希望这篇教程能帮助你掌握它。现在你的抠图服务再也不会因为用户同时上传多张图片而崩溃了。去构建更强大的应用吧获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
