最近组里的小伙伴,搞条形码识别,没搞定,我简单写了个识别的程序,然后用Gunicorn 部署了下,很有参考意义。为了满足可扩展性 (后续增加OCR、图像处理等功能)和内网部署 的需求,使用 FastAPI 作为 Web 框架(因为它性能高、开发快、原生支持异步),配合 Gunicorn 作为 WSGI/ASGI 服务器进行部署。
整个小项目遵循了分层架构 (路由层、业务层、工具层),这样以后要加新功能时,就不需要改动原有的代码。
以下是完整的解决方案,包含代码结构、部署策略和调用方式。
1. 项目结构设计 为了应对未来的需求变更,我们将代码分为以下几层:
Plaintext
1 2 3 4 5 6 7 8 barcode_service/ ├── main.py # 入口文件,负责启动 App ├── gunicorn_conf.py # Gunicorn 配置文件 ├── requirements.txt # 依赖包 └── app/ ├── __init__.py ├── api.py # 路由层:定义 URL 和参数 └── services.py # 业务层:具体的识别逻辑(封装你的代码)
2. 完整代码实现 Step 1: 准备依赖 (requirements.txt) 我们需要 FastAPI, Uvicorn (配合 Gunicorn 使用), 和 python-multipart (用于文件上传)。
Plaintext
1 2 3 4 5 6 fastapi uvicorn gunicorn pyzbar pillow python-multipart
Step 2: 业务逻辑层 (app/services.py) 这里封装核心代码。重点改进是:不再读取本地路径,而是直接读取内存中的文件流 。这样调用方只需要上传图片,不需要关心服务器的文件系统。
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 from pyzbar.pyzbar import decodefrom PIL import Imageimport ioclass ScannerService : @staticmethod def scan_barcode_from_bytes (image_bytes: bytes ): try : img = Image.open (io.BytesIO(image_bytes)) decoded_objects = decode(img) results = [] for obj in decoded_objects: code_data = obj.data.decode('utf-8' ) code_type = obj.type rect = { "left" : obj.rect.left, "top" : obj.rect.top, "width" : obj.rect.width, "height" : obj.rect.height } results.append({ "type" : code_type, "content" : code_data, "location" : rect }) return results except Exception as e: print (f"Error processing image: {e} " ) return []
Step 3: 路由接口层 (app/api.py) 这里定义对外的接口。
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 from fastapi import APIRouter, UploadFile, File, HTTPExceptionfrom app.services import ScannerService router = APIRouter()@router.post("/scan/barcode" ) async def extract_barcode (file: UploadFile = File(... ) ): """ 接收上传的图片文件,返回识别到的条码信息 """ if file.content_type not in ["image/jpeg" , "image/png" , "image/bmp" ]: raise HTTPException(status_code=400 , detail="Invalid image format" ) content = await file.read() results = ScannerService.scan_barcode_from_bytes(content) return { "filename" : file.filename, "count" : len (results), "data" : results }
Step 4: 应用入口 (main.py) Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from fastapi import FastAPIfrom app.api import router app = FastAPI( title ="Intranet Image Service" , description ="内网图像识别微服务" , version ="1.0.0" ) app.include_router(router, prefix ="/api/v1" ) @app.get ("/health" ) def health_check(): return {"status" : "ok" }
Step 5: Gunicorn 配置 (gunicorn_conf.py) 这是生产环境部署的关键。
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 import multiprocessing bind = "0.0.0.0:8000" worker_class = "uvicorn.workers.UvicornWorker" workers = multiprocessing.cpu_count() * 2 + 1 timeout = 60 loglevel = "info" accesslog = "-" errorlog = "-"
3. 部署策略 由于涉及到 pyzbar,它底层依赖系统的 zbar 共享库,这是部署中最容易报错的地方。
第一步:安装系统级依赖 (Linux环境) 我的服务器是 Centos:
Bash
1 2 3 4 5 6 7 8 9 10 11 12 sudo yum install epel-release -ysudo yum makecachesudo yum install zbar zbar-devel python3 python3-devel -ysudo yum install gcc -y
这里第一步一般搞不定,所以需要科学的办法:
做好备份,这是CentOS-Base.repo:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 [base] name=CentOS-$releasever - Base - mirrors.aliyun.com failovermethod=priority baseurl=http://mi rrors.aliyun.com/centos/ $releasever /os/ $basearch / http://mi rrors.aliyuncs.com/centos/ $releasever /os/ $basearch / http://mi rrors.cloud.aliyuncs.com/centos/ $releasever /os/ $basearch / gpgcheck=1 gpgkey=http://mi rrors.aliyun.com/centos/ RPM-GPG-KEY-CentOS-7 [updates] name=CentOS-$releasever - Updates - mirrors.aliyun.com failovermethod=priority baseurl=http://mi rrors.aliyun.com/centos/ $releasever /updates/ $basearch / http://mi rrors.aliyuncs.com/centos/ $releasever /updates/ $basearch / http://mi rrors.cloud.aliyuncs.com/centos/ $releasever /updates/ $basearch / gpgcheck=1 gpgkey=http://mi rrors.aliyun.com/centos/ RPM-GPG-KEY-CentOS-7 [extras] name=CentOS-$releasever - Extras - mirrors.aliyun.com failovermethod=priority baseurl=http://mi rrors.aliyun.com/centos/ $releasever /extras/ $basearch / http://mi rrors.aliyuncs.com/centos/ $releasever /extras/ $basearch / http://mi rrors.cloud.aliyuncs.com/centos/ $releasever /extras/ $basearch / gpgcheck=1 gpgkey=http://mi rrors.aliyun.com/centos/ RPM-GPG-KEY-CentOS-7 [centosplus] name=CentOS-$releasever - Plus - mirrors.aliyun.com failovermethod=priority baseurl=http://mi rrors.aliyun.com/centos/ $releasever /centosplus/ $basearch / http://mi rrors.aliyuncs.com/centos/ $releasever /centosplus/ $basearch / http://mi rrors.cloud.aliyuncs.com/centos/ $releasever /centosplus/ $basearch / gpgcheck=1 enabled=0 gpgkey=http://mi rrors.aliyun.com/centos/ RPM-GPG-KEY-CentOS-7 [contrib] name=CentOS-$releasever - Contrib - mirrors.aliyun.com failovermethod=priority baseurl=http://mi rrors.aliyun.com/centos/ $releasever /contrib/ $basearch / http://mi rrors.aliyuncs.com/centos/ $releasever /contrib/ $basearch / http://mi rrors.cloud.aliyuncs.com/centos/ $releasever /contrib/ $basearch / gpgcheck=1 enabled=0 gpgkey=http://mi rrors.aliyun.com/centos/ RPM-GPG-KEY-CentOS-7
然后是:epel.repo
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 [epel] name =Extra Packages for Enterprise Linux 7 - $basearch baseurl =http://mirrors.aliyun.com/epel/7 /$basearch failovermethod =priorityenabled =1 gpgcheck =0 gpgkey =file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-7 [epel-debuginfo] name =Extra Packages for Enterprise Linux 7 - $basearch - Debugbaseurl =http://mirrors.aliyun.com/epel/7 /$basearch /debugfailovermethod =priorityenabled =0 gpgkey =file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-7 gpgcheck =0 [epel-source] name =Extra Packages for Enterprise Linux 7 - $basearch - Sourcebaseurl =http://mirrors.aliyun.com/epel/7 /SRPMSfailovermethod =priorityenabled =0 gpgkey =file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-7 gpgcheck =0
第二步:安装 Python 环境 Bash
1 2 3 4 5 6 python3 -m venv venvsource venv/bin/activate pip install -r requirements.txt
第三步:使用 Systemd 托管 (推荐) 为了保证服务挂掉自动重启,以及开机自启,建议创建一个 systemd 服务文件。
创建文件 /etc/systemd/system/barcode-service.service:
Ini, TOML
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 [Unit] Description =Gunicorn instance to serve Barcode ServiceAfter =network.target[Service] User =rootGroup =rootWorkingDirectory =/path/to/your/barcode_serviceEnvironment ="PATH=/path/to/your/barcode_service/venv/bin" ExecStart =/path/to/your/barcode_service/venv/bin/gunicorn -c gunicorn_conf.py main:app[Install] WantedBy =multi-user.target
启动服务:
Bash
1 2 3 sudo systemctl daemon-reloadsudo systemctl start barcode-servicesudo systemctl enable barcode-service
4. 外部如何调用 (客户端示例) 现在服务运行在服务器的 8000 端口,接口地址是 http://<服务器IP>:8000/api/v1/scan/barcode。
方式 A: 使用 Python requests (比如其他后端服务调用) Python
1 2 3 4 5 6 7 8 9 10 11 import requests url = "http://192.168.1.100:8000/api/v1/scan/barcode" file_path = "product_photo.jpg" with open (file_path, "rb" ) as f : # 这里的 'file' 对应 FastAPI 中的参数名 files = {"file" : ("filename.jpg" , f , "image/jpeg" )} response = requests.post(url, files =files )print (response.json())
方式 B: 使用 cURL (测试用) Bash
1 2 3 4 curl -X POST "http://192.168.1.100:8000/api/v1/scan/barcode" \ -H "accept: application/json" \ -H "Content-Type: multipart/form-data" \ -F "file=@./product_photo.jpg"
5. 关于“后续的需求更新”策略 这套架构如何应对未来的变更:
新增识别类型 (如 OCR):
在 app/services.py 增加一个 OCRService 类。
在 app/api.py 增加一个 @router.post("/scan/ocr") 接口。
不需要修改现有的条码识别代码,互不影响。
性能瓶颈:
如果识别太慢,Gunicorn 的 workers 数量可以调整。
因为是 HTTP 接口,如果未来压力巨大,可以将 api.py 里的逻辑改为“接收图片 -> 丢入 Redis 队列”,然后另写一个 Worker 脚本消费队列。
内网变公网:
目前的 Gunicorn 配置在 Nginx 后面加一层反向代理即可,无需改动代码。