1. 项目背景与需求分析
在智能制造行业快速发展的今天,传统ERP系统面临着诸多挑战。我最近完成了一个基于Django框架的智能制造业ERP系统定制化项目,包含完整的APP小程序端实现。这个项目源于一家中型汽车零部件制造企业的实际需求,他们原有的ERP系统存在以下痛点:
- 移动办公能力薄弱,管理层无法实时掌握生产数据
- 各业务模块(生产、采购、库存)数据孤岛严重
- 缺乏针对离散制造业的精细化排产功能
- 现有系统二次开发成本高,响应速度慢
经过需求调研,我们确定了系统的核心功能矩阵:
| 功能模块 | 子功能 | 技术实现要点 |
|---|---|---|
| 生产管理 | 工单排程、设备状态监控、质量追溯 | Django Channels实时通信 |
| 供应链管理 | 智能采购预测、供应商评估 | Pandas数据分析集成 |
| 库存管理 | 智能预警、批次追踪 | Redis缓存优化 |
| 移动端 | 生产看板、审批流程 | 微信小程序+Rest API |
2. 技术架构设计
2.1 整体架构方案
系统采用前后端分离架构,后端基于Django REST Framework构建,前端包含管理后台和微信小程序双端。技术选型经过多维度评估:
graph TD A[微信小程序] --> B[Nginx] C[Web管理端] --> B B --> D[Django REST API] D --> E[PostgreSQL] D --> F[Redis] D --> G[Celery]注意:实际项目中我们移除了Mermaid图表,改用文字描述架构,因为发现部分客户环境存在渲染兼容性问题。
2.2 Django模型设计要点
针对制造业特性,核心模型设计采用了这些特殊处理:
class WorkOrder(models.Model): STATUS_CHOICES = [ ('pending', '待生产'), ('processing', '生产中'), ('partial', '部分完工'), ('completed', '已完成'), ] code = models.CharField(max_length=20, unique=True) product = models.ForeignKey(Product, on_delete=models.PROTECT) planned_quantity = models.PositiveIntegerField() actual_quantity = models.PositiveIntegerField(default=0) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending') # 离散制造业特有字段 routing_steps = models.JSONField() # 工艺路线 equipment_requirements = models.JSONField() # 设备需求 def save(self, *args, **kwargs): # 自动计算进度 if self.actual_quantity > 0: if self.actual_quantity >= self.planned_quantity: self.status = 'completed' else: self.status = 'partial' super().save(*args, **kwargs)关键设计决策:
- 使用JSONField存储动态工艺路线,避免频繁修改数据库结构
- 重写save()方法实现业务状态自动流转
- 建立复合索引优化高频查询性能
3. 核心功能实现细节
3.1 智能排产算法实现
制造业排产是系统核心难点,我们开发了基于约束理论的排产引擎:
def generate_schedule(orders, start_date): """ 智能排产核心算法 :param orders: 待排产工单列表 :param start_date: 排产开始日期 :return: 排产结果 """ # 1. 数据预处理 validated_orders = validate_orders(orders) # 2. 优先级排序(交期紧迫度+工艺复杂度) sorted_orders = sorted( validated_orders, key=lambda x: (x.due_date, -x.get_complexity_score()), ) # 3. 资源冲突检测 schedule = [] resource_calendar = ResourceCalendar(start_date) for order in sorted_orders: allocation = resource_calendar.allocate(order) if allocation: schedule.append({ 'order': order, 'start': allocation['start'], 'end': allocation['end'], 'resources': allocation['resources'] }) # 4. 结果优化(并行度最大化) return optimize_schedule(schedule)实际项目中我们发现了几个关键优化点:
- 使用numpy向量化运算替代循环,性能提升40倍
- 对长期工单采用分片排产策略,减少内存占用
- 引入Redis缓存历史排产模式,相似工单直接复用
3.2 微信小程序关键技术
小程序端采用Taro框架实现跨平台开发,主要解决的技术难点包括:
- 实时数据同步:
// 建立WebSocket连接 const socket = wx.connectSocket({ url: 'wss://yourdomain.com/ws/production/', success: () => { console.log('连接成功') } }) socket.onMessage((res) => { const data = JSON.parse(res.data) if (data.type === 'EQUIPMENT_UPDATE') { this.setData({ equipmentStatus: data.payload }) } })- 大文件分片上传(用于质量缺陷照片):
async function uploadFile(filePath) { const uploadId = await getUploadId() const chunkSize = 1024 * 1024 // 1MB const fileSize = (await wx.getFileInfo({filePath})).size const chunkCount = Math.ceil(fileSize / chunkSize) for (let i = 0; i < chunkCount; i++) { const start = i * chunkSize const end = Math.min(fileSize, start + chunkSize) const chunk = await readFileChunk(filePath, start, end) await uploadChunk({ uploadId, chunkIndex: i, chunkData: chunk }) } return completeUpload(uploadId) }4. 性能优化实战
4.1 数据库优化方案
针对制造业高频查询场景,我们实施了这些优化措施:
- 查询优化:
# 反例 - N+1查询问题 orders = WorkOrder.objects.filter(status='processing') for order in orders: print(order.product.name) # 每次循环都查询数据库 # 正例 - select_related优化 orders = WorkOrder.objects.select_related('product').filter(status='processing')- 索引策略:
-- 组合索引提升工单查询性能 CREATE INDEX idx_workorder_status_product ON erp_workorder (status, product_id); -- 函数索引优化日期查询 CREATE INDEX idx_invoice_date ON erp_invoice (DATE(create_time));- 分区表示例(按月分区的生产记录表):
class PartitionedRouter: def db_for_read(self, model, **hints): if model._meta.app_label == 'production': month = hints.get('month') if month: return f'production_{month[:7].replace("-", "_")}' return None4.2 缓存策略设计
采用多级缓存架构大幅提升系统响应速度:
- 热点数据缓存:
def get_equipment_status(equipment_id): cache_key = f'equipment_status:{equipment_id}' status = cache.get(cache_key) if not status: status = Equipment.objects.get(pk=equipment_id).status cache.set(cache_key, status, timeout=60) # 60秒过期 return status- 列表页缓存策略:
class OrderListView(APIView): @method_decorator(cache_page(60*5)) # 5分钟页面缓存 @method_decorator(vary_on_headers('Authorization')) def get(self, request): queryset = self.get_queryset() serializer = OrderSerializer(queryset, many=True) return Response(serializer.data)5. 部署与运维实践
5.1 容器化部署方案
采用Docker Compose实现一键部署:
# Django服务Dockerfile FROM python:3.9-slim ENV PYTHONUNBUFFERED 1 RUN apt-get update && apt-get install -y \ gcc \ libpq-dev \ && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . .# docker-compose.prod.yml version: '3.8' services: web: build: . command: gunicorn erp.wsgi:application --bind 0.0.0.0:8000 volumes: - static_volume:/app/static depends_on: - redis - db db: image: postgres:13 volumes: - pg_data:/var/lib/postgresql/data environment: POSTGRES_PASSWORD: ${DB_PASSWORD} redis: image: redis:6 volumes: - redis_data:/data5.2 监控系统搭建
使用Prometheus+Grafana构建监控看板:
- Django指标暴露:
# prometheus_client集成 from prometheus_client import start_http_server, Counter REQUEST_COUNT = Counter( 'django_requests_total', 'Total HTTP Requests', ['method', 'endpoint', 'status'] ) class MetricsMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): response = self.get_response(request) REQUEST_COUNT.labels( method=request.method, endpoint=request.path, status=response.status_code ).inc() return response- 关键监控指标:
- 数据库查询耗时(P99)
- 接口响应时间分布
- Celery任务队列积压量
- 缓存命中率
- 微信小程序API调用频次
6. 项目总结与经验分享
经过三个月的开发和两个月的试运行,系统最终实现了:
- 生产计划编制效率提升70%
- 库存周转率提高35%
- 移动端审批流程平均耗时从8小时缩短至30分钟
几个值得分享的技术心得:
- Django ORM高级技巧:
# 批量更新避免N+1问题 WorkOrder.objects.filter( status='processing', start_time__lt=timezone.now() - timedelta(hours=8) ).update(status='delayed') # 子查询优化 on_time_orders = WorkOrder.objects.filter( actual_end__lte=F('planned_end') ) Product.objects.filter( id__in=Subquery(on_time_orders.values('product_id')) )- 微信小程序调试技巧:
- 使用Charles抓包分析接口问题
- 开启vConsole查看详细日志
- 利用Taro的跨平台调试工具
- 性能优化黄金法则:
- 先测量(使用Django Debug Toolbar)
- 再优化(数据库>缓存>算法)
- 最后压测(Locust模拟并发)
这个项目让我深刻体会到,制造业ERP系统的核心价值不在于技术的新颖程度,而在于对业务场景的深度理解。比如我们最初设计的精美甘特图排产界面,在实际车间主任使用时反而不如简单的列表视图加颜色标识来得实用。好的技术方案应该是隐形的,让用户专注在业务本身而非工具使用上。