google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
IoT Sensor Data Processing in Java: A Complete Guide for Real-time Data Handling
awiskar acharya
November 23, 2025
3:04 pm

IoT (Internet of Things) sensor data processing involves collecting, analyzing, and acting upon data from connected devices. This comprehensive guide covers the entire pipeline from data ingestion to real-time analytics.

IoT Architecture Overview

Key Components:

  • Sensors/Devices: Data generators (temperature, humidity, motion, etc.)
  • Data Ingestion: Message brokers and protocols
  • Processing Engine: Real-time and batch processing
  • Storage: Time-series databases and data lakes
  • Analytics: Real-time monitoring and machine learning
  • Visualization: Dashboards and alerts

Dependencies and Setup

Maven Dependencies
<properties> <spring-boot.version>3.1.0</spring-boot.version> <kafka.version>3.4.0</kafka.version> <influxdb.version>2.7.0</influxdb.version> <mosquitto.version>1.2.5</mosquitto.version> <jackson.version>2.15.2</jackson.version> </properties> <dependencies> <!-- Spring Boot --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <version>${spring-boot.version}</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> <version>${spring-boot.version}</version> </dependency> <!-- Message Brokers --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>${kafka.version}</version> </dependency> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>${mosquitto.version}</version> </dependency> <!-- Databases --> <dependency> <groupId>com.influxdb</groupId> <artifactId>influxdb-client-java</artifactId> <version>${influxdb.version}</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb</artifactId> <version>${spring-boot.version}</version> </dependency> <!-- Utilities --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>${jackson.version}</version> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-math3</artifactId> <version>3.6.1</version> </dependency> <!-- Testing --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <version>${spring-boot.version}</version> <scope>test</scope> </dependency> </dependencies>

Core Data Models

