第2237篇：制造业AI工程——设备预测性维护系统的完整实现
2026/4/30大约 11 分钟
第2237篇：制造业AI工程——设备预测性维护系统的完整实现
适读人群：制造业技术团队、工业互联网工程师、Java后端开发者 | 阅读时长：约18分钟 | 核心价值：从真实工厂痛点出发，完整实现一套设备预测性维护系统，覆盖数据采集、特征工程、模型推理到告警闭环
去年我去一家汽车零部件工厂做技术调研，车间主任带我参观生产线的时候，正好碰上一台CNC加工中心停机。维修班长赶过来，拆开机器检查了半个小时，最后发现是一个轴承磨损过度导致振动异常，触发了保护停机。
车间主任跟我说，这台设备一停，后面三道工序全部等料，一个小时的损失大概是十几万。更让他头疼的是，这种停机完全没有预兆，是"突发性"的——虽然定期保养也做，但轴承的磨损状态根本看不出来，等到出问题已经晚了。
这就是制造业最典型的痛点：计划外停机。
行业数据显示，非计划停机占制造业生产损失的30%到50%，而其中大部分故障其实是有前兆的——只是人工巡检的频率和精度都不够，没能及时发现。预测性维护（Predictive Maintenance，PdM）就是要解决这个问题，在设备真正故障之前，提前发现异常信号。
预测性维护系统的整体架构
在开始写代码之前，先把问题想清楚。预测性维护系统需要回答三个核心问题：
- 设备当前状态是否正常？（异常检测）
- 如果异常，大概还能撑多久？（剩余寿命预测，RUL）
- 应该怎么处理？（维护决策建议）
系统架构如下：
这个架构有几个关键决策值得说：
为什么用双层过滤：纯规则告警会有大量误报，工人逐渐麻木；纯AI告警的可解释性差，维修人员不信任。两层结合：规则层做快速粗筛，AI层做精细判断。
为什么特征计算在流处理层：原始振动数据采样频率可能高达10kHz，直接存储成本高，也没必要。在流层做统计特征计算（RMS、峰值因子、频域特征），降采样后存储。
数据模型设计
设备数据的核心实体：
// 设备基础信息
@Entity
@Table(name = "equipment")
public class Equipment {
@Id
private String equipmentId;
private String name;
private String type; // CNC/PRESS/CONVEYOR等
private String location; // 产线位置
private LocalDateTime installDate;
private Integer designLifeHours;
@OneToMany(mappedBy = "equipment")
private List<Sensor> sensors;
}
// 传感器定义
@Entity
@Table(name = "sensor")
public class Sensor {
@Id
private String sensorId;
private String equipmentId;
private String sensorType; // VIBRATION/TEMPERATURE/CURRENT/PRESSURE
private String mountPosition; // 安装位置：轴承A/B侧、主轴等
private Double samplingRateHz;
private String unit;
}
// 实时特征数据（TimescaleDB超表）
@Entity
@Table(name = "sensor_features")
public class SensorFeature {
private String sensorId;
private Instant timestamp;
// 时域特征
private Double rms; // 均方根值
private Double peakValue; // 峰值
private Double crestFactor; // 峰值因子
private Double kurtosis; // 峭度
// 频域特征（FFT后的特征频段能量）
private Double lowFreqEnergy; // 0-100Hz
private Double midFreqEnergy; // 100-1000Hz
private Double highFreqEnergy; // 1000Hz以上
// 温度相关
private Double temperature;
private Double tempRiseRate; // 温升速率
}流处理层：实时特征计算
用Flink实现实时特征计算，处理振动传感器的原始数据：
@Component
public class VibrationFeatureJob {
@Autowired
private FlinkConfig flinkConfig;
public void startJob() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
// 从Kafka消费原始传感器数据
KafkaSource<RawSensorData> source = KafkaSource.<RawSensorData>builder()
.setBootstrapServers(flinkConfig.getBootstrapServers())
.setTopics("raw-sensor-data")
.setGroupId("feature-extractor")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new RawSensorDataDeserializer())
.build();
DataStream<RawSensorData> rawStream = env.fromSource(
source, WatermarkStrategy.noWatermarks(), "Kafka Source"
);
// 按传感器ID分组，开1秒滚动窗口计算特征
DataStream<SensorFeature> featureStream = rawStream
.keyBy(RawSensorData::getSensorId)
.window(TumblingEventTimeWindows.of(Time.seconds(1)))
.aggregate(new VibrationFeatureAggregator());
// 输出到TimescaleDB和Redis
featureStream.addSink(new TimescaleDBSink());
featureStream.addSink(new RedisSink());
env.execute("Vibration Feature Extraction");
}
}
// 振动特征聚合器
public class VibrationFeatureAggregator
implements AggregateFunction<RawSensorData, VibrationAccumulator, SensorFeature> {
@Override
public VibrationAccumulator createAccumulator() {
return new VibrationAccumulator();
}
@Override
public VibrationAccumulator add(RawSensorData data, VibrationAccumulator acc) {
double[] values = data.getValues();
acc.addSamples(values);
acc.setSensorId(data.getSensorId());
acc.setTimestamp(data.getTimestamp());
return acc;
}
@Override
public SensorFeature getResult(VibrationAccumulator acc) {
double[] samples = acc.getAllSamples();
SensorFeature feature = new SensorFeature();
feature.setSensorId(acc.getSensorId());
feature.setTimestamp(acc.getTimestamp());
// 计算时域特征
feature.setRms(calculateRMS(samples));
feature.setPeakValue(Arrays.stream(samples).map(Math::abs).max().orElse(0));
feature.setCrestFactor(feature.getPeakValue() / feature.getRms());
feature.setKurtosis(calculateKurtosis(samples));
// FFT计算频域特征
double[] fftResult = performFFT(samples);
feature.setLowFreqEnergy(calculateBandEnergy(fftResult, 0, 100, acc.getSamplingRate()));
feature.setMidFreqEnergy(calculateBandEnergy(fftResult, 100, 1000, acc.getSamplingRate()));
feature.setHighFreqEnergy(calculateBandEnergy(fftResult, 1000, 5000, acc.getSamplingRate()));
return feature;
}
private double calculateRMS(double[] samples) {
double sumSquares = 0;
for (double v : samples) sumSquares += v * v;
return Math.sqrt(sumSquares / samples.length);
}
private double calculateKurtosis(double[] samples) {
double mean = Arrays.stream(samples).average().orElse(0);
double variance = Arrays.stream(samples)
.map(v -> Math.pow(v - mean, 2)).average().orElse(0);
double stdDev = Math.sqrt(variance);
if (stdDev == 0) return 0;
double fourthMoment = Arrays.stream(samples)
.map(v -> Math.pow((v - mean) / stdDev, 4)).average().orElse(0);
return fourthMoment - 3; // 超峭度
}
// 简化FFT实现，实际项目用Apache Commons Math的FastFourierTransformer
private double[] performFFT(double[] samples) {
FastFourierTransformer fft = new FastFourierTransformer(DftNormalization.STANDARD);
// 补零到2的幂次
int n = Integer.highestOneBit(samples.length - 1) << 1;
double[] padded = Arrays.copyOf(samples, n);
Complex[] result = fft.transform(padded, TransformType.FORWARD);
return Arrays.stream(result).mapToDouble(Complex::abs).toArray();
}
private double calculateBandEnergy(double[] fftResult, double lowHz,
double highHz, double samplingRate) {
int lowBin = (int) (lowHz * fftResult.length / samplingRate);
int highBin = (int) (highHz * fftResult.length / samplingRate);
highBin = Math.min(highBin, fftResult.length / 2);
double energy = 0;
for (int i = lowBin; i < highBin; i++) {
energy += fftResult[i] * fftResult[i];
}
return energy;
}
@Override
public VibrationAccumulator merge(VibrationAccumulator a, VibrationAccumulator b) {
a.merge(b);
return a;
}
}推理服务：异常检测与剩余寿命预测
核心推理服务，集成了两个模型：
@Service
public class PredictiveMaintenanceService {
@Autowired
private AnomalyDetectionModel anomalyModel;
@Autowired
private RULPredictionModel rulModel;
@Autowired
private EquipmentRepository equipmentRepo;
@Autowired
private AlertService alertService;
/**
* 对一批特征数据进行推理，返回设备健康评估
*/
public HealthAssessment assess(String equipmentId, List<SensorFeature> recentFeatures) {
Equipment equipment = equipmentRepo.findById(equipmentId)
.orElseThrow(() -> new EquipmentNotFoundException(equipmentId));
// 构建特征向量（取最近30个时间窗口，即30秒）
double[][] featureMatrix = buildFeatureMatrix(recentFeatures);
// 异常检测：基于Isolation Forest或Autoencoder
AnomalyResult anomaly = anomalyModel.detect(featureMatrix);
HealthAssessment assessment = new HealthAssessment();
assessment.setEquipmentId(equipmentId);
assessment.setTimestamp(Instant.now());
assessment.setAnomalyScore(anomaly.getScore());
assessment.setAnomalyProbability(anomaly.getProbability());
// 只有异常概率超过阈值才做RUL预测（节省计算资源）
if (anomaly.getProbability() > 0.3) {
RULResult rul = rulModel.predict(featureMatrix, equipment);
assessment.setEstimatedRULHours(rul.getEstimatedHours());
assessment.setRulConfidenceInterval(rul.getConfidenceInterval());
assessment.setDegradationTrend(rul.getTrend());
// 生成告警
generateAlertIfNeeded(equipment, assessment);
}
// 计算健康指数（0-100）
assessment.setHealthIndex(calculateHealthIndex(anomaly, assessment));
return assessment;
}
private void generateAlertIfNeeded(Equipment equipment, HealthAssessment assessment) {
AlertLevel level = determineAlertLevel(assessment);
if (level == AlertLevel.NONE) return;
Alert alert = Alert.builder()
.equipmentId(equipment.getEquipmentId())
.equipmentName(equipment.getName())
.level(level)
.anomalyScore(assessment.getAnomalyScore())
.estimatedRULHours(assessment.getEstimatedRULHours())
.message(buildAlertMessage(equipment, assessment, level))
.recommendation(generateRecommendation(assessment))
.timestamp(Instant.now())
.build();
alertService.publish(alert);
}
private AlertLevel determineAlertLevel(HealthAssessment assessment) {
double prob = assessment.getAnomalyProbability();
Double rul = assessment.getEstimatedRULHours();
if (prob > 0.8 || (rul != null && rul < 24)) {
return AlertLevel.CRITICAL; // 24小时内可能故障
} else if (prob > 0.6 || (rul != null && rul < 72)) {
return AlertLevel.WARNING; // 72小时内需关注
} else if (prob > 0.3) {
return AlertLevel.INFO; // 开始出现退化趋势
}
return AlertLevel.NONE;
}
private String generateRecommendation(HealthAssessment assessment) {
// 根据特征模式判断可能的故障类型
if (assessment.getDegradationTrend() != null) {
switch (assessment.getDegradationTrend()) {
case BEARING_WEAR:
return "建议检查轴承状态，必要时更换轴承。重点检查润滑油脂是否充足。";
case IMBALANCE:
return "检测到不平衡特征，建议安排动平衡校正作业。";
case LOOSENESS:
return "检测到松动特征，建议检查紧固件和安装底座。";
default:
return "建议安排专业维修人员进行详细检查。";
}
}
return "设备状态出现异常，建议增加巡检频率。";
}
private double calculateHealthIndex(AnomalyResult anomaly, HealthAssessment assessment) {
double base = 100 - anomaly.getProbability() * 60;
if (assessment.getEstimatedRULHours() != null) {
// RUL越短，健康指数越低
double rulFactor = Math.min(assessment.getEstimatedRULHours() / 720.0, 1.0);
base = base * 0.7 + rulFactor * 30;
}
return Math.max(0, Math.min(100, base));
}
private double[][] buildFeatureMatrix(List<SensorFeature> features) {
// 取最近的特征，按时间排序，转为矩阵
return features.stream()
.sorted(Comparator.comparing(SensorFeature::getTimestamp))
.limit(30)
.map(f -> new double[]{
f.getRms(), f.getPeakValue(), f.getCrestFactor(),
f.getKurtosis(), f.getLowFreqEnergy(),
f.getMidFreqEnergy(), f.getHighFreqEnergy(),
f.getTemperature()
})
.toArray(double[][]::new);
}
}告警闭环：从告警到工单
告警生成后，需要自动推送到维修工单系统：
@Service
public class AlertService {
@Autowired
private AlertRepository alertRepo;
@Autowired
private MESIntegrationClient mesClient;
@Autowired
private DingTalkNotifier dingTalkNotifier;
// 告警抑制：同一设备5分钟内不重复告警
private final Cache<String, Long> alertSuppressCache =
CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).build();
public void publish(Alert alert) {
// 告警抑制检查
String cacheKey = alert.getEquipmentId() + "_" + alert.getLevel();
if (alertSuppressCache.getIfPresent(cacheKey) != null) {
return;
}
alertSuppressCache.put(cacheKey, System.currentTimeMillis());
// 保存告警记录
alertRepo.save(alert);
// 根据告警级别决定后续动作
switch (alert.getLevel()) {
case CRITICAL:
// 紧急：立即创建维修工单
createMaintenanceOrder(alert, Priority.URGENT);
dingTalkNotifier.sendCriticalAlert(alert);
break;
case WARNING:
// 警告：创建计划维修工单
createMaintenanceOrder(alert, Priority.HIGH);
dingTalkNotifier.sendWarningAlert(alert);
break;
case INFO:
// 信息：记录并汇总到日报
break;
}
}
private void createMaintenanceOrder(Alert alert, Priority priority) {
MaintenanceOrderRequest request = MaintenanceOrderRequest.builder()
.equipmentId(alert.getEquipmentId())
.equipmentName(alert.getEquipmentName())
.priority(priority)
.description(alert.getMessage())
.recommendation(alert.getRecommendation())
.expectedCompletionTime(calculateExpectedTime(alert, priority))
.build();
try {
MaintenanceOrderResponse response = mesClient.createOrder(request);
// 更新告警记录，关联工单号
alert.setWorkOrderId(response.getOrderId());
alertRepo.save(alert);
log.info("维修工单已创建: orderId={}, equipmentId={}",
response.getOrderId(), alert.getEquipmentId());
} catch (Exception e) {
log.error("创建维修工单失败: equipmentId={}", alert.getEquipmentId(), e);
// 工单创建失败时，升级为人工通知
dingTalkNotifier.sendEscalationAlert(alert, "工单系统异常，请人工处理");
}
}
}模型训练策略：解决标注数据稀缺问题
制造业AI最大的难题不是算法，是数据——正常数据很多，故障数据极少。我们采用了几种策略：
策略1：无监督异常检测先行
在没有足够故障标签时，先用Isolation Forest做基线。用正常运行时的数据训练，把偏离正常模式的样本识别为异常。
策略2：迁移学习
同类型设备（比如同型号CNC）之间有共性故障模式，可以用一台有丰富历史数据的设备训练基础模型，迁移到同类型的新设备上，只需少量数据做微调。
策略3：故障数据增强
对已有的少量故障数据，通过时间序列插值、添加受控噪声、时间拉伸/压缩等方式扩充训练集。
@Service
public class ModelTrainingService {
@Autowired
private SensorFeatureRepository featureRepo;
@Autowired
private ModelRegistry modelRegistry;
/**
* 定时重训练：每周执行一次，融合新数据
*/
@Scheduled(cron = "0 0 2 * * SUN") // 每周日凌晨2点
public void retrainModels() {
List<Equipment> equipments = equipmentRepo.findAll();
for (Equipment equipment : equipments) {
try {
retrainForEquipment(equipment);
} catch (Exception e) {
log.error("设备{}模型训练失败", equipment.getEquipmentId(), e);
}
}
}
private void retrainForEquipment(Equipment equipment) {
// 获取最近90天的特征数据
Instant start = Instant.now().minus(90, ChronoUnit.DAYS);
List<SensorFeature> features = featureRepo
.findByEquipmentIdAndTimestampAfter(equipment.getEquipmentId(), start);
if (features.size() < 1000) {
log.warn("设备{}数据量不足（{}条），跳过训练",
equipment.getEquipmentId(), features.size());
return;
}
// 转换为训练数据
TrainingDataset dataset = buildTrainingDataset(features);
// 调用Python训练服务（通过gRPC）
ModelTrainingRequest request = ModelTrainingRequest.newBuilder()
.setEquipmentId(equipment.getEquipmentId())
.setEquipmentType(equipment.getType())
.addAllFeatures(dataset.toProto())
.setEnableTransferLearning(true)
.build();
ModelTrainingResponse response = modelTrainingStub.train(request);
if (response.getSuccess()) {
// 验证新模型性能，只有提升才更新
if (response.getValidationScore() > getCurrentModelScore(equipment)) {
modelRegistry.updateModel(
equipment.getEquipmentId(),
response.getModelPath(),
response.getValidationScore()
);
log.info("设备{}模型已更新，验证分数: {}",
equipment.getEquipmentId(), response.getValidationScore());
}
}
}
}从数字到价值：效果衡量
这套系统上线后，我们在评估效果时遇到了一个很有意思的问题：如何量化"避免了什么"？
系统识别出了异常，提前安排了维护，设备没有停机，那到底省了多少钱？
我们的做法是建立一个对照组：把系统有告警但维护成本较低的情况（更换耗材）和历史上同类设备突发故障的情况做对比，计算平均避免停机时长和维修费用差。
运行6个月后的数据：
- 计划外停机次数下降67%
- 预防性更换轴承的平均费用 vs 带伤运行最终损坏主轴的维修费用，相差约30倍
- 虚报率（告警了但实际没故障）控制在15%以内
15%的虚报率是个关键指标。太高了，维修工人会失去信任；太低了，可能是阈值太松了，漏掉了真实告警。这个平衡需要在实际运行中持续调整。
工程落地的几个坑
坑1：传感器数据质量问题
工厂环境恶劣，传感器会断线、漂移、数据异常。必须在特征计算之前做严格的数据质量检查，包括缺失值检测、离群点过滤、时间戳连续性验证。
坑2：设备型号差异
即使是同型号设备，因为使用年限、加工物料不同，正常状态的特征分布可能差异很大。每台设备需要独立的基线，不能用同型号其他设备的基线代替。
坑3：季节和班次影响
夏天高温时设备温度特征会整体偏高，连续三班生产时特征和单班生产也不一样。模型要把这些环境因素作为协变量考虑进去，否则夏天会狂报温度告警。
坑4：维修人员的接受度
系统告警之后，维修班长如果认为设备"看起来没问题"，很可能不理会告警。系统落地不是技术问题，是信任建立过程。建议初期不要直接创建工单，而是先走人工确认流程，通过几次"告警准确预测了故障"的案例来建立信任。