离散制造、尤其是 MTO（按订单生产）场景里，产品品种多、批次小、交期紧、变更频繁。很多企业常见的问题有：计划和车间脱节、领料与库存不一致、报工延迟、看板不可信、物料消耗难统计。生产管理模块正是把客户订单和车间执行连起来的“最后一公里”——它决定了交付率、在制品水平、物料利用率和交期稳定性。做好这一块，企业能明显提高准交率、减少在制库存、下降物料浪费。
本文你将了解
- 什么是 ERP（离散制造 — MTO）
- 生产管理模块要解决的问题与明确目标
- 系统总体架构（附架构图）
- 生产管理板块功能清单
- 关键业务流程（附流程图）
- 开发技巧与落地建议
- 参考实现：数据库表（DDL）+ 后端参考代码
- 实现效果、验收指标与交付建议
- 部署、运维与后续迭代建议
注：本文示例所用方案模板：简道云ERP系统，给大家示例的是一些通用的功能和模块，都是支持自定义修改的，你可以根据自己的需求修改里面的功能。
一、什么是 ERP（离散制造-MTO）系统
简单说，ERP 是企业资源计划。在离散制造里，常聚焦的核心是 BOM（物料清单）、工艺路线、工单、车间执行、质量追溯与库存。MTO 模式下，排产更偏订单驱动，不是按产能预测大批量生产，而是根据订单生成工单、按工单领料、按工单完工入库。系统要支持弹性的工单拆分、及时的领退料、精确的报工与快速的看板反馈。
二、生产管理板块要解决的问题与目标（KPI）
要解决的问题：
- 计划→工单→车间执行链路打通；
- 领料/退料与库存数据一致；
- 报工及时且准确；
- 看板能实时反映真实的车间状态；
- 生产用料/产出可做统计和追溯。
建议的目标/KPI（举例）：
- 交付准时率 ≥ 95%；
- 在制品（WIP）减少 15%+；
- 领退料差异率 < 1%；
- 报工及时率 ≥ 98%；
- 关键异常响应时间 ≤ 30 分钟。
三、系统总体架构（附架构图）
这里给出一个常见且利于落地的模块化架构：
+-----------------------------+ +----------------------------+
| 客户订单系统 / CRM / 销售 | <--> | 排产引擎（APS）/ 排产服务 |
+-----------------------------+ +----------------------------+
| |
v v
+---------------------------------------------------------------+
| ERP 中台（微服务/模块化） |
| +----------------+ +-----------------+ +----------------+ |
| | 生产管理服务 | | 库存服务(Inventory) | | 采购/供应链 | |
| | (Plan/WO/Pick) | | (WMS) | | (PO/收货) | |
| +----------------+ +-----------------+ +----------------+ |
| | | | |
| v v v |
| +-------------------------------+ |
| | 数据持久层 (RDBMS：Postgres/MySQL) | |
| +-------------------------------+ |
+---------------------------------------------------------------+
| |
v v
+----------------------+ +-------------------------------+
| 车间终端/扫码枪/手持 | | 生产看板 (Web / TV / 移动) |
| (领料/报工/质检) | | (看板 + 甘特 + 实时告警) |
+----------------------+ +-------------------------------+
说明：生产管理服务负责 Plan、WorkOrder、Pick、JobReport、Finish。
库存服务专责库存事务，二者通过 API 或事件（Kafka）保持一致性。
看板通过 WebSocket/Socket.IO 实时推送。
四、生产管理板块功能清单
生产管理包含以下模块与能力：
- 生产计划（Plan） 销售订单转计划；手工调整；周期性计划复制；与 APS 对接。
- 生产工单（Work Order） 工单生命周期：DRAFT → RELEASED → IN_PROGRESS → COMPLETED → CLOSED；支持工单拆分、合并、批次号分配。
- 生产领料（Pick） 按工单领料、支持批次和库位、支持 FIFO、支持部分领料/补领。
- 生产退料（Return） 领料后退料、报废扣减、物料返库流程。
- 生产报工（Job Report） 人工/自动报工；计件/计时/计量；记录不合格品、报废、工时、设备停机。
- 生产入库（Finish Goods） 完工入库、检验入库、与库存服务同步库存增加。
- 生产计划看板（Plan Board） 甘特图、泳道图、按线/班/订单查看；支持实时钻取。
- 生产执行跟踪 工单、工序、物料、人员、设备的全链路追踪与时间戳记录。
- 生产用料统计 / 生产数据统计 实时与历史统计：产量、良率、用料消耗、物料差异、生产效率。
- 辅助表 BOM、工艺路线、工序、资源（产线/设备）、人员、物料主数据、仓位等。
五、关键业务流程
流程概览： 销售订单 → 计划 → 生成工单 → 工单下达（Release）→ 领料 → 加工/报工（多次）→ 质检 → 完工入库 → 统计与关闭。
流程图：
[销售订单]
|
v
[生产计划] ---> [排产引擎]
|
v
[生成工单]
|
v
[工单下达/车间接收]
|
v
[仓库领料] <-----> [库存服务]
|
v
[车间加工/报工] --(异常/不合格)-> [返修/报废]
|
v
[质检]
|
v
[完工入库]
|
v
[统计/追溯/结案]
细化场景示例：工单下达后系统自动根据 BOM 生成 PickList（领料单）；领料执行会在库存服务做扣减并记录批次；车间报工时记录产出和不合格数量，一旦累计产出达到工单数量则触发完工入库并把成品入默认仓位或检验仓。
六、开发技巧与落地建议
- MVP 优先：先把核心链路（工单、领料、报工、完工入库、基础看板）做通；复杂的 APS、自动排产、MES 设备集成后续迭代。
- 事务边界与一致性：领料、退料、完工入库这些操作必须保证库存的一致性。单体应用里靠数据库事务；分布式时建议使用事件驱动 + 补偿（Saga 模式）或库存服务做幂等化处理。
- 工单状态机：把工单生命周期状态化，所有状态变更通过受控接口或状态机组件（如 state machine lib）变更，防止并发冲突。
- 接口与事件解耦：生产服务与库存、采购、质量、看板通过 API + 事件（Kafka/RabbitMQ）解耦，确保模块独立演进。
- 车间友好交互：扫码优先、最小输入、离线容错设计（临时缓存）、语音或大字模式适配老工人。
- 看板实时性：使用 WebSocket / Socket.IO 推送关键事件；对高频统计用 Redis 缓存或物化视图避免 DB 压力。
- 审计与追溯：每次领料/退料/报工都保存操作人、时间、终端、批次，支持回溯与责任定位。
- 异常处理与告警：物料短缺、报废率异常、设备停机需做告警策略并推送到看板/移动端。
- 统计分层：事务表存原子业务，统计/分析走异步日志到 Data Warehouse（ClickHouse/Redshift）做 OLAP。
- 性能注意：热点表（如库存）设计合理索引、分库分表或缓存策略；统计表做预聚合或每日批处理。
七、参考实现
8.1 数据库表
-- 物料表
CREATE TABLE material (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
code VARCHAR(64) UNIQUE NOT NULL,
name VARCHAR(255) NOT NULL,
uom VARCHAR(32) NOT NULL,
lead_time_days INT DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 仓位 / 库存表
CREATE TABLE warehouse_location (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
code VARCHAR(64),
name VARCHAR(255)
);
CREATE TABLE inventory (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
material_id BIGINT NOT NULL,
location_id BIGINT NOT NULL,
qty DECIMAL(18,4) DEFAULT 0,
batch_no VARCHAR(128),
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT fk_inv_material FOREIGN KEY (material_id) REFERENCES material(id)
);
-- BOM 表
CREATE TABLE bom (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
parent_material_id BIGINT NOT NULL,
component_material_id BIGINT NOT NULL,
qty DECIMAL(18,6) NOT NULL,
CONSTRAINT fk_bom_parent FOREIGN KEY (parent_material_id) REFERENCES material(id),
CONSTRAINT fk_bom_comp FOREIGN KEY (component_material_id) REFERENCES material(id)
);
-- 工单表
CREATE TABLE work_order (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
wo_no VARCHAR(64) UNIQUE NOT NULL,
sales_order_no VARCHAR(64),
material_id BIGINT NOT NULL,
qty DECIMAL(18,6) NOT NULL,
produced_qty DECIMAL(18,6) DEFAULT 0,
status VARCHAR(32) DEFAULT 'DRAFT',
release_date TIMESTAMP NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_wo_material FOREIGN KEY (material_id) REFERENCES material(id)
);
-- 领料单与行项
CREATE TABLE pick_list (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
pick_no VARCHAR(64) UNIQUE,
work_order_id BIGINT,
status VARCHAR(32) DEFAULT 'PENDING',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_pick_wo FOREIGN KEY (work_order_id) REFERENCES work_order(id)
);
CREATE TABLE pick_line (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
pick_list_id BIGINT,
material_id BIGINT,
qty DECIMAL(18,6),
picked_qty DECIMAL(18,6) DEFAULT 0,
CONSTRAINT fk_pickline_pick FOREIGN KEY (pick_list_id) REFERENCES pick_list(id),
CONSTRAINT fk_pickline_material FOREIGN KEY (material_id) REFERENCES material(id)
);
-- 报工表
CREATE TABLE job_report (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
work_order_id BIGINT,
op_seq INT,
report_qty DECIMAL(18,6),
report_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
report_user VARCHAR(64),
is_ok BOOLEAN DEFAULT TRUE,
scrap_qty DECIMAL(18,6) DEFAULT 0,
CONSTRAINT fk_report_wo FOREIGN KEY (work_order_id) REFERENCES work_order(id)
);
-- 完工入库记录
CREATE TABLE finish_goods_receipt (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
work_order_id BIGINT,
material_id BIGINT,
qty DECIMAL(18,6),
receipt_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_fgr_wo FOREIGN KEY (work_order_id) REFERENCES work_order(id),
CONSTRAINT fk_fgr_material FOREIGN KEY (material_id) REFERENCES material(id)
);
8.2 后端参考：TypeScript + Express（核心 API）
下面是一个单文件风格的示例，展示工单创建、下达（生成领料单）、领料执行（扣库存）、报工与完工入库的流程。可作为项目骨架快速落地。
// server.ts - 简化示例 (需安装 express, sequelize, mysql2 等)
import express from 'express';
import bodyParser from 'body-parser';
import { Sequelize, DataTypes, Op } from 'sequelize';
// 初始化 Sequelize（示例用 MySQL）
const sequelize = new Sequelize('erp', 'user', 'pass', {
dialect: 'mysql',
host: 'localhost',
logging: false
});
// 定义模型（仅示例字段）
const Material = sequelize.define('material', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true },
code: DataTypes.STRING, name: DataTypes.STRING, uom: DataTypes.STRING
}, { timestamps: false });
const Inventory = sequelize.define('inventory', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true },
material_id: DataTypes.BIGINT, location_id: DataTypes.BIGINT, qty: DataTypes.DECIMAL(18,4), batch_no: DataTypes.STRING
}, { timestamps: false });
const WorkOrder = sequelize.define('work_order', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true },
wo_no: DataTypes.STRING, sales_order_no: DataTypes.STRING, material_id: DataTypes.BIGINT, qty: DataTypes.DECIMAL(18,6),
produced_qty: { type: DataTypes.DECIMAL(18,6), defaultValue: 0 }, status: DataTypes.STRING, release_date: DataTypes.DATE
}, { timestamps: false });
const PickList = sequelize.define('pick_list', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true }, pick_no: DataTypes.STRING, work_order_id: DataTypes.BIGINT, status: DataTypes.STRING
}, { timestamps: false });
const PickLine = sequelize.define('pick_line', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true }, pick_list_id: DataTypes.BIGINT, material_id: DataTypes.BIGINT, qty: DataTypes.DECIMAL(18,6), picked_qty: { type: DataTypes.DECIMAL(18,6), defaultValue: 0 }
}, { timestamps: false });
const JobReport = sequelize.define('job_report', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true }, work_order_id: DataTypes.BIGINT, op_seq: DataTypes.INTEGER,
report_qty: DataTypes.DECIMAL(18,6), report_time: DataTypes.DATE, report_user: DataTypes.STRING, is_ok: DataTypes.BOOLEAN, scrap_qty: DataTypes.DECIMAL(18,6)
}, { timestamps: false });
const FinishGoodsReceipt = sequelize.define('finish_goods_receipt', {
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true }, work_order_id: DataTypes.BIGINT, material_id: DataTypes.BIGINT, qty: DataTypes.DECIMAL(18,6), receipt_time: DataTypes.DATE
}, { timestamps: false });
// 关系（简化）
PickList.hasMany(PickLine, { foreignKey: 'pick_list_id' });
// Express App
const app = express();
app.use(bodyParser.json());
// 创建工单
app.post('/api/work-orders', async (req, res) => {
const { wo_no, sales_order_no, material_id, qty } = req.body;
const wo = await WorkOrder.create({ wo_no, sales_order_no, material_id, qty, status: 'DRAFT' });
return res.json(wo);
});
// 下达工单：生成领料单（按 BOM 简化：假设 bomLines 已知）
app.post('/api/work-orders/:id/release', async (req, res) => {
const id = req.params.id;
const t = await sequelize.transaction();
try {
const wo = await WorkOrder.findByPk(id, { transaction: t });
if (!wo) throw new Error('WO not found');
if (wo.get('status') !== 'DRAFT') throw new Error('WO status not allowed');
// 状态更新
await wo.update({ status: 'RELEASED', release_date: new Date() }, { transaction: t });
// 假设我们有 bomLines（真实系统从 bom 表查）
// 这里演示：一个 parent->多个 components
const bomLines = await sequelize.query('SELECT component_material_id, qty FROM bom WHERE parent_material_id = ?', { replacements: [wo.get('material_id')], transaction: t });
const pick = await PickList.create({ pick_no: `PICK-${Date.now()}`, work_order_id: wo.get('id'), status: 'PENDING' }, { transaction: t });
for (const bl of (bomLines as any)[0]) {
const needQty = parseFloat(bl.qty) * parseFloat(wo.get('qty') as any);
await PickLine.create({ pick_list_id: pick.get('id'), material_id: bl.component_material_id, qty: needQty }, { transaction: t });
}
await t.commit();
return res.json({ success: true, pick_no: pick.get('pick_no') });
} catch (err: any) {
await t.rollback();
return res.status(400).json({ error: err.message });
}
});
// 领料执行（扣库存）
app.post('/api/picklists/:id/execute', async (req, res) => {
const id = req.params.id;
// lines: [{ pickLineId, locationId, qty, batchNo }]
const { lines } = req.body;
const t = await sequelize.transaction();
try {
const pick = await PickList.findByPk(id, { include: [{ model: PickLine }], transaction: t });
if (!pick) throw new Error('Pick not found');
for (const l of lines) {
const pl = await PickLine.findByPk(l.pickLineId, { transaction: t });
if (!pl) throw new Error('PickLine not found');
const remain = parseFloat(pl.get('qty') as any) - parseFloat(pl.get('picked_qty') as any);
if (l.qty > remain + 1e-9) throw new Error('Pick qty exceed remain');
// 减库存（简单示例按 location 扣）
const inv = await Inventory.findOne({ where: { material_id: pl.get('material_id'), location_id: l.locationId }, transaction: t, lock: t.LOCK.UPDATE });
if (!inv || parseFloat(inv.get('qty') as any) < l.qty) throw new Error('库存不足');
await inv.update({ qty: (parseFloat(inv.get('qty') as any) - l.qty).toFixed(4) }, { transaction: t });
// 更新 pick line
await pl.update({ picked_qty: (parseFloat(pl.get('picked_qty') as any) + l.qty).toFixed(6) }, { transaction: t });
}
// 更新 pick 状态
const pending = await PickLine.count({ where: { pick_list_id: pick.get('id'), picked_qty: { [Op.lt]: sequelize.col('qty') } }, transaction: t });
await pick.update({ status: pending === 0 ? 'COMPLETED' : 'PARTIAL' }, { transaction: t });
await t.commit();
return res.json({ success: true });
} catch (err: any) {
await t.rollback();
return res.status(400).json({ error: err.message });
}
});
// 报工（含完工入库逻辑）
app.post('/api/work-orders/:id/report', async (req, res) => {
const id = req.params.id;
const { op_seq, report_qty, is_ok = true, scrap_qty = 0, user } = req.body;
const t = await sequelize.transaction();
try {
const wo = await WorkOrder.findByPk(id, { transaction: t, lock: t.LOCK.UPDATE });
if (!wo) throw new Error('WO not found');
const status = wo.get('status');
if (!['RELEASED', 'IN_PROGRESS'].includes(status as string)) throw new Error('WO status not allowed');
// 写报工
await JobReport.create({ work_order_id: id, op_seq, report_qty, report_time: new Date(), report_user: user, is_ok, scrap_qty }, { transaction: t });
// 更新生产数量
const newProduced = parseFloat(wo.get('produced_qty') as any) + parseFloat(report_qty);
let newStatus = 'IN_PROGRESS';
if (newProduced + 1e-9 >= parseFloat(wo.get('qty') as any)) {
newStatus = 'COMPLETED';
// 完工入库
await FinishGoodsReceipt.create({ work_order_id: id, material_id: wo.get('material_id'), qty: newProduced, receipt_time: new Date() }, { transaction: t });
// 增加库存到默认仓位 id=1（示例）
const defaultLoc = 1;
let inv = await Inventory.findOne({ where: { material_id: wo.get('material_id'), location_id: defaultLoc }, transaction: t, lock: t.LOCK.UPDATE });
if (!inv) inv = await Inventory.create({ material_id: wo.get('material_id'), location_id: defaultLoc, qty: 0 }, { transaction: t });
await inv.update({ qty: (parseFloat(inv.get('qty') as any) + newProduced).toFixed(4) }, { transaction: t });
}
await wo.update({ produced_qty: newProduced.toFixed(6), status: newStatus }, { transaction: t });
await t.commit();
return res.json({ success: true, status: newStatus });
} catch (err: any) {
await t.rollback();
return res.status(400).json({ error: err.message });
}
});
app.listen(3000, async () => {
console.log('Server listening on 3000');
try { await sequelize.authenticate(); console.log('DB connected'); } catch (e) { console.error(e); }
});
说明与落地提示：
- 生产环境需加幂等、重试、权限校验、日志、审计（操作人、终端、IP）等。
- 若系统拆分为微服务，建议库存服务负责库存事务并提供幂等出入库 API，生产服务只负责生成业务事件并监听库存结果（或采用 Saga 补偿）。
八、实现效果
交付时可以按以下维度验收并衡量效果：
- 功能验收：工单从 DRAFT→RELEASED→IN_PROGRESS→COMPLETED 流程完整；领料、退料、报工、完工入库链路完整且数据一致。
- 数据一致性：库存（系统记录）与实际盘点差异在可接受范围内；报工后完成数量与入库记录一致。
- 看板与执行：看板能实时展示工单状态、产线负载、异常告警；车间人员能通过扫码快速完成领料与报工。
- 性能验收：并发场景（高峰期）下关键接口延迟满足 SLO（例如 95% 请求 < 500ms）；批量统计通过预聚合或异步任务完成。
- 业务效果：交付准时率、在制品周期、领退料差异率等 KPI 达到预期提升值。
九、部署、运维与后续迭代建议
- CI/CD：代码走自动化构建、测试、分环境部署（dev/staging/prod）。
- 监控与告警：关键接口、库存扣减失败率、报工异常率、看板推送失败率都需要指标与告警。
- 数据备份与仓库同步：事务数据定期备份，生产业务日志同步到数据仓库（如 ClickHouse）用于 OLAP 报表。
- 后续迭代优先级建议： 完善 APS 集成与更智能排产； 接入 MES/PLC 实现自动报工与设备数据； 引入 WMS 做复杂库位/波次/交叉对接； 做 BI 仪表盘与预测分析（异常预测、短料预警）。
FAQ
FAQ 1：如何保证领料/退料与库存高度一致？
领料与退料直接影响库存可用量，稍有差池会导致排产错误或延迟。技术上建议把库存操作集中到库存服务，由库存服务提供幂等的出入库 API（请求带唯一操作号），内部用行级锁或乐观锁保证并发安全；若是分布式事务场景，使用 Saga 模式或事件补偿流程，确保在部分步骤失败时能回滚或补偿。业务上要求扫码确认领料、双人确认大批量领料、并在系统中保留领退料凭证与操作人信息。最后再配合定期库存差异分析和盘点，发现问题及时调整策略与制度。通过技术+流程两端保障，能把差异率降到可控水平。
FAQ 2：实时统计与系统性能如何平衡？
生产现场的数据读写量大，特别是高峰期的报工与领料会频繁触发数据库写操作。若把统计聚合放在事务里直接执行，会影响主业务性能。实践中常用异步架构：把业务操作写入事务表后，产生日志事件（Kafka/RabbitMQ）到统计服务，由统计服务异步更新物化视图或 OLAP 表，用 Redis 做热点缓存满足看板秒级展示需求。对于要求严格的实时看板，可以通过变更数据流（CDC）或基于事件的实时物化来实现近实时（秒级）更新。总之，主业务尽量轻量化，统计通过异步与缓存手段来支撑高并发查询。
FAQ 3：MTO 场景的排产策略有哪些实操建议？
MTO 以订单驱动，排产既要满足交期又要考虑产能与物料。实操上建议首先引入规则化优先级（如交期优先、同客户合并减少换线），对紧急单做单独标识并支持拆批生产；排产初期用简单启发式规则（FCFS + 优先级 + 产能约束）即可，待积累数据后再引入 APS 或优化算法（线性/整数规划、遗传算法等）。此外把物料约束（关键长料）纳入排产逻辑，若缺料应自动联动采购触发或把排产结果标为待料。最后，排产结果须和车间、仓库、采购做闭环确认，避免单边决策导致执行偏差。
