☕ 스프링 부트를 활용한 실시간 데이터 스트리밍 시스템

// Maven 의존성 추가
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.0</version>
</dependency>

// Kafka 데이터 수신 및 처리 예제
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class DataStreamConsumer {
    private final Map<String, Long> statsMap = new ConcurrentHashMap<>();

    @KafkaListener(topics = "sensor-topic", groupId = "realtime-group")
    public void consume(String message) {
        try {
            // 메시지 파싱 (예: JSON)
            Map<String, Object> data = parseJson(message);
            String sensorId = (String) data.get("id");
            long value = ((Number) data.get("value")).longValue();

            // 실시간 통계 계산
            statsMap.put(sensorId, statsMap.getOrDefault(sensorId, 0L) + value);
        } catch (Exception e) {
            System.err.println("Error processing message: " + e.getMessage());
        }
    }

    private Map<String, Object> parseJson(String json) {
        // 실제 로직은 JSON 파서(예: Jackson) 사용
        return new ConcurrentHashMap<>(); // 예시로 빈 맵 반환
    }
}

// REST API를 통한 결과 전송
import org.springframework.web.bind.annotation.*;
import java.util.Map;

@RestController
@RequestMapping("/api/v1/stats")
public class StatsController {
    private final Map<String, Long> statsMap = new ConcurrentHashMap<>();

    @GetMapping("/{sensorId}")
    public long getSensorStats(@PathVariable String sensorId) {
        return statsMap.getOrDefault(sensorId, 0L);
    }

    @PostMapping("/update")
    public void updateSensorData(@RequestBody Map<String, Object> data) {
        String sensorId = (String) data.get("id");
        long value = ((Number) data.get("value")).longValue();
        statsMap.put(sensor, statsMap.getOrDefault(sensorId, 0L) + value);
    }
}