Spring Integration이란? 개념부터 데이터 수집 시스템 적용까지
Spring 기반 애플리케이션을 개발하다 보면 단순한 CRUD를 넘어 여러 시스템과 데이터를 주고받아야 하는 경우가 있습니다.
예를 들어 데이터 수집 시스템에서는 다음과 같은 처리가 필요할 수 있습니다.
센서 / 장비 / 외부 API
↓
데이터 수신
↓
JSON 파싱
↓
데이터 검증
↓
DB 저장
↓
데이터 변환/매핑
↓
데이터허브 전송
단순한 시스템이라면 각각의 기능을 Service에서 순차적으로 호출해도 충분합니다.
하지만 수집 대상과 프로토콜이 많아지고 오류 처리, 분기, 재처리 등의 요구사항이 추가되면 코드가 빠르게 복잡해집니다.
이러한 **시스템 간 데이터 흐름과 메시지 처리를 체계적으로 구현하기 위한 Spring 프레임워크가 Spring Integration**입니다.
1. Spring Integration이란?
Spring Integration은 Spring Framework 기반의 EAI(Enterprise Application Integration) 프레임워크입니다.
쉽게 표현하면,
여러 시스템에서 들어오는 데이터를 메시지 형태로 받아 변환, 검증, 분기, 저장, 전송하는 흐름을 구성하는 프레임워크
라고 할 수 있습니다.
Spring Integration은 메시징 기반으로 동작하며 다음과 같은 다양한 시스템과 연동할 수 있습니다.
- HTTP
- TCP
- UDP
- MQTT
- Kafka
- JMS
- FTP
- SFTP
- File
- Database
예를 들어 IoT 장비에서 TCP로 데이터를 받고 이를 JSON으로 변환하여 Oracle DB에 저장한 뒤 데이터허브 API로 전달하는 구조를 만들 수 있습니다.
IoT 장비
↓
TCP
↓
Spring Integration
↓
JSON Parsing
↓
Validation
↓
Oracle
↓
HTTP
↓
DataHub
2. Spring Integration의 핵심 개념
Spring Integration을 이해하려면 다음 개념을 먼저 알아두는 것이 좋습니다.
Message
Channel
Endpoint
Adapter
Gateway
Transformer
Filter
Router
Service Activator
Integration Flow
Error Channel
하나씩 살펴보겠습니다.
3. Message
Spring Integration에서 데이터 처리의 기본 단위는 Message입니다.
Message는 크게 두 부분으로 구성됩니다.
Message
├── Payload
└── Headers
Payload
실제로 처리할 데이터입니다.
예를 들어 장비에서 다음 JSON이 들어왔다고 가정하겠습니다.
{
"deviceId": "DEV001",
"temperature": 25.4,
"humidity": 62
}
이 JSON 문자열 또는 이를 변환한 객체가 Payload가 됩니다.
Headers
데이터 처리에 필요한 부가정보입니다.
예를 들면 다음과 같습니다.
sourceIp
protocol
deviceId
receivedAt
serviceType
개념적으로 다음과 같은 Message가 만들어질 수 있습니다.
Message
├── Payload
│ └── {
│ "deviceId": "DEV001",
│ "temperature": 25.4
│ }
│
└── Headers
├── sourceIp = 192.168.0.10
├── protocol = TCP
└── receivedAt = 2026-08-12 13:30:00
4. Message Channel
Message Channel은 Message가 이동하는 통로입니다.
Producer
↓
Channel
↓
Consumer
Spring Integration에서는 처리 단계 사이를 Channel로 연결합니다.
데이터 수신
↓
receiveChannel
↓
JSON Parsing
↓
parseChannel
↓
DB 저장
대표적인 Channel 종류는 다음과 같습니다.
DirectChannel
가장 기본적인 Channel입니다.
메시지를 보내면 동일한 처리 흐름에서 다음 Handler가 즉시 실행됩니다.
@Bean
public MessageChannel receiveChannel() {
return new DirectChannel();
}
구조는 단순합니다.
A → B
대부분의 기본 Integration Flow에서는 DirectChannel부터 사용하면 됩니다.
QueueChannel
메시지를 Queue에 저장한 후 Consumer가 가져가는 방식입니다.
Producer
↓
Queue
↓
Consumer
수집 속도와 처리 속도가 다르거나 일시적으로 메시지를 보관해야 할 때 사용할 수 있습니다.
PublishSubscribeChannel
하나의 메시지를 여러 Consumer에게 전달할 때 사용합니다.
→ DB 저장
Message ────→ 로그 저장
→ 모니터링
하나의 수신 데이터를 여러 처리기에 동시에 전달해야 할 때 유용합니다.
5. Integration Flow
IntegrationFlow는 Spring Integration의 핵심입니다.
데이터가 어떤 순서로 처리될지를 정의합니다.
예를 들어 다음과 같은 수집 프로세스가 있다고 가정합니다.
수신
↓
JSON Parsing
↓
Validation
↓
DB 저장
↓
DataHub 전송
Java DSL을 사용하면 다음처럼 표현할 수 있습니다.
@Bean
public IntegrationFlow collectFlow() {
return IntegrationFlow
.from("receiveChannel")
.handle(jsonParseService, "parse")
.handle(validationService, "validate")
.handle(collectService, "save")
.handle(dataHubService, "send")
.get();
}
일반적인 Spring 코드라면 다음과 비슷합니다.
public void collect(String data) {
Object parsedData = jsonParseService.parse(data);
validationService.validate(parsedData);
collectService.save(parsedData);
dataHubService.send(parsedData);
}
두 방식의 차이는 처리 흐름 자체를 코드에서 직접 관리하느냐, Integration Flow가 관리하느냐에 있습니다.
6. Inbound Adapter와 Outbound Adapter
Spring Integration에서 외부 시스템과 데이터를 주고받는 역할을 Adapter가 담당합니다.
Inbound Adapter
외부 시스템에서 Spring 애플리케이션으로 데이터가 들어오는 부분입니다.
외부 시스템
↓
Inbound Adapter
↓
Spring Integration
예를 들면 다음과 같습니다.
TCP 장비 → Spring
HTTP API → Spring
MQTT → Spring
File → Spring
Outbound Adapter
Spring에서 외부 시스템으로 데이터를 보내는 역할입니다.
Spring Integration
↓
Outbound Adapter
↓
외부 시스템
예:
Spring → HTTP API
Spring → Kafka
Spring → MQTT
Spring → TCP Server
따라서 데이터 수집 시스템에서는 다음과 같은 구성이 가능합니다.
센서
↓
TCP Inbound Adapter
↓
Spring Integration
↓
Oracle
↓
HTTP Outbound Adapter
↓
데이터허브
7. Transformer
Transformer는 데이터를 다른 형태로 변환할 때 사용합니다.
장비마다 보내는 데이터 규격이 다를 경우 특히 유용합니다.
예를 들어 장비가 다음과 같이 데이터를 전송한다고 가정합니다.
{
"tm": "25.3",
"hm": "62"
}
하지만 내부 시스템에서는 다음 형식을 사용한다고 하겠습니다.
{
"temperature": 25.3,
"humidity": 62
}
이때 Transformer를 이용하여 다음과 같이 변환합니다.
장비 원본 데이터
↓
Transformer
↓
내부 표준 데이터
Java DSL에서는 다음과 같이 사용할 수 있습니다.
.transform(message -> {
return convertData(message);
})
또는 별도의 Service로 구현할 수도 있습니다.
.handle(dataTransformService, "transform")
실제 프로젝트에서는 변환 로직이 복잡해질 가능성이 높기 때문에 별도의 Service 클래스로 분리하는 방식이 관리하기 편합니다.
8. Filter
Filter는 조건에 맞지 않는 메시지를 걸러냅니다.
예를 들어 온도 데이터가 허용 가능한 범위인지 확인할 수 있습니다.
.filter(data ->
data.getTemperature() >= -50
&& data.getTemperature() <= 100
)
처리 흐름은 다음과 같습니다.
수신 데이터
↓
Filter
↙ ↘
정상 비정상
↓ ↓
처리 오류 처리
다음과 같은 검증에 사용할 수 있습니다.
- 필수 데이터 존재 여부
- 값 범위 검증
- 장비 등록 여부
- 서비스 활성화 여부
- 데이터 중복 여부
9. Router
Router는 메시지의 조건에 따라 처리 흐름을 나누는 역할을 합니다.
예를 들어 하나의 수집 서버에서 여러 종류의 서비스를 처리한다고 가정해보겠습니다.
환경센서
CCTV
교통센서
스마트부이
수신된 서비스 종류에 따라 다음과 같이 분기할 수 있습니다.
Message
↓
Router
┌─────────┼─────────┐
↓ ↓ ↓
환경센서 CCTV 교통
↓ ↓ ↓
Env Flow CCTV Flow Traffic Flow
예를 들면 다음과 같이 구성할 수 있습니다.
.route(
CollectData::getServiceType,
mapping -> mapping
.subFlowMapping("ENV", environmentFlow())
.subFlowMapping("CCTV", cctvFlow())
.subFlowMapping("TRAFFIC", trafficFlow())
)
여러 종류의 서비스를 하나의 수집 서버에서 처리할 때 매우 유용한 기능입니다.
10. Service Activator
Service Activator는 메시지를 실제 Spring Service에 전달하여 업무 로직을 실행합니다.
예를 들어 수집 데이터를 DB에 저장하는 Service가 있다고 가정합니다.
@Service
public class CollectService {
public CollectData save(CollectData data) {
// DB 저장
return data;
}
}
Integration Flow에서는 다음과 같이 호출할 수 있습니다.
.handle(collectService, "save")
즉 Spring Integration을 사용한다고 해서 기존 Spring의 Service나 MyBatis 구조를 버리는 것은 아닙니다.
오히려 다음과 같이 역할을 분리하는 것이 좋습니다.
Spring Integration
↓
데이터 흐름 제어
Service
↓
업무 로직
Mapper / MyBatis
↓
Database
11. Error Channel
데이터 수집 시스템에서는 오류 처리가 매우 중요합니다.
Spring Integration에서는 처리 중 발생한 오류를 errorChannel로 전달할 수 있습니다.
데이터 수신
↓
JSON Parsing
↓
ERROR
↓
errorChannel
↓
오류 이력 저장
다음과 같이 별도의 Error Flow를 만들 수 있습니다.
@Bean
public IntegrationFlow errorFlow() {
return IntegrationFlow
.from("errorChannel")
.handle(errorService, "handle")
.get();
}
예를 들어 다음과 같은 오류코드를 정의할 수 있습니다.
COL-1001 CONNECTION_FAILED
COL-1002 CONNECTION_TIMEOUT
COL-2001 EMPTY_DATA
COL-2003 JSON_PARSE_ERROR
COL-2005 REQUIRED_FIELD_MISSING
COL-2006 INVALID_DATA_TYPE
COL-2101 SCHEMA_NOT_FOUND
COL-2102 SCHEMA_MISMATCH
COL-3002 DB_INSERT_FAILED
COL-4002 HUB_SEND_FAILED
COL-5001 INTERNAL_ERROR
오류 발생 시 Error Channel에서 에러코드를 판별하여 DB에 저장할 수 있습니다.
public void handle(ErrorMessage message) {
Throwable error = message.getPayload();
// 오류 분석
// 오류 코드 생성
// DB 오류 이력 저장
}
12. 데이터 수집 시스템 적용 예시
Spring Integration은 특히 IoT 또는 도시데이터 수집 시스템에서 활용하기 좋습니다.
전체 흐름을 다음과 같이 구성할 수 있습니다.
장비 / 센서 / API
↓
Inbound Adapter
↓
receiveChannel
↓
원본 데이터 저장
↓
JSON Parser
↓
Schema Validator
↓
Transformer
↓
Router
┌────┼────┐
↓ ↓ ↓
환경 CCTV 교통
└────┼────┘
↓
Oracle 저장
↓
DataHub 변환
↓
Outbound Adapter
↓
데이터허브
오류가 발생하면 별도의 흐름으로 전달합니다.
각 처리 단계
↓
ERROR
↓
errorChannel
↓
오류 코드 판별
↓
오류 이력 저장
13. JSON 원본 데이터를 CLOB으로 저장하는 구조
수집 시스템에서는 수신한 원본 데이터를 그대로 보존하는 것이 중요합니다.
예를 들어 다음 JSON을 수신했다고 가정합니다.
{
"deviceId": "DEV001",
"temperature": 25.3,
"humidity": 60
}
Oracle에서는 원본 데이터를 CLOB 컬럼에 저장할 수 있습니다.
CREATE TABLE TBL_COLLECT_DATA (
COLLECT_NO NUMBER
, DEVICE_ID VARCHAR2(50)
, COLLECT_STATUS VARCHAR2(20)
, ERROR_CODE VARCHAR2(20)
, ERROR_MSG VARCHAR2(1000)
, RAW_DATA CLOB
, CREATE_AT DATE
);
Java에서는 CLOB을 직접 다루기보다 일반적으로 String으로 처리합니다.
public class CollectData {
private Long collectNo;
private String deviceId;
private String collectStatus;
private String errorCode;
private String errorMsg;
private String rawData;
}
수신 데이터 처리 순서는 다음과 같이 구성할 수 있습니다.
JSON 수신
↓
RAW_DATA CLOB 저장
↓
JSON Parsing
↓
Schema 검증
↓
데이터 변환
↓
처리
이 구조의 장점은 JSON Parsing에 실패해도 수신 원본 데이터가 남아 있다는 것입니다.
예를 들어 잘못된 JSON이 들어왔다면 다음과 같은 상태를 남길 수 있습니다.
COLLECT_STATUS = FAIL
ERROR_CODE = COL-2003
ERROR_MSG = JSON_PARSE_ERROR
RAW_DATA = 실제 수신한 원본 데이터
이를 통해 운영자가 장애 원인을 확인하거나 나중에 데이터를 재처리할 수 있습니다.
14. 일반 Spring Service 방식과의 차이
Spring Integration을 반드시 사용해야 하는 것은 아닙니다.
단순한 시스템이라면 기존 Spring Service 방식이 오히려 더 단순합니다.
일반 Spring
public void collect(String data) {
parse(data);
validate(data);
save(data);
send(data);
}
하지만 처리 종류가 많아지면 다음과 같은 코드가 생길 수 있습니다.
if ("TCP".equals(protocol)) {
if ("ENV".equals(serviceType)) {
// 환경센서 처리
} else if ("CCTV".equals(serviceType)) {
// CCTV 처리
}
} else if ("HTTP".equals(protocol)) {
// HTTP 처리
}
서비스 종류와 프로토콜이 증가할수록 관리가 어려워집니다.
Spring Integration에서는 이를 각각의 Flow로 분리할 수 있습니다.
TCP Inbound Flow
HTTP Inbound Flow
MQTT Inbound Flow
Parsing Flow
Validation Flow
Environment Flow
CCTV Flow
Traffic Flow
Storage Flow
DataHub Flow
Error Flow
15. Spring Integration의 장점
처리 흐름을 명확하게 표현할 수 있다
Receive
↓
Parse
↓
Validate
↓
Transform
↓
Save
↓
Send
데이터가 어떤 순서로 처리되는지 쉽게 확인할 수 있습니다.
외부 시스템 연동이 편리하다
Adapter를 이용하여 다양한 시스템과 연결할 수 있습니다.
HTTP
TCP
UDP
MQTT
Kafka
JMS
FTP
SFTP
File
프로토콜별 연결 코드와 비즈니스 로직을 분리할 수 있습니다.
데이터 변환 및 분기가 쉽다
Transformer와 Router를 이용하면 시스템별 데이터 규격을 표준화하기 쉽습니다.
센서 A ─┐
센서 B ─┼→ Transformer → 내부 표준 데이터
센서 C ─┘
오류 처리를 통합할 수 있다
각 Service마다 반복적으로 try-catch를 작성하기보다 Error Channel을 이용하여 공통 오류 흐름을 만들 수 있습니다.
모든 처리 단계
↓
Exception
↓
errorChannel
↓
공통 오류 처리
시스템 확장이 편리하다
기존에는 환경센서만 처리하다가 교통센서가 추가되더라도 별도의 Flow를 추가할 수 있습니다.
기존
└── Environment Flow
추가
├── Environment Flow
├── Traffic Flow
└── CCTV Flow
16. Spring Integration의 단점
Spring Integration이 모든 프로젝트에 필요한 것은 아닙니다.
학습해야 할 개념이 많다
일반적인 Spring MVC 개발과 달리 다음 개념을 이해해야 합니다.
Message
Channel
Endpoint
Adapter
Gateway
Router
Transformer
Filter
Flow
처음 사용할 경우 구조가 오히려 복잡하게 느껴질 수 있습니다.
단순 CRUD에는 과하다
예를 들어 다음과 같은 기능에는 Spring Integration을 사용할 이유가 거의 없습니다.
회원 등록
게시판 조회
장비 정보 수정
공지사항 관리
이러한 기능은 일반적인 구조가 더 적합합니다.
Controller
↓
Service
↓
Mapper
↓
Database
Spring Integration은 시스템 연계와 데이터 흐름이 복잡한 영역에서 사용하는 것이 적합합니다.