1. Sensor Data Models
import com.fasterxml.jackson.annotation.JsonFormat; import java.time.Instant; import java.util.Map; public class SensorData { private String deviceId; private String sensorType; private double value; private String unit; private Map<String, Object> metadata; @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timezone = "UTC") private Instant timestamp; private Location location; // Constructors public SensorData() {} public SensorData(String deviceId, String sensorType, double value, String unit, Instant timestamp) { this.deviceId = deviceId; this.sensorType = sensorType; this.value = value; this.unit = unit; this.timestamp = timestamp; } // Getters and setters public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getSensorType() { return sensorType; } public void setSensorType(String sensorType) { this.sensorType = sensorType; } public double getValue() { return value; } public void setValue(double value) { this.value = value; } public String getUnit() { return unit; } public void setUnit(String unit) { this.unit = unit; } public Map<String, Object> getMetadata() { return metadata; } public void setMetadata(Map<String, Object> metadata) { this.metadata = metadata; } public Instant getTimestamp() { return timestamp; } public void setTimestamp(Instant timestamp) { this.timestamp = timestamp; } public Location getLocation() { return location; } public void setLocation(Location location) { this.location = location; } @Override public String toString() { return String.format("SensorData{deviceId='%s', type='%s', value=%.2f, unit='%s', timestamp=%s}", deviceId, sensorType, value, unit, timestamp); } } public class Location { private double latitude; private double longitude; private Double altitude; public Location() {} public Location(double latitude, double longitude) { this.latitude = latitude; this.longitude = longitude; } public Location(double latitude, double longitude, Double altitude) { this.latitude = latitude; this.longitude = longitude; this.altitude = altitude; } // Getters and setters public double getLatitude() { return latitude; } public void setLatitude(double latitude) { this.latitude = latitude; } public double getLongitude() { return longitude; } public void setLongitude(double longitude) { this.longitude = longitude; } public Double getAltitude() { return altitude; } public void setAltitude(Double altitude) { this.altitude = altitude; } } public class MultiSensorData { private String deviceId; private Instant timestamp; private Map<String, SensorReading> readings; private DeviceStatus status; private Location location; // Getters and setters public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public Instant getTimestamp() { return timestamp; } public void setTimestamp(Instant timestamp) { this.timestamp = timestamp; } public Map<String, SensorReading> getReadings() { return readings; } public void setReadings(Map<String, SensorReading> readings) { this.readings = readings; } public DeviceStatus getStatus() { return status; } public void setStatus(DeviceStatus status) { this.status = status; } public Location getLocation() { return location; } public void setLocation(Location location) { this.location = location; } } public class SensorReading { private double value; private String unit; private Double minValue; private Double maxValue; private SensorQuality quality; // Getters and setters public double getValue() { return value; } public void setValue(double value) { this.value = value; } public String getUnit() { return unit; } public void setUnit(String unit) { this.unit = unit; } public Double getMinValue() { return minValue; } public void setMinValue(Double minValue) { this.minValue = minValue; } public Double getMaxValue() { return maxValue; } public void setMaxValue(Double maxValue) { this.maxValue = maxValue; } public SensorQuality getQuality() { return SensorQuality.GOOD; } public void setQuality(SensorQuality quality) { this.quality = quality; } } public enum SensorQuality { EXCELLENT, GOOD, FAIR, POOR, ERROR } public enum DeviceStatus { ONLINE, OFFLINE, DEGRADED, MAINTENANCE }
2. Device Management Models
public class IoTDevice { private String deviceId; private String deviceName; private String deviceType; private String manufacturer; private String firmwareVersion; private Location installationLocation; private Map<String, String> specifications; private DeviceStatus status; private Instant lastSeen; private Map<String, SensorConfig> sensorConfigs; // Getters and setters public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getDeviceName() { return deviceName; } public void setDeviceName(String deviceName) { this.deviceName = deviceName; } public String getDeviceType() { return deviceType; } public void setDeviceType(String deviceType) { this.deviceType = deviceType; } public String getManufacturer() { return manufacturer; } public void setManufacturer(String manufacturer) { this.manufacturer = manufacturer; } public String getFirmwareVersion() { return firmwareVersion; } public void setFirmwareVersion(String firmwareVersion) { this.firmwareVersion = firmwareVersion; } public Location getInstallationLocation() { return installationLocation; } public void setInstallationLocation(Location installationLocation) { this.installationLocation = installationLocation; } public Map<String, String> getSpecifications() { return specifications; } public void setSpecifications(Map<String, String> specifications) { this.specifications = specifications; } public DeviceStatus getStatus() { return status; } public void setStatus(DeviceStatus status) { this.status = status; } public Instant getLastSeen() { return lastSeen; } public void setLastSeen(Instant lastSeen) { this.lastSeen = lastSeen; } public Map<String, SensorConfig> getSensorConfigs() { return sensorConfigs; } public void setSensorConfigs(Map<String, SensorConfig> sensorConfigs) { this.sensorConfigs = sensorConfigs; } } public class SensorConfig { private String sensorType; private double samplingRate; // Hz private Double minThreshold; private Double maxThreshold; private String unit; private int retentionDays; private boolean enabled; // Getters and setters public String getSensorType() { return sensorType; } public void setSensorType(String sensorType) { this.sensorType = sensorType; } public double getSamplingRate() { return samplingRate; } public void setSamplingRate(double samplingRate) { this.samplingRate = samplingRate; } public Double getMinThreshold() { return minThreshold; } public void setMinThreshold(Double minThreshold) { this.minThreshold = minThreshold; } public Double getMaxThreshold() { return maxThreshold; } public void setMaxThreshold(Double maxThreshold) { this.maxThreshold = maxThreshold; } public String getUnit() { return unit; } public void setUnit(String unit) { this.unit = unit; } public int getRetentionDays() { return retentionDays; } public void setRetentionDays(int retentionDays) { this.retentionDays = retentionDays; } public boolean isEnabled() { return enabled; } public void setEnabled(boolean enabled) { this.enabled = enabled; } }

Data Ingestion Layer

1. MQTT Client for Sensor Data
import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.stereotype.Component; import com.fasterxml.jackson.databind.ObjectMapper; @Component public class MqttSensorClient implements MqttCallback { private MqttClient mqttClient; private final ObjectMapper objectMapper; private final SensorDataProcessor dataProcessor; private final String brokerUrl; private final String clientId; public MqttSensorClient(ObjectMapper objectMapper, SensorDataProcessor dataProcessor, @Value("${mqtt.broker.url}") String brokerUrl) { this.objectMapper = objectMapper; this.dataProcessor = dataProcessor; this.brokerUrl = brokerUrl; this.clientId = "sensor-processor-" + System.currentTimeMillis(); } public void connect() throws MqttException { MqttConnectOptions options = new MqttConnectOptions(); options.setAutomaticReconnect(true); options.setCleanSession(true); options.setConnectionTimeout(10); options.setKeepAliveInterval(60); mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence()); mqttClient.setCallback(this); mqttClient.connect(options); // Subscribe to sensor topics subscribeToTopics(); } private void subscribeToTopics() throws MqttException { String[] topics = { "sensors/+/temperature", "sensors/+/humidity", "sensors/+/pressure", "sensors/+/multi", "devices/+/status" }; int[] qos = {1, 1, 1, 1, 1}; // Quality of Service level 1 mqttClient.subscribe(topics, qos); } @Override public void messageArrived(String topic, MqttMessage message) throws Exception { String payload = new String(message.getPayload()); try { if (topic.contains("/multi")) { // Handle multi-sensor data MultiSensorData multiData = objectMapper.readValue(payload, MultiSensorData.class); dataProcessor.processMultiSensorData(multiData); } else if (topic.contains("/status")) { // Handle device status handleDeviceStatus(topic, payload); } else { // Handle single sensor data SensorData sensorData = objectMapper.readValue(payload, SensorData.class); dataProcessor.processSensorData(sensorData); } } catch (Exception e) { System.err.println("Error processing MQTT message: " + e.getMessage()); // Could send to dead letter queue } } private void handleDeviceStatus(String topic, String payload) throws Exception { String deviceId = extractDeviceIdFromTopic(topic); DeviceStatus status = objectMapper.readValue(payload, DeviceStatus.class); dataProcessor.updateDeviceStatus(deviceId, status); } private String extractDeviceIdFromTopic(String topic) { String[] parts = topic.split("/"); return parts[1]; // Assuming format: devices/{deviceId}/status } @Override public void connectionLost(Throwable cause) { System.err.println("MQTT Connection lost: " + cause.getMessage()); // Implement reconnection logic } @Override public void deliveryComplete(IMqttDeliveryToken token) { // Not used for subscriptions } public void disconnect() throws MqttException { if (mqttClient != null && mqttClient.isConnected()) { mqttClient.disconnect(); } } }
2. Kafka Consumer for High-Volume Data
import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; @Component public class KafkaSensorConsumer { private final SensorDataProcessor dataProcessor; private final ObjectMapper objectMapper; public KafkaSensorConsumer(SensorDataProcessor dataProcessor, ObjectMapper objectMapper) { this.dataProcessor = dataProcessor; this.objectMapper = objectMapper; } @KafkaListener(topics = "${kafka.topics.sensor-data}", groupId = "${kafka.consumer.group}") public void consumeSensorData(@Payload String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, @Header(KafkaHeaders.RECEIVED_KEY) String key) { try { SensorData sensorData = objectMapper.readValue(message, SensorData.class); dataProcessor.processSensorData(sensorData); } catch (Exception e) { System.err.println("Error processing Kafka message: " + e.getMessage()); // Send to DLQ or error topic } } @KafkaListener(topics = "${kafka.topics.batch-sensor-data}", groupId = "${kafka.consumer.group}") public void consumeBatchSensorData(@Payload String message) { try { List<SensorData> batchData = objectMapper.readValue(message, objectMapper.getTypeFactory().constructCollectionType(List.class, SensorData.class)); dataProcessor.processBatchSensorData(batchData); } catch (Exception e) { System.err.println("Error processing batch sensor data: " + e.getMessage()); } } }

Data Processing Engine

1. Core Sensor Data Processor
import org.springframework.stereotype.Component; import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics; import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; @Component public class SensorDataProcessor { private final SensorDataRepository dataRepository; private final AlertService alertService; private final AnalyticsService analyticsService; private final Map<String, SensorStatistics> sensorStatistics; private final Map<String, Instant> deviceLastSeen; public SensorDataProcessor(SensorDataRepository dataRepository, AlertService alertService, AnalyticsService analyticsService) { this.dataRepository = dataRepository; this.alertService = alertService; this.analyticsService = analyticsService; this.sensorStatistics = new ConcurrentHashMap<>(); this.deviceLastSeen = new ConcurrentHashMap<>(); } public void processSensorData(SensorData sensorData) { // Validate data if (!validateSensorData(sensorData)) { System.err.println("Invalid sensor data: " + sensorData); return; } // Update device last seen deviceLastSeen.put(sensorData.getDeviceId(), Instant.now()); // Check for anomalies checkForAnomalies(sensorData); // Update statistics updateStatistics(sensorData); // Store data dataRepository.save(sensorData); // Perform real-time analytics analyticsService.processRealTimeData(sensorData); } public void processMultiSensorData(MultiSensorData multiData) { // Update device status and location updateDeviceStatus(multiData.getDeviceId(), multiData.getStatus()); // Process individual sensor readings multiData.getReadings().forEach((sensorType, reading) -> { SensorData sensorData = new SensorData(); sensorData.setDeviceId(multiData.getDeviceId()); sensorData.setSensorType(sensorType); sensorData.setValue(reading.getValue()); sensorData.setUnit(reading.getUnit()); sensorData.setTimestamp(multiData.getTimestamp()); sensorData.setLocation(multiData.getLocation()); processSensorData(sensorData); }); } public void processBatchSensorData(List<SensorData> batchData) { // Validate and filter data List<SensorData> validData = batchData.stream() .filter(this::validateSensorData) .toList(); // Batch insert for efficiency dataRepository.saveAll(validData); // Batch analytics analyticsService.processBatchData(validData); } private boolean validateSensorData(SensorData sensorData) { if (sensorData.getDeviceId() == null || sensorData.getDeviceId().trim().isEmpty()) { return false; } if (sensorData.getSensorType() == null || sensorData.getSensorType().trim().isEmpty()) { return false; } if (sensorData.getTimestamp() == null) { return false; } if (Double.isNaN(sensorData.getValue()) || Double.isInfinite(sensorData.getValue())) { return false; } return true; } private void checkForAnomalies(SensorData sensorData) { String sensorKey = sensorData.getDeviceId() + ":" + sensorData.getSensorType(); SensorStatistics stats = sensorStatistics.get(sensorKey); if (stats != null && stats.getDataCount() > 10) { double zScore = Math.abs((sensorData.getValue() - stats.getMean()) / stats.getStandardDeviation()); if (zScore > 3.0) { // 3 standard deviations Alert alert = new Alert(); alert.setDeviceId(sensorData.getDeviceId()); alert.setSensorType(sensorData.getSensorType()); alert.setValue(sensorData.getValue()); alert.setTimestamp(sensorData.getTimestamp()); alert.setType(AlertType.ANOMALY); alert.setMessage(String.format("Anomalous reading detected: %.2f (Z-score: %.2f)", sensorData.getValue(), zScore)); alertService.sendAlert(alert); } } // Check threshold violations checkThresholdViolations(sensorData); } private void checkThresholdViolations(SensorData sensorData) { // This would typically check against configured thresholds for the device/sensor // For example: if (sensorData.getSensorType().equals("temperature")) { if (sensorData.getValue() > 40.0) { // High temperature threshold Alert alert = new Alert(); alert.setDeviceId(sensorData.getDeviceId()); alert.setSensorType(sensorData.getSensorType()); alert.setValue(sensorData.getValue()); alert.setTimestamp(sensorData.getTimestamp()); alert.setType(AlertType.THRESHOLD_VIOLATION); alert.setMessage(String.format("High temperature alert: %.2f°C", sensorData.getValue())); alertService.sendAlert(alert); } } } private void updateStatistics(SensorData sensorData) { String sensorKey = sensorData.getDeviceId() + ":" + sensorData.getSensorType(); SensorStatistics stats = sensorStatistics.computeIfAbsent(sensorKey, k -> new SensorStatistics(1000)); // Keep last 1000 readings stats.addValue(sensorData.getValue()); } public void updateDeviceStatus(String deviceId, DeviceStatus status) { // Update device status in database // This could also trigger alerts for device offline/online status changes } public SensorStatistics getSensorStatistics(String deviceId, String sensorType) { String sensorKey = deviceId + ":" + sensorType; return sensorStatistics.get(sensorKey); } public Map<String, SensorStatistics> getAllStatistics() { return Collections.unmodifiableMap(sensorStatistics); } }
2. Sensor Statistics Calculator
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics; import java.util.concurrent.locks.ReentrantLock; public class SensorStatistics { private final DescriptiveStatistics stats; private final ReentrantLock lock; private final int windowSize; public SensorStatistics(int windowSize) { this.stats = new DescriptiveStatistics(windowSize); this.lock = new ReentrantLock(); this.windowSize = windowSize; } public void addValue(double value) { lock.lock(); try { stats.addValue(value); } finally { lock.unlock(); } } public double getMean() { lock.lock(); try { return stats.getMean(); } finally { lock.unlock(); } } public double getStandardDeviation() { lock.lock(); try { return stats.getStandardDeviation(); } finally { lock.unlock(); } } public double getMin() { lock.lock(); try { return stats.getMin(); } finally { lock.unlock(); } } public double getMax() { lock.lock(); try { return stats.getMax(); } finally { lock.unlock(); } } public double getPercentile(double percentile) { lock.lock(); try { return stats.getPercentile(percentile); } finally { lock.unlock(); } } public long getDataCount() { lock.lock(); try { return stats.getN(); } finally { lock.unlock(); } } public void clear() { lock.lock(); try { stats.clear(); } finally { lock.unlock(); } } }

Data Storage Layer

1. Time-Series Database (InfluxDB)
import com.influxdb.client.*; import com.influxdb.client.domain.WritePrecision; import com.influxdb.client.write.Point; import org.springframework.stereotype.Repository; import java.time.Instant; import java.util.List; @Repository public class InfluxDbSensorRepository { private final WriteApiBlocking writeApi; private final QueryApi queryApi; private final String bucket; private final String org; public InfluxDbSensorRepository(InfluxDBClient influxDBClient, @Value("${influxdb.bucket}") String bucket, @Value("${influxdb.org}") String org) { this.writeApi = influxDBClient.getWriteApiBlocking(); this.queryApi = influxDBClient.getQueryApi(); this.bucket = bucket; this.org = org; } public void save(SensorData sensorData) { Point point = Point.measurement("sensor_data") .addTag("device_id", sensorData.getDeviceId()) .addTag("sensor_type", sensorData.getSensorType()) .addField("value", sensorData.getValue()) .addField("unit", sensorData.getUnit()) .time(sensorData.getTimestamp().toEpochMilli(), WritePrecision.MS); if (sensorData.getLocation() != null) { point.addTag("latitude", String.valueOf(sensorData.getLocation().getLatitude())) .addTag("longitude", String.valueOf(sensorData.getLocation().getLongitude())); } writeApi.writePoint(bucket, org, point); } public void saveAll(List<SensorData> sensorDataList) { List<Point> points = sensorDataList.stream() .map(this::convertToPoint) .toList(); writeApi.writePoints(bucket, org, points); } private Point convertToPoint(SensorData sensorData) { Point point = Point.measurement("sensor_data") .addTag("device_id", sensorData.getDeviceId()) .addTag("sensor_type", sensorData.getSensorType()) .addField("value", sensorData.getValue()) .addField("unit", sensorData.getUnit()) .time(sensorData.getTimestamp().toEpochMilli(), WritePrecision.MS); if (sensorData.getLocation() != null) { point.addTag("latitude", String.valueOf(sensorData.getLocation().getLatitude())) .addTag("longitude", String.valueOf(sensorData.getLocation().getLongitude())); } return point; } public List<SensorData> queryData(String deviceId, String sensorType, Instant startTime, Instant endTime) { String fluxQuery = String.format( "from(bucket: \"%s\") " + "|> range(start: %s, stop: %s) " + "|> filter(fn: (r) => r._measurement == \"sensor_data\") " + "|> filter(fn: (r) => r.device_id == \"%s\") " + "|> filter(fn: (r) => r.sensor_type == \"%s\") " + "|> filter(fn: (r) => r._field == \"value\")", bucket, startTime, endTime, deviceId, sensorType); List<SensorData> results = queryApi.query(fluxQuery, org, SensorData.class); return results; } }
2. MongoDB Repository for Device Management
import org.springframework.data.mongodb.repository.MongoRepository; import org.springframework.data.mongodb.repository.Query; import org.springframework.stereotype.Repository; import java.time.Instant; import java.util.List; import java.util.Optional; @Repository public interface DeviceRepository extends MongoRepository<IoTDevice, String> { Optional<IoTDevice> findByDeviceId(String deviceId); List<IoTDevice> findByDeviceType(String deviceType); List<IoTDevice> findByStatus(DeviceStatus status); List<IoTDevice> findByInstallationLocationLatitudeBetweenAndInstallationLocationLongitudeBetween( double minLat, double maxLat, double minLon, double maxLon); @Query("{ 'lastSeen': { $lt: ?0 } }") List<IoTDevice> findDevicesOfflineSince(Instant timestamp); @Query("{ 'sensorConfigs.?0.enabled': true }") List<IoTDevice> findDevicesWithSensorType(String sensorType); }

Analytics and Monitoring

1. Real-time Analytics Service
import org.springframework.stereotype.Service; import java.time.Duration; import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; @Service public class AnalyticsService { private final Map<String, RollingWindow> rollingWindows; private final Map<String, TrendAnalysis> trendAnalyses; public AnalyticsService() { this.rollingWindows = new ConcurrentHashMap<>(); this.trendAnalyses = new ConcurrentHashMap<>(); } public void processRealTimeData(SensorData sensorData) { String key = sensorData.getDeviceId() + ":" + sensorData.getSensorType(); // Update rolling window RollingWindow window = rollingWindows.computeIfAbsent(key, k -> new RollingWindow(Duration.ofMinutes(10))); // 10-minute window window.addDataPoint(sensorData); // Perform trend analysis performTrendAnalysis(sensorData, key); // Detect patterns detectPatterns(sensorData, key); } public void processBatchData(List<SensorData> batchData) { // Aggregate analysis for batch data Map<String, List<SensorData>> groupedData = new HashMap<>(); for (SensorData data : batchData) { String key = data.getDeviceId() + ":" + data.getSensorType(); groupedData.computeIfAbsent(key, k -> new ArrayList<>()).add(data); } // Perform batch analytics for each group groupedData.forEach(this::performBatchAnalysis); } private void performTrendAnalysis(SensorData sensorData, String key) { TrendAnalysis trend = trendAnalyses.computeIfAbsent(key, k -> new TrendAnalysis(100)); // Last 100 points trend.addDataPoint(sensorData); // Check for significant trends if (trend.hasEnoughData()) { double slope = trend.calculateSlope(); if (Math.abs(slope) > 0.1) { // Significant trend threshold System.out.printf("Significant trend detected for %s: slope=%.3f%n", key, slope); // Could trigger alerts or adaptive sampling if (slope > 0) { System.out.println("Increasing trend detected"); } else { System.out.println("Decreasing trend detected"); } } } } private void detectPatterns(SensorData sensorData, String key) { // Implement pattern detection logic // This could include: // - Seasonal patterns // - Spike detection // - Correlation between different sensors } private void performBatchAnalysis(String key, List<SensorData> data) { // Perform comprehensive analysis on batch data DescriptiveStatistics stats = new DescriptiveStatistics(); data.forEach(d -> stats.addValue(d.getValue())); System.out.printf("Batch Analysis for %s: count=%d, mean=%.2f, std=%.2f%n", key, stats.getN(), stats.getMean(), stats.getStandardDeviation()); } public RollingWindow getRollingWindow(String deviceId, String sensorType) { String key = deviceId + ":" + sensorType; return rollingWindows.get(key); } } public class RollingWindow { private final LinkedList<SensorData> dataPoints; private final Duration windowDuration; private final int maxPoints; public RollingWindow(Duration windowDuration) { this.dataPoints = new LinkedList<>(); this.windowDuration = windowDuration; this.maxPoints = 1000; // Prevent memory issues } public void addDataPoint(SensorData dataPoint) { dataPoints.addLast(dataPoint); // Remove old data points Instant cutoff = Instant.now().minus(windowDuration); while (!dataPoints.isEmpty() && dataPoints.getFirst().getTimestamp().isBefore(cutoff)) { dataPoints.removeFirst(); } // Enforce maximum size while (dataPoints.size() > maxPoints) { dataPoints.removeFirst(); } } public List<SensorData> getDataPoints() { return new ArrayList<>(dataPoints); } public int getSize() { return dataPoints.size(); } public Optional<Double> getAverage() { if (dataPoints.isEmpty()) return Optional.empty(); double sum = dataPoints.stream() .mapToDouble(SensorData::getValue) .sum(); return Optional.of(sum / dataPoints.size()); } } public class TrendAnalysis { private final LinkedList<SensorData> dataPoints; private final int maxPoints; public TrendAnalysis(int maxPoints) { this.dataPoints = new LinkedList<>(); this.maxPoints = maxPoints; } public void addDataPoint(SensorData dataPoint) { dataPoints.addLast(dataPoint); if (dataPoints.size() > maxPoints) { dataPoints.removeFirst(); } } public boolean hasEnoughData() { return dataPoints.size() >= 10; // Minimum points for trend analysis } public double calculateSlope() { if (dataPoints.size() < 2) return 0.0; // Simple linear regression int n = dataPoints.size(); double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0; for (int i = 0; i < n; i++) { double x = i; // Time index double y = dataPoints.get(i).getValue(); sumX += x; sumY += y; sumXY += x * y; sumX2 += x * x; } double slope = (n * sumXY - sumX * sumY) / (n * sumX2 - sumX * sumX); return slope; } }
2. Alert Management System
import org.springframework.stereotype.Service; import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; @Service public class AlertService { private final Map<String, List<Alert>> deviceAlerts; private final List<AlertListener> listeners; public AlertService() { this.deviceAlerts = new ConcurrentHashMap<>(); this.listeners = new ArrayList<>(); } public void sendAlert(Alert alert) { // Store alert deviceAlerts.computeIfAbsent(alert.getDeviceId(), k -> new ArrayList<>()) .add(alert); // Notify listeners notifyListeners(alert); // Log alert System.out.println("ALERT: " + alert.getMessage()); } public void addListener(AlertListener listener) { listeners.add(listener); } private void notifyListeners(Alert alert) { for (AlertListener listener : listeners) { try { listener.onAlert(alert); } catch (Exception e) { System.err.println("Error notifying alert listener: " + e.getMessage()); } } } public List<Alert> getAlertsForDevice(String deviceId, Instant since) { List<Alert> deviceAlertList = deviceAlerts.getOrDefault(deviceId, new ArrayList<>()); return deviceAlertList.stream() .filter(alert -> alert.getTimestamp().isAfter(since)) .toList(); } public void clearOldAlerts(Instant cutoff) { for (List<Alert> alerts : deviceAlerts.values()) { alerts.removeIf(alert -> alert.getTimestamp().isBefore(cutoff)); } } } public class Alert { private String deviceId; private String sensorType; private double value; private Instant timestamp; private AlertType type; private String message; private AlertSeverity severity; // Getters and setters public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getSensorType() { return sensorType; } public void setSensorType(String sensorType) { this.sensorType = sensorType; } public double getValue() { return value; } public void setValue(double value) { this.value = value; } public Instant getTimestamp() { return timestamp; } public void setTimestamp(Instant timestamp) { this.timestamp = timestamp; } public AlertType getType() { return type; } public void setType(AlertType type) { this.type = type; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public AlertSeverity getSeverity() { return severity; } public void setSeverity(AlertSeverity severity) { this.severity = severity; } } public enum AlertType { THRESHOLD_VIOLATION, ANOMALY, DEVICE_OFFLINE, DEVICE_ERROR, MAINTENANCE_REQUIRED } public enum AlertSeverity { LOW, MEDIUM, HIGH, CRITICAL } public interface AlertListener { void onAlert(Alert alert); }

REST API Controllers

1. Sensor Data API
import org.springframework.web.bind.annotation.*; import org.springframework.http.ResponseEntity; import java.time.Instant; import java.util.List; @RestController @RequestMapping("/api/sensors") public class SensorDataController { private final SensorDataProcessor dataProcessor; private final SensorDataRepository dataRepository; private final AnalyticsService analyticsService; public SensorDataController(SensorDataProcessor dataProcessor, SensorDataRepository dataRepository, AnalyticsService analyticsService) { this.dataProcessor = dataProcessor; this.dataRepository = dataRepository; this.analyticsService = analyticsService; } @PostMapping("/data") public ResponseEntity<SensorData> receiveSensorData(@RequestBody SensorData sensorData) { sensorData.setTimestamp(Instant.now()); // Ensure timestamp is set dataProcessor.processSensorData(sensorData); return ResponseEntity.ok(sensorData); } @PostMapping("/data/batch") public ResponseEntity<String> receiveBatchSensorData(@RequestBody List<SensorData> sensorDataList) { dataProcessor.processBatchSensorData(sensorDataList); return ResponseEntity.ok("Processed " + sensorDataList.size() + " data points"); } @GetMapping("/data/{deviceId}/{sensorType}") public ResponseEntity<List<SensorData>> getSensorData( @PathVariable String deviceId, @PathVariable String sensorType, @RequestParam Instant startTime, @RequestParam Instant endTime) { List<SensorData> data = dataRepository.queryData(deviceId, sensorType, startTime, endTime); return ResponseEntity.ok(data); } @GetMapping("/statistics/{deviceId}/{sensorType}") public ResponseEntity<SensorStatistics> getSensorStatistics( @PathVariable String deviceId, @PathVariable String sensorType) { SensorStatistics stats = dataProcessor.getSensorStatistics(deviceId, sensorType); return ResponseEntity.ok(stats); } @GetMapping("/analytics/{deviceId}/{sensorType}") public ResponseEntity<RollingWindow> getAnalyticsData( @PathVariable String deviceId, @PathVariable String sensorType) { RollingWindow window = analyticsService.getRollingWindow(deviceId, sensorType); return ResponseEntity.ok(window); } }
2. Device Management API
@RestController @RequestMapping("/api/devices") public class DeviceController { private final DeviceRepository deviceRepository; private final AlertService alertService; public DeviceController(DeviceRepository deviceRepository, AlertService alertService) { this.deviceRepository = deviceRepository; this.alertService = alertService; } @PostMapping public ResponseEntity<IoTDevice> registerDevice(@RequestBody IoTDevice device) { IoTDevice savedDevice = deviceRepository.save(device); return ResponseEntity.ok(savedDevice); } @GetMapping("/{deviceId}") public ResponseEntity<IoTDevice> getDevice(@PathVariable String deviceId) { Optional<IoTDevice> device = deviceRepository.findByDeviceId(deviceId); return device.map(ResponseEntity::ok) .orElse(ResponseEntity.notFound().build()); } @GetMapping("/{deviceId}/alerts") public ResponseEntity<List<Alert>> getDeviceAlerts( @PathVariable String deviceId, @RequestParam(defaultValue = "1") int hours) { Instant since = Instant.now().minusSeconds(hours * 3600); List<Alert> alerts = alertService.getAlertsForDevice(deviceId, since); return ResponseEntity.ok(alerts); } @GetMapping("/status/{status}") public ResponseEntity<List<IoTDevice>> getDevicesByStatus(@PathVariable DeviceStatus status) { List<IoTDevice> devices = deviceRepository.findByStatus(status); return ResponseEntity.ok(devices); } @PutMapping("/{deviceId}/config") public ResponseEntity<IoTDevice> updateDeviceConfig( @PathVariable String deviceId, @RequestBody Map<String, SensorConfig> sensorConfigs) { Optional<IoTDevice> deviceOpt = deviceRepository.findByDeviceId(deviceId); if (deviceOpt.isEmpty()) { return ResponseEntity.notFound().build(); } IoTDevice device = deviceOpt.get(); device.setSensorConfigs(sensorConfigs); IoTDevice updatedDevice = deviceRepository.save(device); return ResponseEntity.ok(updatedDevice); } }

WebSocket for Real-time Updates

import org.springframework.web.socket.*; import org.springframework.web.socket.handler.TextWebSocketHandler; import org.springframework.stereotype.Component; import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; @Component public class SensorWebSocketHandler extends TextWebSocketHandler implements AlertListener { private final List<WebSocketSession> sessions = new CopyOnWriteArrayList<>(); private final ObjectMapper objectMapper; public SensorWebSocketHandler(ObjectMapper objectMapper, AlertService alertService) { this.objectMapper = objectMapper; alertService.addListener(this); } @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { sessions.add(session); System.out.println("WebSocket connection established: " + session.getId()); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { sessions.remove(session); System.out.println("WebSocket connection closed: " + session.getId()); } @Override public void onAlert(Alert alert) { // Broadcast alert to all connected clients broadcastMessage("ALERT", alert); } public void broadcastSensorData(SensorData sensorData) { broadcastMessage("SENSOR_DATA", sensorData); } private void broadcastMessage(String type, Object data) { Map<String, Object> message = new HashMap<>(); message.put("type", type); message.put("data", data); message.put("timestamp", Instant.now()); String jsonMessage; try { jsonMessage = objectMapper.writeValueAsString(message); } catch (IOException e) { System.err.println("Error serializing WebSocket message: " + e.getMessage()); return; } for (WebSocketSession session : sessions) { if (session.isOpen()) { try { session.sendMessage(new TextMessage(jsonMessage)); } catch (IOException e) { System.err.println("Error sending WebSocket message: " + e.getMessage()); } } } } }

Configuration Classes

1. Application Configuration
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; @Configuration public class AppConfig { @Bean public ObjectMapper objectMapper() { ObjectMapper mapper = new ObjectMapper(); mapper.registerModule(new JavaTimeModule()); return mapper; } @Bean public MqttSensorClient mqttSensorClient(ObjectMapper objectMapper, SensorDataProcessor dataProcessor, @Value("${mqtt.broker.url}") String brokerUrl) { return new MqttSensorClient(objectMapper, dataProcessor, brokerUrl); } }
2. Spring Boot Application
import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableScheduling public class IoTApplication { public static void main(String[] args) { SpringApplication.run(IoTApplication.class, args); } }

Usage Examples

1. Simulating Sensor Data
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.time.Instant; import java.util.Random; @Component public class SensorDataSimulator { private final SensorDataProcessor dataProcessor; private final Random random = new Random(); public SensorDataSimulator(SensorDataProcessor dataProcessor) { this.dataProcessor = dataProcessor; } @Scheduled(fixedRate = 5000) // Every 5 seconds public void simulateTemperatureSensor() { SensorData data = new SensorData(); data.setDeviceId("temp-sensor-001"); data.setSensorType("temperature"); data.setValue(20 + random.nextGaussian() * 5); // 20°C ± 5°C data.setUnit("°C"); data.setTimestamp(Instant.now()); dataProcessor.processSensorData(data); } @Scheduled(fixedRate = 10000) // Every 10 seconds public void simulateMultiSensorDevice() { MultiSensorData multiData = new MultiSensorData(); multiData.setDeviceId("multi-sensor-001"); multiData.setTimestamp(Instant.now()); multiData.setStatus(DeviceStatus.ONLINE); Map<String, SensorReading> readings = new HashMap<>(); // Temperature reading SensorReading tempReading = new SensorReading(); tempReading.setValue(22.5 + random.nextGaussian() * 2); tempReading.setUnit("°C"); readings.put("temperature", tempReading); // Humidity reading SensorReading humidityReading = new SensorReading(); humidityReading.setValue(45 + random.nextGaussian() * 10); humidityReading.setUnit("%"); readings.put("humidity", humidityReading); // Pressure reading SensorReading pressureReading = new SensorReading(); pressureReading.setValue(1013 + random.nextGaussian() * 5); pressureReading.setUnit("hPa"); readings.put("pressure", pressureReading); multiData.setReadings(readings); dataProcessor.processMultiSensorData(multiData); } }
2. Monitoring Dashboard
@RestController @RequestMapping("/api/dashboard") public class DashboardController { private final SensorDataProcessor dataProcessor; private final DeviceRepository deviceRepository; private final AnalyticsService analyticsService; public DashboardController(SensorDataProcessor dataProcessor, DeviceRepository deviceRepository, AnalyticsService analyticsService) { this.dataProcessor = dataProcessor; this.deviceRepository = deviceRepository; this.analyticsService = analyticsService; } @GetMapping("/summary") public ResponseEntity<DashboardSummary> getDashboardSummary() { DashboardSummary summary = new DashboardSummary(); // Get device counts long totalDevices = deviceRepository.count(); long onlineDevices = deviceRepository.findByStatus(DeviceStatus.ONLINE).size(); summary.setTotalDevices(totalDevices); summary.setOnlineDevices(onlineDevices); summary.setOfflineDevices(totalDevices - onlineDevices); // Get recent alerts count // This would typically query the alert service // Get system status summary.setSystemStatus("OPERATIONAL"); summary.setLastUpdated(Instant.now()); return ResponseEntity.ok(summary); } } public class DashboardSummary { private long totalDevices; private long onlineDevices; private long offlineDevices; private int activeAlerts; private String systemStatus; private Instant lastUpdated; // Getters and setters public long getTotalDevices() { return totalDevices; } public void setTotalDevices(long totalDevices) { this.totalDevices = totalDevices; } public long getOnlineDevices() { return onlineDevices; } public void setOnlineDevices(long onlineDevices) { this.onlineDevices = onlineDevices; } public long getOfflineDevices() { return offlineDevices; } public void setOfflineDevices(long offlineDevices) { this.offlineDevices = offlineDevices; } public int getActiveAlerts() { return activeAlerts; } public void setActiveAlerts(int activeAlerts) { this.activeAlerts = activeAlerts; } public String getSystemStatus() { return systemStatus; } public void setSystemStatus(String systemStatus) { this.systemStatus = systemStatus; } public Instant getLastUpdated() { return lastUpdated; } public void setLastUpdated(Instant lastUpdated) { this.lastUpdated = lastUpdated; } }

Best Practices

  1. Scalability: Use message brokers for high-volume data
  2. Fault Tolerance: Implement retry mechanisms and circuit breakers
  3. Security: Encrypt sensitive data and use secure protocols
  4. Monitoring: Implement comprehensive logging and metrics
  5. Data Quality: Validate and clean sensor data
  6. Performance: Use efficient data structures and algorithms
// Example of circuit breaker pattern for external services @Component public class CircuitBreaker { private final int failureThreshold; private final long timeout; private int failureCount = 0; private long lastFailureTime = 0; private State state = State.CLOSED; enum State { CLOSED, OPEN, HALF_OPEN } public CircuitBreaker(@Value("${circuitbreaker.failure-threshold}") int failureThreshold, @Value("${circuitbreaker.timeout}") long timeout) { this.failureThreshold = failureThreshold; this.timeout = timeout; } public boolean allowRequest() { if (state == State.OPEN) { if (System.currentTimeMillis() - lastFailureTime > timeout) { state = State.HALF_OPEN; return true; } return false; } return true; } public void recordSuccess() { failureCount = 0; state = State.CLOSED; } public void recordFailure() { failureCount++; lastFailureTime = System.currentTimeMillis(); if (failureCount >= failureThreshold) { state = State.OPEN; } } }

Conclusion

This comprehensive IoT sensor data processing system provides:

  • Real-time data ingestion from multiple protocols (MQTT, Kafka, HTTP)
  • Advanced data processing with anomaly detection and trend analysis
  • Scalable storage using time-series and document databases
  • Real-time monitoring with WebSocket updates
  • Alert management for proactive issue detection
  • RESTful APIs for integration and management

The modular architecture allows for easy extension and adaptation to specific IoT use cases, from smart home applications to industrial monitoring systems.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper