Java物联网案例

wen java案例 2026-08-01 168

本文目录导读：

Java物联网案例

案例一：智能家居温度监控系统（基础版）
案例二：智能工厂设备监测系统（进阶版）
案例三：完整智慧农业解决方案（企业级）
案例四：车联网（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);

本文地址： https://bmjup.com/post/35448.html