MES（Manufacturing Execution System，制造执行系统）是连接企业ERP与车间控制层的核心系统，而数据采集是MES系统的基础功能。本文将从实际项目经验出发，详细解析工业MES系统开发中的数据采集架构设计，包括设备对接方案、数据模型设计、实时处理机制、存储优化策略等核心技术要点。
一、MES数据采集的整体架构
一个完整的MES数据采集系统通常包含以下几个层次：
- 设备层：PLC、CNC、传感器、仪表等工业设备
- 采集层：数据采集网关、边缘计算节点、OPC UA服务器
- 传输层：MQTT、Kafka、RabbitMQ等消息中间件
- 处理层：流式计算引擎（如Flink、Spark Streaming）
- 存储层：时序数据库、关系型数据库、数据湖
- 应用层：MES业务功能、数据可视化、报表分析
二、设备对接方案设计
工业现场设备种类繁多，协议各异，需要设计灵活的对接方案。
1. 协议适配层设计
采用适配器模式，为不同协议实现统一的接口：
// 设备适配器接口
public interface IDeviceAdapter
{
Task ConnectAsync(DeviceConfig config);
Task DisconnectAsync();
Task> ReadDataAsync(string[] tags);
Task WriteDataAsync(Dictionary values);
event EventHandler DataChanged;
}
// Modbus TCP适配器实现
public class ModbusTcpAdapter : IDeviceAdapter
{
private ModbusTcpClient _client;
public async Task ConnectAsync(DeviceConfig config)
{
_client = new ModbusTcpClient(config.IpAddress, config.Port);
await _client.ConnectAsync();
}
public async Task> ReadDataAsync(string[] tags)
{
var result = new Dictionary();
// 解析标签地址并读取数据
foreach (var tag in tags)
{
var address = ParseAddress(tag);
var value = await _client.ReadHoldingRegistersAsync(address.Start, address.Length);
result[tag] = ConvertValue(value, address.DataType);
}
return result;
}
}
// OPC UA适配器实现
public class OpcUaAdapter : IDeviceAdapter
{
private OpcUaClient _client;
public async Task ConnectAsync(DeviceConfig config)
{
_client = new OpcUaClient(config.EndpointUrl);
await _client.ConnectAsync();
}
public async Task> ReadDataAsync(string[] tags)
{
return await _client.ReadNodesAsync(tags);
}
} 2. 边缘计算节点
在设备密集的场景中，部署边缘计算节点进行本地数据预处理：
// 边缘计算节点配置
public class EdgeNode
{
private List _adapters;
private IDataProcessor _processor;
private IMessagePublisher _publisher;
public async Task StartAsync()
{
// 启动数据采集任务
foreach (var adapter in _adapters)
{
adapter.DataChanged += async (s, e) =>
{
// 本地预处理：滤波、聚合、异常检测
var processedData = await _processor.ProcessAsync(e.Data);
// 发布到消息队列
await _publisher.PublishAsync("mes/raw-data", processedData);
};
}
}
} 三、数据模型设计
MES系统的数据模型需要兼顾实时性和历史查询性能。
1. 实时数据模型
// 实时数据点
public class RealtimeDataPoint
{
public string DeviceId { get; set; } // 设备ID
public string TagName { get; set; } // 标签名
public object Value { get; set; } // 当前值
public DateTime Timestamp { get; set; } // 时间戳
public DataQuality Quality { get; set; } // 数据质量
// 计算属性
public bool IsAlarm => Quality == DataQuality.Bad;
}2. 历史数据模型
// 历史数据记录
public class HistoryDataRecord
{
public long Id { get; set; }
public string DeviceId { get; set; }
public string TagName { get; set; }
public double Value { get; set; }
public DateTime Timestamp { get; set; }
// 索引优化：按设备+时间建立复合索引
// CREATE INDEX idx_device_time ON history_data(device_id, timestamp DESC);
}3. 生产事件模型
// 生产事件（如报警、状态变化、质量异常）
public class ProductionEvent
{
public string EventId { get; set; }
public EventType Type { get; set; } // 事件类型
public string DeviceId { get; set; }
public string Description { get; set; }
public DateTime OccurTime { get; set; }
public DateTime? ResolveTime { get; set; }
public EventSeverity Severity { get; set; }
public Dictionary Context { get; set; } // 上下文数据
} 四、实时数据处理机制
MES系统需要对采集的数据进行实时处理，包括数据清洗、聚合计算、规则引擎等。
1. 流式处理架构
// 使用Flink进行流式处理
public class MesStreamProcessor
{
public void StartProcessing()
{
var env = StreamExecutionEnvironment.GetExecutionEnvironment();
// 从Kafka读取原始数据
var source = env.AddSource(new FlinkKafkaConsumer<>(
"mes/raw-data",
new RawDataDeserializationSchema(),
kafkaProps
));
// 数据清洗：过滤无效数据
var cleaned = source.filter(data => data.Quality == DataQuality.Good);
// 窗口聚合：按设备、按分钟聚合
var aggregated = cleaned
.keyBy(data => data.DeviceId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.apply(new AggregationFunction());
// 规则引擎：检测异常
var alarms = aggregated
.filter(data => CheckAlarmRules(data))
.map(data => GenerateAlarmEvent(data));
// 输出到不同目的地
aggregated.addSink(new JdbcSink<>(...)); // 写入数据库
alarms.addSink(new KafkaSink<>("mes/alarms")); // 发布报警
}
}2. 规则引擎实现
// 报警规则引擎
public class AlarmRuleEngine
{
private List _rules;
public AlarmRuleEngine()
{
_rules = LoadRulesFromConfig();
}
public List Evaluate(Dictionary data)
{
var alarms = new List();
foreach (var rule in _rules)
{
if (rule.IsMatch(data))
{
alarms.Add(new AlarmEvent
{
RuleId = rule.Id,
Severity = rule.Severity,
Message = rule.FormatMessage(data),
Timestamp = DateTime.Now
});
}
}
return alarms;
}
}
// 报警规则定义
public class AlarmRule
{
public string Id { get; set; }
public string TagName { get; set; }
public AlarmCondition Condition { get; set; } // 如：>、<、==、between
public double Threshold { get; set; }
public EventSeverity Severity { get; set; }
public bool IsMatch(Dictionary data)
{
if (!data.ContainsKey(TagName)) return false;
var value = Convert.ToDouble(data[TagName]);
return Condition switch
{
AlarmCondition.GreaterThan => value > Threshold,
AlarmCondition.LessThan => value < Threshold,
AlarmCondition.Between => value >= Threshold && value <= Threshold * 1.1,
_ => false
};
}
} 五、存储优化策略
MES系统的数据量巨大，需要合理的存储策略保证查询性能。
1. 时序数据库选型
对于高频采集的实时数据，使用时序数据库（如InfluxDB、TimescaleDB）：
// InfluxDB写入示例
public class InfluxDbWriter
{
private InfluxDBClient _client;
public async Task WriteBatchAsync(List dataPoints)
{
var points = dataPoints.Select(dp => PointData
.Measurement("device_data")
.Tag("device_id", dp.DeviceId)
.Tag("tag_name", dp.TagName)
.Field("value", Convert.ToDouble(dp.Value))
.Timestamp(dp.Timestamp, WritePrecision.Ms)
).ToList();
using var writeApi = _client.GetWriteApi();
await writeApi.WritePointsAsync("mes_db", "autogen", points);
}
} 2. 数据分层存储
// 数据生命周期管理
public class DataLifecycleManager
{
// 热数据：最近7天，存储在内存/SSD
// 温数据：7-90天，存储在SSD/HDD
// 冷数据：90天以上，归档到对象存储
public async Task ArchiveOldDataAsync()
{
var cutoffDate = DateTime.Now.AddDays(-90);
// 1. 从时序数据库导出冷数据
var coldData = await _influxDb.QueryAsync(
$"SELECT * FROM device_data WHERE time < '{cutoffDate}'"
);
// 2. 压缩后上传到对象存储
var compressed = CompressData(coldData);
await _ossClient.UploadAsync($"archive/{DateTime.Now:yyyyMMdd}.parquet", compressed);
// 3. 从时序数据库删除
await _influxDb.DeleteAsync($"time < '{cutoffDate}'");
}
}3. 查询优化
// 常用查询优化
public class DataQueryService
{
// 1. 使用预聚合表加速历史查询
public async Task> GetHourlyDataAsync(
string deviceId, DateTime startTime, DateTime endTime)
{
// 从预聚合表查询，而不是原始数据
return await _dbContext.HourlyAggregates
.Where(x => x.DeviceId == deviceId
&& x.Timestamp >= startTime
&& x.Timestamp <= endTime)
.OrderBy(x => x.Timestamp)
.ToListAsync();
}
// 2. 使用物化视图加速报表查询
public async Task GetDailyReportAsync(DateTime date)
{
return await _dbContext.DailyProductionReports
.Where(x => x.Date == date)
.FirstOrDefaultAsync();
}
} 六、实际应用案例
在某汽车零部件工厂的MES项目中，我们设计了完整的数据采集架构：
- 对接50台注塑机、20台CNC、10台检测设备
- 采集频率：关键参数100ms，一般参数1s
- 日数据量：约5000万条记录
- 使用InfluxDB存储时序数据，PostgreSQL存储业务数据
- 通过Flink进行实时报警检测，响应时间<500ms
- 数据查询性能：历史趋势查询<2s，报表生成<5s
七、总结
工业MES系统的数据采集架构设计需要综合考虑设备多样性、数据实时性、存储可扩展性等因素。通过协议适配层实现设备解耦，通过边缘计算降低网络压力，通过流式处理保证实时性，通过分层存储优化成本。在实际项目中，需要根据具体场景选择合适的技术栈，并通过性能测试验证架构的可行性。