本文目录导读：
- 案例一：智能家居温度监控系统（基础版）
- 案例二：智能工厂设备监测系统（进阶版）
- 案例三：完整智慧农业解决方案（企业级）
- 案例四：车联网（Vehicle-to-Everything, V2X）平台
- 案例五：工业物联网数据采集与分析系统
智能家居温度监控系统（基础版）
场景描述
通过传感器收集室内温度数据，通过MQTT协议传输到服务器，后端进行存储和监控,支持Web端实时查看和报警通知。
技术栈
- 设备端：Java + Eclipse Paho MQTT Client
- 通信协议：MQTT
- 后端：Spring Boot + MySQL + WebSocket
- 前端：Vue/React + ECharts
设备端代码示例
import org.eclipse.paho.client.mqttv3.*;
import java.util.Random;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class TemperatureSensor {
private static final String BROKER_URL = "tcp://localhost:1883";
private static final String TOPIC = "home/livingroom/temperature";
private static final String CLIENT_ID = "sensor_temperature_001";
private MqttClient client;
private Random random = new Random();
public void connect() throws MqttException {
client = new MqttClient(BROKER_URL, CLIENT_ID);
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setConnectionTimeout(10);
options.setKeepAliveInterval(30);
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
System.out.println("连接丢失，尝试重连...");
reconnect();
}
@Override
public void messageArrived(String topic, MqttMessage message) {
// 接收控制指令
System.out.println("收到指令: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// 发送完成
}
});
client.connect(options);
System.out.println("传感器连接成功");
// 订阅控制指令
client.subscribe("home/livingroom/control");
// 启动数据发送任务
startPublishing();
}
private void startPublishing() {
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
try {
if (client.isConnected()) {
double temperature = 20 + random.nextDouble() * 15; // 20-35度
String payload = String.format(
"{\"sensorId\":\"%s\",\"temperature\":%.2f,\"humidity\":%.2f,\"timestamp\":%d}",
CLIENT_ID,
temperature,
40 + random.nextDouble() * 40,
System.currentTimeMillis()
);
MqttMessage message = new MqttMessage(payload.getBytes());
message.setQos(1);
client.publish(TOPIC, message);
System.out.println("发布数据: " + payload);
}
} catch (MqttException e) {
e.printStackTrace();
}
}, 0, 10, TimeUnit.SECONDS); // 每10秒上报一次
}
private void reconnect() {
while (!client.isConnected()) {
try {
System.out.println("正在重连...");
client.connect();
System.out.println("重连成功");
} catch (MqttException e) {
try {
Thread.sleep(5000);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
}
public static void main(String[] args) {
TemperatureSensor sensor = new TemperatureSensor();
try {
sensor.connect();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
后端服务（Spring Boot）
@RestController
@RequestMapping("/api/iot")
public class IotDataController {
@Autowired
private IoTDataService iotDataService;
@Autowired
private SimpMessagingTemplate messagingTemplate;
// MQTT配置
@Service
public static class MqttSubscriber {
@Autowired
private IoTDataService iotDataService;
@Autowired
private SimpMessagingTemplate messagingTemplate;
@PostConstruct
public void init() {
try {
MqttClient client = new MqttClient("tcp://localhost:1883", "server_" + UUID.randomUUID());
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {}
@Override
public void messageArrived(String topic, MqttMessage message) {
// 处理接收到的传感器数据
String payload = new String(message.getPayload());
System.out.println("收到数据: " + payload);
// 解析数据
IoTData data = parseData(payload);
// 存储到数据库
iotDataService.save(data);
// 实时推送给Web端
messagingTemplate.convertAndSend("/topic/sensor-data", data);
// 温度超过阈值报警
if (data.getTemperature() > 30.0) {
Alert alert = new Alert("HIGH_TEMPERATURE",
"温度过高: " + data.getTemperature() + "°C", data);
messagingTemplate.convertAndSend("/topic/alerts", alert);
}
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {}
});
client.connect(options);
client.subscribe("home/+/temperature");
} catch (MqttException e) {
e.printStackTrace();
}
}
private IoTData parseData(String payload) {
// 使用Jackson或Gson解析JSON
ObjectMapper mapper = new ObjectMapper();
return mapper.readValue(payload, IoTData.class);
}
}
// 供前端查询历史数据
@GetMapping("/history")
public List<IoTData> getHistory(
@RequestParam(required = false) String sensorId,
@RequestParam String startTime,
@RequestParam String endTime) {
return iotDataService.findByTimeRange(sensorId, startTime, endTime);
}
}
智能工厂设备监测系统（进阶版）
场景描述
监测工厂设备的运行状态（振动、噪音、温度、运行时长等），通过MQTT实时上传数据，后端进行设备健康度分析、预测性维护，通过Web界面展示设备状态看板，并通过短信/邮件/钉钉发送告警通知。
架构图
[传感器设备] --MQTT--> [EMQX Broker] --订阅--> [业务处理服务] --存储--> [InfluxDB/MySQL]
| |
|---WebSocket---> [Web看板]
|
---告警通知---> [钉钉/邮件/短信]核心业务逻辑代码
@Service
public class DeviceMonitorService {
@Autowired
private DeviceRepository deviceRepository;
@Autowired
private AlertService alertService;
@Autowired
private InfluxDBClient influxDBClient;
// 接收设备数据并进行处理
public void processDeviceData(DeviceData deviceData) {
// 1. 校验数据
if (!validateDeviceData(deviceData)) {
log.warn("无效的设备数据: {}", deviceData);
return;
}
// 2. 存储时序数据到InfluxDB
saveToInfluxDB(deviceData);
// 3. 计算设备健康度
double healthScore = calculateHealthScore(deviceData);
deviceData.setHealthScore(healthScore);
// 4. 更新设备状态
updateDeviceStatus(deviceData.getDeviceId(), healthScore);
// 5. 判断是否需要告警
checkAndAlert(deviceData, healthScore);
}
// 计算健康度评分
private double calculateHealthScore(DeviceData data) {
double score = 100.0;
// 温度评分
if (data.getTemperature() > 85) {
score -= 30;
} else if (data.getTemperature() > 70) {
score -= 15;
}
// 振动评分
if (data.getVibration() > 10) {
score -= 40;
} else if (data.getVibration() > 5) {
score -= 20;
}
// 噪声评分
if (data.getNoise() > 120) {
score -= 20;
} else if (data.getNoise() > 90) {
score -= 10;
}
// 累计运行时间因素（设备老化）
if (data.getRunningHours() > 7200) { // 超过1年
score -= 5;
}
return Math.max(0, Math.min(100, score));
}
// 检查和触发告警
private void checkAndAlert(DeviceData data, double healthScore) {
AlertLevel level = null;
String message = null;
if (healthScore < 50) {
level = AlertLevel.CRITICAL;
message = String.format("设备%s严重异常，健康度仅%.1f%，需要立即检修",
data.getDeviceId(), healthScore);
} else if (healthScore < 70) {
level = AlertLevel.WARNING;
message = String.format("设备%s存在异常，健康度%.1f%，请关注",
data.getDeviceId(), healthScore);
}
if (level != null) {
// 检查是否已发送过相同告警（避免频繁发送）
if (!alertService.isDuplicatedAlert(data.getDeviceId(), level, 15)) {
alertService.sendAlert(data.getDeviceId(), level, message);
}
}
}
}
设备数据管理（InfluxDB集成）
@Configuration
public class InfluxDBConfig {
@Bean
public InfluxDBClient influxDBClient() {
InfluxDBClient client = InfluxDBClientFactory.create(
"http://localhost:8086",
"token".toCharArray(),
"myorg",
"devices"
);
return client;
}
}
@Repository
public class DeviceDataRepository {
@Autowired
private InfluxDBClient influxDBClient;
public void save(DeviceData data) {
String point = String.format(
"device_data,device_id=%s device_status=\"%s\",temperature=%.2f," +
"vibration=%.2f,noise=%.2f,health_score=%.1f,running_hours=%d %d",
data.getDeviceId(),
data.getStatus(),
data.getTemperature(),
data.getVibration(),
data.getNoise(),
data.getHealthScore(),
data.getRunningHours(),
data.getTimestamp() * 1000000L // 纳秒时间戳
);
try (WriteApi writeApi = influxDBClient.getWriteApi()) {
writeApi.writeRecord(point);
}
}
public List<DeviceData> queryHistory(String deviceId, String startTime, String endTime) {
String fluxQuery = String.format(
"from(bucket:\"devices\") \n" +
" |> range(start: %s, stop: %s) \n" +
" |> filter(fn: (r) => r._measurement == \"device_data\") \n" +
" |> filter(fn: (r) => r.device_id == \"%s\")",
startTime, endTime, deviceId
);
// 执行查询并转换为DeviceData对象
List<FluxTable> tables = influxDBClient.getQueryApi().query(fluxQuery);
return parseTables(tables);
}
}
完整智慧农业解决方案（企业级）
场景描述
智能大棚种植管理系统，实现光照、温湿度、土壤pH值、CO₂浓度监测，自动控制灌溉、通风、补光等设备，通过AI算法实现智能决策,提供App远程控制。
系统架构
┌─────────────────────────────────────────────────────────────┐
│ 客户端应用层 │
│ [微信小程序] [Android/iOS App] [Web管理后台] │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ 业务服务层 │
│ 用户管理 设备管理 数据服务 告警管理 │
│ 决策引擎 报表统计 远程控制 系统配置 │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ 基础设施层 │
│ MQTT Broker (EMQX) Kafka消息队列 Redis缓存 │
│ MySQL/PostgreSQL InfluxDB时序数据库 MinIO对象存储 │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ 设备接入层 │
│ 智能网关 摄像头 传感器节点 PLC控制器 │
│ 气象站 土壤墒情仪 虫情监测 水肥一体机 │
└─────────────────────────────────────────────────────────────┘设备接入服务
@Component
public class DeviceGatewayService {
private static final String TOPIC_PREFIX = "agri/device/";
@Autowired
private DeviceRespository deviceRespository;
@Autowired
private CommandService commandService;
private MqttClient mqttClient;
private Map<String, DeviceSession> activeSessions = new ConcurrentHashMap<>();
@PostConstruct
public void init() {
try {
mqttClient = new MqttClient("tcp://localhost:1883",
"agri_server_" + UUID.randomUUID());
MqttConnectOptions options = new MqttConnectOptions();
options.setAutomaticReconnect(true);
options.setCleanSession(true);
options.setConnectionTimeout(30);
options.setKeepAliveInterval(60);
mqttClient.connect(options);
// 订阅设备上报主题
mqttClient.subscribe(TOPIC_PREFIX + "+/+/data", 1);
mqttClient.subscribe(TOPIC_PREFIX + "+/+/status", 1);
mqttClient.subscribe(TOPIC_PREFIX + "+/+/event", 1);
// 注册回调
mqttClient.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
// 自动重连由MqttConnectOptions配置
}
@Override
public void messageArrived(String topic, MqttMessage message) {
handleDeviceMessage(topic, message);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {}
});
} catch (MqttException e) {
log.error("MQTT连接失败", e);
}
}
private void handleDeviceMessage(String topic, MqttMessage message) {
// 解析主题：agri/device/{deviceId}/{type}
String[] parts = topic.split("/");
if (parts.length < 4) {
log.warn("无效的MQTT主题: {}", topic);
return;
}
String deviceId = parts[2];
String messageType = parts[3];
String payload = new String(message.getPayload());
switch (messageType) {
case "data":
handleDeviceData(deviceId, payload);
break;
case "status":
handleDeviceStatus(deviceId, payload);
break;
case "event":
handleDeviceEvent(deviceId, payload);
break;
default:
log.warn("不支持的消息类型: {}", messageType);
}
}
}
智能决策引擎
@Service
public class IntelligentDecisionEngine {
@Autowired
private EnvDataService envDataService;
@Autowired
private DeviceControlService controlService;
@Autowired
private RuleEngine ruleEngine;
@Autowired
private MLModelService mlModelService;
/**
* 智能灌溉决策
* 综合土壤湿度、天气预测、作物生长阶段等多因素
*/
public IrrigationDecision decideIrrigation(String greenhouseId) {
// 获取当前环境数据
EnvData currentEnv = envDataService.getLatestData(greenhouseId);
// 获取作物信息
CropInfo crop = cropService.getCropByGreenhouse(greenhouseId);
// 获取天气预报
WeatherForecast forecast = weatherService.getForecast(greenhouseId);
// 获取历史灌水记录
List<IrrigationRecord> history = recordService.getRecentIrrigations(greenhouseId, 48);
// 方案1：基于规则的决策
RuleBasedResult ruleResult = ruleEngine.evaluateIrrigationRule(
currentEnv, crop, forecast
);
// 方案2：基于机器学习的预测模型
double soilMoisturePrediction = mlModelService.predictSoilMoisture(
currentEnv, history, forecast
);
// 综合决策
IrrigationDecision decision = new IrrigationDecision();
decision.setGreenhouseId(greenhouseId);
decision.setDecisionTime(new Date());
// 判断是否需要灌溉
if (ruleResult.isNeedIrrigation() || soilMoisturePrediction < crop.getMinSoilMoisture()) {
decision.setNeedIrrigation(true);
// 计算灌溉量
double irrigationAmount = calculateIrrigationAmount(
currentEnv, crop, ruleResult, soilMoisturePrediction
);
decision.setAmount(irrigationAmount);
decision.setTime(estimateIrrigationTime(irrigationAmount));
decision.setReason("土壤湿度不足，预测未来24小时无降雨");
// 执行灌溉
controlService.sendIrrigationCommand(greenhouseId, irrigationAmount);
} else {
decision.setNeedIrrigation(false);
decision.setReason("当前土壤湿度适宜，无需灌溉");
}
// 异步执行其他决策
CompletableFuture.runAsync(() -> {
// 通风决策
decideVentilation(greenhouseId, currentEnv, crop);
// 补光决策
decideSupplementaryLighting(greenhouseId, currentEnv, crop, forecast);
// 施肥决策
decideFertilization(greenhouseId, currentEnv, crop);
});
// 记录决策日志
logService.recordDecision(decision);
return decision;
}
/**
* 计算灌溉量（考虑作物类型、生长阶段、蒸腾速率等）
*/
private double calculateIrrigationAmount(EnvData env, CropInfo crop,
RuleBasedResult ruleResult,
double predictedMoisture) {
double baseAmount = crop.getIrrigationBaseAmount(); // 基础灌溉量
double coeffTomp = 1 + (env.getTemperature() - 25) * 0.02; // 温度系数
double coeffHumidity = 1 - (env.getHumidity() - 50) * 0.005; // 湿度系数
double coeffGrowth = crop.getGrowthStageCoefficient(); // 生长阶段系数
// 考虑近期降雨
double rainCoeff = ruleResult.getExpectedRainfall() > 10 ? 0.7 : 1.0;
return baseAmount * coeffTomp * coeffHumidity * coeffGrowth * rainCoeff;
}
}
设备控制服务
@Service
public class DeviceControlService {
@Autowired
private MqttGateway mqttGateway;
@Autowired
private CommandHistoryRepository historyRepo;
/**
* 发送灌溉命令（带状态确认）
*/
public asyncTask sendIrrigationCommand(String greenhouseId, double amount) {
// 创建命令
DeviceCommand command = new DeviceCommand();
command.setDeviceId(greenhouseId);
command.setCommandType(CommandType.IRRIGATION_START);
command.setParameters(Map.of("amount", amount, "unit", "L"));
command.setId(UUID.randomUUID().toString());
// 设置超时与重试
command.setTimeout(30);
command.setRetryTimes(3);
// 通过MQTT发送
mqttGateway.sendCommand(command);
// 记录命令历史
historyRepo.save(command);
// 等待设备响应（异步）
CompletableFuture<CommandResult> future = new CompletableFuture<>();
// 将future存入pendingCommands
pendingCommands.put(command.getId(), future);
return future.get(30, TimeUnit.SECONDS)
.whenComplete((result, error) -> {
if (error != null) {
log.error("命令执行失败", error);
// 触发重试或告警
} else {
log.info("命令执行成功: {}", result);
// 更新状态
}
});
}
/**
* 批量设备控制
*/
public void batchControl(List<DeviceCommand> commands) {
// 使用协程或并行流批量处理
commands.parallelStream().forEach(this::sendCommand);
}
/**
* 定时任务 - 轮询设备状态（心跳检测）
*/
@Scheduled(fixedDelay = 5000)
public void checkDeviceHealth() {
// 检查超过10分钟未上报的在线设备
List<Device> staleDevices = deviceService.getDevicesByStatus(DeviceStatus.ONLINE);
for (Device device : staleDevices) {
if (device.getLastReportTime().before(new Date(System.currentTimeMillis() - 600000))) {
// 设备可能离线，发送状态查询
DeviceCommand queryCommand = DeviceCommand.builder()
.deviceId(device.getId())
.commandType(CommandType.STATUS_QUERY)
.build();
sendCommand(queryCommand);
}
}
}
}
数据分析与预测模型
# Python端训练模型并导出（与Java配合）
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
import joblib
# 读取历史环境数据
env_data = pd.read_csv('env_data.csv', parse_dates=['timestamp'])
crop_data = pd.read_csv('crop_data.csv')
# 特征工程
features = env_data[['temperature', 'humidity', 'light', 'soil_moisture']]
target = env_data['irrigation_amount'].shift(-1) # 预测下一次灌溉量
# 训练模型
model = RandomForestRegressor(n_estimators=100, random_state=42)
model.fit(features, target)
# 保存模型
joblib.dump(model, 'irrigation_model.joblib')
// Java调用Python模型推理
@Service
public class MLModelService {
@Autowired
private HttpClient httpClient;
private static final String MODEL_SERVER_URL = "http://localhost:5001/predict";
public double predictSoilMoisture(EnvData currentEnv,
List<IrrigationRecord> history,
WeatherForecast forecast) {
// 构建特征向量
Map<String, Double> features = new HashMap<>();
features.put("temperature", currentEnv.getTemperature());
features.put("humidity", currentEnv.getHumidity());
features.put("light", currentEnv.getLight());
features.put("soil_moisture", currentEnv.getSoilMoisture());
features.put("hours_since_last_irrigation",
history.isEmpty() ? 24 : calcHoursSince(history.get(0)));
features.put("forecast_rain_probability", forecast.getRainProbability());
features.put("evapotranspiration", calcET(currentEnv));
// 调用模型服务
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(MODEL_SERVER_URL))
.POST(HttpRequest.BodyPublishers.ofString(JsonUtil.toJson(features)))
.build();
HttpResponse<String> response = httpClient.send(request, BodyHandlers.ofString());
PredictionResult result = JsonUtil.fromJson(response.body(), PredictionResult.class);
return result.getPrediction();
}
}
车联网（Vehicle-to-Everything, V2X）平台
场景描述
车辆实时定位追踪、行驶数据采集（车速、油耗、里程）、驾驶行为分析（急加速/急刹车）、远程控制（远程锁车/远程启动）和故障诊断。
设备侧模拟代码
import org.eclipse.paho.client.mqttv3.*;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
public class VehicleDevice {
private static final String BROKER_URL = "tcp://localhost:1883";
private static final String VIN = "LSVAM4187C2184847";
private MqttClient mqttClient;
private Random random = new Random();
private AtomicInteger speed = new AtomicInteger(0);
private AtomicInteger fuelLevel = new AtomicInteger(80);
public void start() throws MqttException {
mqttClient = new MqttClient(BROKER_URL, "veh_" + VIN);
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setAutomaticReconnect(true);
mqttClient.connect(options);
// 订阅远程控制指令
subscribeToCommands();
// 启动数据上报
startReportThread();
}
private void subscribeToCommands() throws MqttException {
mqttClient.subscribe("v2x/vehicle/" + VIN + "/command", (topic, message) -> {
String command = new String(message.getPayload());
handleCommand(command);
});
}
private void handleCommand(String command) {
JsonNode node = new ObjectMapper().readTree(command);
String cmdType = node.get("type").asText();
switch (cmdType) {
case "lock":
// 执行锁车
System.out.println("车辆被远程锁定");
publishStatus("locked", "success");
break;
case "unlock":
System.out.println("车辆被远程解锁");
publishStatus("unlocked", "success");
break;
case "start_engine":
System.out.println("远程启动引擎");
publishStatus("engine_started", "success");
break;
case "diagnostic":
// 返回诊断信息
publishDiagnostics();
break;
}
}
private void startReportThread() {
Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
try {
// 模拟车辆行驶数据
speed.set(0 + random.nextInt(120)); // 0-120km/h
double longitude = 121.4737 + (random.nextDouble() - 0.5) * 0.01;
double latitude = 31.2304 + (random.nextDouble() - 0.5) * 0.01;
// 构建GPS数据
GpsData gpsData = new GpsData();
gpsData.setVin(VIN);
gpsData.setLongitude(longitude);
gpsData.setLatitude(latitude);
gpsData.setSpeed(speed.get());
gpsData.setFuelLevel(fuelLevel.getAndDecrement());
gpsData.setEngineTemp(80 + random.nextInt(30));
gpsData.setBatteryVoltage(12.5 + random.nextDouble() * 2);
gpsData.setTimestamp(System.currentTimeMillis());
// 发布到主题
publishToTopic("v2x/vehicle/" + VIN + "/gps", JsonUtil.toJson(gpsData));
// 检测驾驶行为
if (random.nextInt(100) == 0) { // 1%概率触发急加速
publishEvent("v2x/vehicle/" + VIN + "/event",
"harsh_acceleration",
"急加速: 0-60km/h in 2.5s");
}
// 检测故障
if (fuelLevel.get() < 10) {
publishEvent("v2x/vehicle/" + VIN + "/event",
"low_fuel",
"燃油不足: " + fuelLevel.get() + "%");
}
} catch (Exception e) {
e.printStackTrace();
}
}, 0, 5, TimeUnit.SECONDS);
}
private void publishStatus(String status, String result) {
publishToTopic("v2x/vehicle/" + VIN + "/status",
String.format("{\"status\":\"%s\",\"result\":\"%s\",\"time\":%d}",
status, result, System.currentTimeMillis()));
}
private void publishEvent(String topic, String eventType, String message) {
String payload = String.format(
"{\"eventType\":\"%s\",\"message\":\"%s\",\"time\":%d}",
eventType, message, System.currentTimeMillis()
);
publishToTopic(topic, payload);
}
private void publishDiagnostics() {
Map<String, String> diags = Map.of(
"engine_health", "good",
"tire_pressure", "normal",
"battery", "charging",
"errors", "no errors found"
);
publishToTopic("v2x/vehicle/" + VIN + "/diagnostics", JsonUtil.toJson(diags));
}
private void publishToTopic(String topic, String payload) throws MqttException {
MqttMessage message = new MqttMessage(payload.getBytes());
message.setQos(1);
message.setRetained(false);
mqttClient.publish(topic, message);
}
public static void main(String[] args) {
VehicleDevice vehicle = new VehicleDevice();
try {
vehicle.start();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
工业物联网数据采集与分析系统
场景描述
从PLC、仪表、仪器等工业设备中采集数据，通过Modbus、OPC-UA等协议接入，将数据发送到MQTT Broker，后端进行边缘计算、异常检测和数据可视化。
核心配置与代码
@Configuration
@EnableScheduling
public class IotDataCollector {
@Autowired
private MqttPublisher mqttPublisher;
// Modbus TCP客户端配置
private ModbusMaster getModbusMaster(String host, int port) {
IpParameters params = new IpParameters();
params.setHost(host);
params.setPort(port);
return new ModbusMaster(params);
}
/**
* 采集PLC数据任务
*/
@Scheduled(fixedDelay = 500) // 每500ms采集一次
public void collectPLCData() throws Exception {
List<String> deviceIPs = Arrays.asList("192.168.1.100", "192.168.1.101", "192.168.1.102");
for (String ip : deviceIPs) {
ModbusMaster plc = getModbusMaster(ip, 502);
plc.init();
try {
// 读取寄存器数据
ReadInputRegistersResponse response = plc.readInputRegisters(1, 0, 10);
// 提取关键参数
int[] values = response.getRegisters();
// 构建数据对象
PLCData data = new PLCData();
data.setDeviceIp(ip);
data.setPower(values[0]);
data.setSpeed(values[1]);
data.setTemperature(values[2] / 100.0); // 精确到0.01度
data.setPressure(values[3] / 10.0);
data.setFlowRate(values[4]);
data.setQuality(generateQualityCode());
// 发布到MQTT
mqttPublisher.publish("factory/plc/" + ip, data, 1);
} finally {
plc.disconnect();
}
}
}
/**
* 边缘计算 - 简单的异常检测
*/
private boolean anomalyDetection(PLCData data) {
// 温度报警阈值
if (data.getTemperature() > 85) {
log.error("设备温度过高: {}", data);
return true;
}
// 速率变化检测（判断速率突变）
if (calculateSpeedChangeRate(data) > 0.3) {
return true;
}
// 压力异常
if (data.getPressure() < 0.1 || data.getPressure() > 5.0) {
return true;
}
return false;
}
private double calculateSpeedChangeRate(PLCData current) {
// 使用Redis存储历史数据
String key = "plc:speed:" + current.getDeviceIp();
Double previousSpeed = redisTemplate.opsForValue().get(key);
if (previousSpeed != null && previousSpeed != 0) {
return Math.abs(current.getSpeed() - previousSpeed) / previousSpeed;
}
// 更新缓存
redisTemplate.opsForValue().set(key, current.getSpeed(), Duration.ofSeconds(5));
return 0;
}
}
Kafka集成（处理海量数据）
@Configuration
public class KafkaConfig {
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "iot-data-processor");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return new DefaultKafkaConsumerFactory<>(props,
new StringDeserializer(),
new StringDeserializer());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(5); // 并发消费线程数
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
}
@Component
public class KafkaIoTDataProcessor {
@Autowired
private InfluxDBClient influxDBClient;
@Autowired
private AlertService alertService;
/**
* 消费来自Kafka的IoT数据
*/
@KafkaListener(topics = "iot-raw-data",
concurrency = "5",
containerFactory = "kafkaListenerContainerFactory")
public void processRawData(String message) throws JsonProcessingException {
ObjectMapper mapper = new ObjectMapper();
RawData data = mapper.readValue(message, RawData.class);
// 1. 数据校验与清洗
if (!validate(data)) {
log.warn("数据校验失败: {}", message);
return;
}
// 2. 数据转换
TimeSeriesData tsData = convertToTimeSeries(data);
// 3. 批量写入InfluxDB（提高性能）
writeToInfluxDB(tsData);
