MES系统架构设计实战：模块化设计从混乱到清晰的演进

实战指数：★★★★☆ | 收藏指数：★★★★★

一、问题背景：模块耦合严重的痛苦

2018年我接手了一个自研MES项目，前任团队写了大概15万行C#代码，但几乎所有模块都直接调用数据库，一个简单的改工单状态操作，牵扯到5个不同的类，改一个Bug影响三个模块简直是家常便饭。

那个系统上线后，光是一个"修改工艺路线"的需求，开发+测试+部署整整用了3周，而且上线后还引入了2个新Bug。这种耦合到骨髓的架构，逼得我们不得不启动重构。

二、技术原理：分层架构与领域驱动设计

成熟MES通常采用三层/四层架构：表现层（Web/APP/大屏）→ 业务服务层（领域模型+应用服务）→ 数据访问层（Repository模式）→ 基础设施层（消息队列/缓存/文件）。每层只对上下层负责，领域模型保持技术无关性。

DDD（Domain-Driven Design）核心概念：聚合根（Aggregate Root）是领域对象的统一入口，比如工单（WorkOrder）就是一个聚合根，所有对工单的操作都通过WorkOrderService进行，保证聚合内的一致性约束。

三、实战：从三层到微服务的演进过程

第一步：纵向拆分。按业务领域（工单、库存、SPC、设备）拆分为独立服务，每个服务独立数据库（Database per Service）。工单服务只关心工单相关逻辑，不再直接读写库存表。

第二步：依赖倒置。数据访问层定义为接口（IWorkOrderRepository），基础设施层实现具体逻辑，上层业务代码只依赖接口，不依赖实现。这一步花了2个月，但效果立竿见影——换数据库从改2周变成改2天。

第三步：事件驱动。用RabbitMQ做服务间解耦，工单状态变更发布事件，库存服务订阅事件后自动更新，彻底消灭跨服务的同步调用。

图1：MES架构从网状耦合（左）到清晰分层（右）的演进

图2：MES架构重构前后关键指标对比

四、完整代码：服务注册与接口调用示例

以下代码展示服务注册中心和依赖注入的配置，以及工单服务的标准调用流程。使用反射自动扫描并注册服务实现类，减少手工配置。

MES架构示例：服务注册中心

from abc import ABC, abstractmethod from typing import Dict, Type, Any import inspect # 1. 定义仓储接口（依赖倒置） class IWorkOrderRepository(ABC): @abstractmethod def get_by_id(self, wo_id: str) -> Dict[str, Any]: pass @abstractmethod def update_status(self, wo_id: str, status: str) -> bool: pass @abstractmethod def list_by_state(self, state: str) -> list: pass class ISpcRepository(ABC): @abstractmethod def save_control_chart(self, data: Dict) -> bool: pass # 2. 服务容器（简单DI容器） class ServiceContainer: _services: Dict[Type, Type] = {} _instances: Dict[Type, Any] = {} @classmethod def register(cls, interface: Type, impl: Type): cls._services[interface] = impl @classmethod def resolve(cls, interface: Type) -> Any: if interface not in cls._instances: impl = cls._services.get(interface) if not impl: raise ValueError(f"Service {interface} not registered") cls._instances[interface] = impl() return cls._instances[interface] @classmethod def auto_register(cls, base_package): # 反射扫描：自动将接口的实现类注册到容器 for name, obj in inspect.getmembers(__import__(base_package)): if inspect.isclass(obj) and hasattr(obj, '__bases__'): for base in obj.__bases__: if base != ABC and issubclass(base, ABC): cls.register(base, obj) # 3. 应用服务（用例编排层） class WorkOrderService: def __init__(self): self.repo = ServiceContainer.resolve(IWorkOrderRepository) def change_route(self, wo_id: str, new_route: str) -> Dict: wo = self.repo.get_by_id(wo_id) if not wo: raise ValueError(f"WO {wo_id} not found") if wo['status'] not in ['released', 'pending']: raise ValueError(f"Cannot change route at status {wo['status']}") success = self.repo.update_status(wo_id, 'route_changed') return {'success': success, 'wo_id': wo_id, 'new_route': new_route}

为什么这样写：IWorkOrderRepository接口定义数据访问契约，使业务层与具体数据库解耦；ServiceContainer通过简单DI容器实现依赖注入，支持接口替换（如测试时注入MockRepo）；auto_register用反射自动扫描，符合开闭原则，新增服务无需修改注册代码。

五、效果对比

六、实施建议

第一，不要一开始就想做微服务。先在单体应用内做好模块化拆分（Namespace/Assembly分离），验证领域边界后再考虑服务化。第二，API版本管理要提前规划。v1/api/workorders和v2/api/workorders的路由分离，避免升级时影响旧客户端。第三，引入Swagger/OpenAPI做接口文档自动化，防止接口文档过时引发的扯皮。

七、进阶方向

Kubernetes+Docker容器化部署，支持各服务独立扩缩容；Istio服务网格实现熔断、限流、链路追踪；GraphQL统一API网关，解决REST API过度获取/不足获取问题；低代码平台集成，让工艺工程师通过配置而非代码来调整业务流程。

📝 互动话题

你们FAB的设备综合效率OEE目前大概在什么水平？最大的损失来源是哪一块？

在实施OEE改善项目时，有什么坑是特别容易踩的？欢迎评论区分享！

觉得这篇文章有收获？欢迎收藏、点赞支持！ 您的支持是我持续输出的最大动力！ 本文首发于：blog.csdn.net/yeflashzhihui