gc/snp-batch-validation
1
0
포크 0
코드 이슈 풀 리퀘스트 위키 활동

✨ batch_api_log 관리 프로세스 추가

소스 검색
This commit is contained in:
hyojin kim 2026-01-15 15:58:20 +09:00
부모 f2c4e0d14f
커밋 64a3a55e78
33개의 변경된 파일과 854개의 추가작업 그리고 673개의 파일을 삭제
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

177
src/main/java/com/snp/batch/common/batch/reader/BaseApiReader.java
파일 보기

@ -1,5 +1,7 @@
package com.snp.batch.common.batch.reader;
import com.snp.batch.global.model.BatchApiLog;
import com.snp.batch.service.BatchApiLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
@ -7,8 +9,13 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import org.springframework.web.util.UriBuilder;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.util.UriComponentsBuilder;
import java.net.URI;
import java.time.LocalDateTime;
@ -72,12 +79,180 @@ public abstract class BaseApiReader<T> implements ItemReader<T> {
private int totalApiCalls = 0;
private int completedApiCalls = 0;
// Batch Execution Id
private Long jobExecutionId; // 현재 Job 실행 ID
private Long stepExecutionId; // 현재 Step 실행 ID
/**
* 스프링 배치가 Step을 시작할 때 실행 ID를 주입해줍니다.
*/
public void setExecutionIds(Long jobExecutionId, Long stepExecutionId) {
this.jobExecutionId = jobExecutionId;
this.stepExecutionId = stepExecutionId;
}
/**
* 기본 생성자 (WebClient 없이 사용 - Mock 데이터용)
*/
protected BaseApiReader() {
this.webClient = null;
}
/**
* API 호출 및 로그 적재 통합 메서드
* Response Json 구조 : [...]
*/
protected <R> List<R> executeListApiCall(
String baseUrl,
String path,
Map<String, String> params,
ParameterizedTypeReference<List<R>> typeReference,
BatchApiLogService logService) {
// 1. 전체 URI 생성 (로그용)
MultiValueMap<String, String> multiValueParams = new LinkedMultiValueMap<>();
if (params != null) {
params.forEach((key, value) ->
multiValueParams.put(key, Collections.singletonList(value))
);
}
String fullUri = UriComponentsBuilder.fromHttpUrl(baseUrl)
.path(path)
.queryParams(multiValueParams)
.build()
.toUriString();
long startTime = System.currentTimeMillis();
int statusCode = 200;
String errorMessage = null;
Long responseSize = 0L;
try {
log.info("[{}] API 요청 시작: {}", getReaderName(), fullUri);
List<R> result = webClient.get()
.uri(uriBuilder -> {
uriBuilder.path(path);
if (params != null) params.forEach(uriBuilder::queryParam);
return uriBuilder.build();
})
.retrieve()
.bodyToMono(typeReference)
.block();
responseSize = (result != null) ? (long) result.size() : 0L;
return result;
} catch (WebClientResponseException e) {
// API 서버에서 응답은 왔으나 에러인 경우 (4xx, 5xx)
statusCode = e.getStatusCode().value();
errorMessage = String.format("API Error: %s", e.getResponseBodyAsString());
throw e;
} catch (Exception e) {
// 네트워크 오류, 타임아웃 등 기타 예외
statusCode = 500;
errorMessage = String.format("System Error: %s", e.getMessage());
throw e;
} finally {
// 성공/실패 여부와 관계없이 무조건 로그 저장
long duration = System.currentTimeMillis() - startTime;
logService.saveLog(BatchApiLog.builder()
.apiRequestLocation(getReaderName())
.requestUri(fullUri)
.httpMethod("GET")
.statusCode(statusCode)
.responseTimeMs(duration)
.responseCount(responseSize)
.errorMessage(errorMessage)
.createdAt(LocalDateTime.now())
.jobExecutionId(this.jobExecutionId) // 추가
.stepExecutionId(this.stepExecutionId) // 추가
.build());
}
}
/**
* API 호출 및 로그 적재 통합 메서드
* Response Json 구조 : { "data": [...] }
*/
protected <R> R executeSingleApiCall(
String baseUrl,
String path,
Map<String, String> params,
ParameterizedTypeReference<R> typeReference,
BatchApiLogService logService,
Function<R, Long> sizeExtractor) { // 사이즈 추출 함수 추가
// 1. 전체 URI 생성 (로그용)
MultiValueMap<String, String> multiValueParams = new LinkedMultiValueMap<>();
if (params != null) {
params.forEach((key, value) ->
multiValueParams.put(key, Collections.singletonList(value))
);
}
String fullUri = UriComponentsBuilder.fromHttpUrl(baseUrl)
.path(path)
.queryParams(multiValueParams)
.build()
.toUriString();
long startTime = System.currentTimeMillis();
int statusCode = 200;
String errorMessage = null;
R result = null;
try {
log.info("[{}] Single API 요청 시작: {}", getReaderName(), fullUri);
result = webClient.get()
.uri(uriBuilder -> {
uriBuilder.path(path);
if (params != null) params.forEach(uriBuilder::queryParam);
return uriBuilder.build();
})
.retrieve()
.bodyToMono(typeReference)
.block();
return result;
} catch (WebClientResponseException e) {
statusCode = e.getStatusCode().value();
errorMessage = String.format("API Error: %s", e.getResponseBodyAsString());
throw e;
} catch (Exception e) {
statusCode = 500;
errorMessage = String.format("System Error: %s", e.getMessage());
throw e;
} finally {
long duration = System.currentTimeMillis() - startTime;
// 2. 주입받은 함수를 통해 데이터 건수(size) 계산
long size = 0L;
if (result != null && sizeExtractor != null) {
try {
size = sizeExtractor.apply(result);
} catch (Exception e) {
log.warn("[{}] 사이즈 추출 중 오류 발생: {}", getReaderName(), e.getMessage());
}
}
// 3. 로그 저장 (api_request_location, response_size 반영)
logService.saveLog(BatchApiLog.builder()
.apiRequestLocation(getReaderName())
.jobExecutionId(this.jobExecutionId)
.stepExecutionId(this.stepExecutionId)
.requestUri(fullUri)
.httpMethod("GET")
.statusCode(statusCode)
.responseTimeMs(duration)
.responseCount(size)
.errorMessage(errorMessage)
.createdAt(LocalDateTime.now())
.build());
}
}
/**
* WebClient를 주입받는 생성자 (실제 API 연동용)
@ -87,7 +262,7 @@ public abstract class BaseApiReader<T> implements ItemReader<T> {
protected BaseApiReader(WebClient webClient) {
this.webClient = webClient;
}
/**
/**
* Step 실행 전 초기화 및 API 정보 저장
* Spring Batch가 자동으로 StepExecution을 주입하고 이 메서드를 호출함

24
src/main/java/com/snp/batch/global/config/AsyncConfig.java Normal file
파일 보기

@ -0,0 +1,24 @@
package com.snp.batch.global.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
@EnableAsync // 비동기 기능 활성화
public class AsyncConfig {
@Bean(name = "apiLogExecutor")
public Executor apiLogExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2); // 기본 스레드 수
executor.setMaxPoolSize(5); // 최대 스레드 수
executor.setQueueCapacity(500); // 대기 큐 크기
executor.setThreadNamePrefix("ApiLogThread-");
executor.initialize();
return executor;
}
}

10
src/main/java/com/snp/batch/global/config/MaritimeApiWebClientConfig.java
파일 보기

@ -61,11 +61,6 @@ public class MaritimeApiWebClientConfig {
return WebClient.builder()
.baseUrl(maritimeApiUrl)
.filter((request, next) -> {
// [핵심] 여기서 최종 완성된 URI를 로그로 출력합니다.
log.info(">>>> API Request: [{} {}]", request.method(), request.url());
return next.exchange(request);
})
.defaultHeaders(headers -> headers.setBasicAuth(maritimeApiUsername, maritimeApiPassword))
.codecs(configurer -> configurer
.defaultCodecs()
@ -98,11 +93,6 @@ public class MaritimeApiWebClientConfig {
return WebClient.builder()
.baseUrl(maritimeServiceApiUrl)
.filter((request, next) -> {
// [핵심] 여기서 최종 완성된 URI를 로그로 출력합니다.
log.info(">>>> API Request: [{} {}]", request.method(), request.url());
return next.exchange(request);
})
.defaultHeaders(headers -> headers.setBasicAuth(maritimeApiUsername, maritimeApiPassword))
.codecs(configurer -> configurer
.defaultCodecs()

46
src/main/java/com/snp/batch/global/model/BatchApiLog.java Normal file
파일 보기

@ -0,0 +1,46 @@
package com.snp.batch.global.model;
import jakarta.persistence.*;
import lombok.*;
import org.hibernate.annotations.CreationTimestamp;
import java.time.LocalDateTime;
@Entity
@Table(name = "batch_api_log", schema = "snp_data")
@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@AllArgsConstructor
@Builder
public class BatchApiLog {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY) // PostgreSQL BIGSERIAL과 매핑
private Long logId;
@Column(name = "api_request_location") // job_name에서 변경
private String apiRequestLocation;
@Column(columnDefinition = "TEXT", nullable = false)
private String requestUri;
@Column(nullable = false, length = 10)
private String httpMethod;
private Integer statusCode;
private Long responseTimeMs;
@Column(name = "response_count")
private Long responseCount;
@Column(columnDefinition = "TEXT")
private String errorMessage;
@CreationTimestamp // 엔티티가 생성될 때 자동으로 시간 설정
@Column(updatable = false)
private LocalDateTime createdAt;
private Long jobExecutionId; // 추가
private Long stepExecutionId; // 추가
}

10
src/main/java/com/snp/batch/global/repository/BatchApiLogRepository.java Normal file
파일 보기

@ -0,0 +1,10 @@
package com.snp.batch.global.repository;
import com.snp.batch.global.model.BatchApiLog;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
@Repository
public interface BatchApiLogRepository extends JpaRepository<BatchApiLog, Long> {
}

19
src/main/java/com/snp/batch/jobs/compliance/batch/config/ComplianceImportRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.compliance.batch.entity.ComplianceEntity;
import com.snp.batch.jobs.compliance.batch.processor.ComplianceDataProcessor;
import com.snp.batch.jobs.compliance.batch.reader.ComplianceDataRangeReader;
import com.snp.batch.jobs.compliance.batch.writer.ComplianceDataWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -39,6 +41,10 @@ public class ComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<Compl
private final ComplianceDataWriter complianceDataWriter;
private final ComplianceDataRangeReader complianceDataRangeReader;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "COMPLIANCE_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -55,7 +61,8 @@ public class ComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<Compl
JdbcTemplate jdbcTemplate,
@Qualifier("maritimeServiceApiWebClient")WebClient maritimeServiceApiWebClient,
ComplianceDataRangeReader complianceDataRangeReader,
BatchDateService batchDateService) {
BatchDateService batchDateService,
BatchApiLogService batchApiLogService) {
super(jobRepository, transactionManager);
this.jdbcTemplate = jdbcTemplate;
this.maritimeServiceApiWebClient = maritimeServiceApiWebClient;
@ -63,6 +70,7 @@ public class ComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<Compl
this.complianceDataWriter = complianceDataWriter;
this.complianceDataRangeReader = complianceDataRangeReader;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -92,8 +100,13 @@ public class ComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<Compl
@Bean
@StepScope
public ComplianceDataRangeReader complianceDataRangeReader() {
return new ComplianceDataRangeReader(maritimeServiceApiWebClient, jdbcTemplate, batchDateService);
public ComplianceDataRangeReader complianceDataRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
ComplianceDataRangeReader reader = new ComplianceDataRangeReader(maritimeServiceApiWebClient, jdbcTemplate, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<ComplianceDto, ComplianceEntity> createProcessor() {

63
src/main/java/com/snp/batch/jobs/compliance/batch/reader/ComplianceDataRangeReader.java
파일 보기

@ -2,13 +2,12 @@ package com.snp.batch.jobs.compliance.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.compliance.batch.dto.ComplianceDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpStatusCode;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -18,15 +17,18 @@ public class ComplianceDataRangeReader extends BaseApiReader<ComplianceDto> {
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
String maritimeServiceApiUrl;
private List<ComplianceDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
private String fromDate;
private String toDate;
public ComplianceDataRangeReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService) {
public ComplianceDataRangeReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@ -34,13 +36,6 @@ public class ComplianceDataRangeReader extends BaseApiReader<ComplianceDto> {
protected String getReaderName() {
return "ComplianceDataRangeReader";
}
@Override
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
protected String getApiPath() {
return "/RiskAndCompliance/UpdatedComplianceList";
@ -50,6 +45,12 @@ public class ComplianceDataRangeReader extends BaseApiReader<ComplianceDto> {
return "COMPLIANCE_IMPORT_API";
}
@Override
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
protected List<ComplianceDto> fetchNextBatch() throws Exception {
// 모든 배치 처리 완료 확인
@ -102,36 +103,14 @@ public class ComplianceDataRangeReader extends BaseApiReader<ComplianceDto> {
private List<ComplianceDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
String url = getApiPath();
log.debug("[{}] API 호출: {}", getReaderName(), url);
return webClient.get()
.uri(url, uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("fromDate", params.get("fromDate"))
.queryParam("toDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(new ParameterizedTypeReference<List<ComplianceDto>>() {})
.block();
// 부모 클래스의 공통 모듈 호출 (단 한 줄로 처리 가능)
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<ComplianceDto>>() {},
batchApiLogService
);
}
}

35
src/main/java/com/snp/batch/jobs/event/batch/config/EventImportJobConfig.java
파일 보기

@ -6,10 +6,12 @@ import com.snp.batch.jobs.event.batch.entity.EventDetailEntity;
import com.snp.batch.jobs.event.batch.processor.EventDataProcessor;
import com.snp.batch.jobs.event.batch.reader.EventDataReader;
import com.snp.batch.jobs.event.batch.writer.EventDataWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
@ -19,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -28,13 +31,16 @@ import org.springframework.web.reactive.function.client.WebClient;
@Slf4j
@Configuration
public class EventImportJobConfig extends BaseMultiStepJobConfig<EventDetailDto, EventDetailEntity> {
private final EventDataProcessor eventDataProcessor;
private final EventDataWriter eventDataWriter;
private final EventDataReader eventDataReader;
private final JdbcTemplate jdbcTemplate;
private final WebClient maritimeApiWebClient;
private final EventDataProcessor eventDataProcessor;
private final EventDataWriter eventDataWriter;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.ship-api.url}")
private String maritimeApiUrl;
protected String getApiKey() {return "EVENT_IMPORT_API";}
protected String getBatchUpdateSql() {
@ -42,22 +48,26 @@ public class EventImportJobConfig extends BaseMultiStepJobConfig<EventDetailDto,
@Override
protected int getChunkSize() {
return 10; // API에서 5000개씩 가져오므로 chunk도 5000으로 설정
return 1000; // API에서 5000개씩 가져오므로 chunk도 5000으로 설정
}
public EventImportJobConfig(
JobRepository jobRepository,
PlatformTransactionManager transactionManager,
EventDataProcessor eventDataProcessor,
EventDataWriter eventDataWriter,
EventDataReader eventDataReader,
JdbcTemplate jdbcTemplate,
@Qualifier("maritimeApiWebClient")WebClient maritimeApiWebClient,
BatchDateService batchDateService) {
BatchDateService batchDateService,
BatchApiLogService batchApiLogService) {
super(jobRepository, transactionManager);
this.jdbcTemplate = jdbcTemplate;
this.maritimeApiWebClient = maritimeApiWebClient;
this.eventDataProcessor = eventDataProcessor;
this.eventDataWriter = eventDataWriter;
this.eventDataReader = eventDataReader;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -78,9 +88,20 @@ public class EventImportJobConfig extends BaseMultiStepJobConfig<EventDetailDto,
.build();
}
@Bean
@StepScope
public EventDataReader eventDataReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
EventDataReader reader = new EventDataReader(maritimeApiWebClient, jdbcTemplate, batchDateService, batchApiLogService, maritimeApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemReader<EventDetailDto> createReader() {
return new EventDataReader(maritimeApiWebClient, jdbcTemplate, batchDateService);
return eventDataReader;
}
@Override

56
src/main/java/com/snp/batch/jobs/event/batch/reader/EventDataReader.java
파일 보기

@ -2,8 +2,10 @@ package com.snp.batch.jobs.event.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.event.batch.dto.*;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpStatusCode;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.reactive.function.client.WebClient;
@ -16,14 +18,18 @@ import java.util.stream.Collectors;
@Slf4j
public class EventDataReader extends BaseApiReader<EventDetailDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeApiUrl;
private Map<Long, EventPeriod> eventPeriodMap;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
public EventDataReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService) {
public EventDataReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeApiUrl) {
super(webClient);
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeApiUrl = maritimeApiUrl;
enableChunkMode(); // ✨ Chunk 모드 활성화
}
@ -31,16 +37,13 @@ public class EventDataReader extends BaseApiReader<EventDetailDto> {
protected String getReaderName() {
return "EventDataReader";
}
@Override
protected String getApiPath() {
return "/MaritimeWCF/MaritimeAndTradeEventsService.svc/RESTFul/GetEventListByEventChangeDateRange";
}
protected String getEventDetailApiPath() {
return "/MaritimeWCF/MaritimeAndTradeEventsService.svc/RESTFul/GetEventDataByEventID";
}
protected String getApiKey() {
return "EVENT_IMPORT_API";
}
@ -126,9 +129,9 @@ public class EventDataReader extends BaseApiReader<EventDetailDto> {
// API 호출 통계 업데이트
updateApiCallStats(totalBatches, currentBatchNumber);
// API 과부하 방지 (다음 배치 전 0.5초 대기)
// API 과부하 방지 (다음 배치 전 1.0초 대기)
if (currentBatchIndex < eventIds.size()) {
Thread.sleep(500);
Thread.sleep(1000);
}
return eventDetailList;
@ -188,37 +191,14 @@ public class EventDataReader extends BaseApiReader<EventDetailDto> {
private EventResponse callEventApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithoutTimeParams(getApiKey());
String url = getApiPath();
return webClient.get()
.uri(url, uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("fromYear", params.get("fromYear"))
.queryParam("fromMonth", params.get("fromMonth"))
.queryParam("fromDay", params.get("fromDay"))
.queryParam("toYear", params.get("toYear"))
.queryParam("toMonth", params.get("toMonth"))
.queryParam("toDay", params.get("toDay"))
.build())
.retrieve()
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(EventResponse.class)
.block();
return executeSingleApiCall(
maritimeApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<EventResponse>() {},
batchApiLogService,
res -> res.getMaritimeEvents() != null ? (long) res.getMaritimeEvents().size() : 0L // 람다 적용
);
}
private EventDetailResponse callEventDetailApiWithBatch(Long eventId) {

19
src/main/java/com/snp/batch/jobs/movement/batch/config/AnchorageCallsRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.AnchorageCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.AnchorageCallsProcessor;
import com.snp.batch.jobs.movement.batch.reader.AnchorageCallsRangeReader;
import com.snp.batch.jobs.movement.batch.writer.AnchorageCallsWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class AnchorageCallsRangeJobConfig extends BaseMultiStepJobConfig<Anchora
private final WebClient maritimeServiceApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "ANCHORAGE_CALLS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -49,7 +55,8 @@ public class AnchorageCallsRangeJobConfig extends BaseMultiStepJobConfig<Anchora
AnchorageCallsRangeReader anchorageCallsRangeReader,
@Qualifier("maritimeServiceApiWebClient")WebClient maritimeServiceApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) {
super(jobRepository, transactionManager);
this.anchorageCallsProcessor = anchorageCallsProcessor;
@ -58,6 +65,7 @@ public class AnchorageCallsRangeJobConfig extends BaseMultiStepJobConfig<Anchora
this.maritimeServiceApiWebClient = maritimeServiceApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -85,8 +93,13 @@ public class AnchorageCallsRangeJobConfig extends BaseMultiStepJobConfig<Anchora
@Bean
@StepScope
public AnchorageCallsRangeReader anchorageCallsReader() {
return new AnchorageCallsRangeReader(maritimeServiceApiWebClient, batchDateService);
public AnchorageCallsRangeReader anchorageCallsReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
AnchorageCallsRangeReader reader = new AnchorageCallsRangeReader(maritimeServiceApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override

19
src/main/java/com/snp/batch/jobs/movement/batch/config/BerthCallsRangJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entiity.BerthCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.BerthCallsProcessor;
import com.snp.batch.jobs.movement.batch.reader.BerthCallsRangeReader;
import com.snp.batch.jobs.movement.batch.writer.BerthCallsWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class BerthCallsRangJobConfig extends BaseMultiStepJobConfig<BerthCallsDt
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "BERTH_CALLS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -48,7 +54,8 @@ public class BerthCallsRangJobConfig extends BaseMultiStepJobConfig<BerthCallsDt
BerthCallsRangeReader berthCallsRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) {
super(jobRepository, transactionManager);
this.berthCallsProcessor = berthCallsProcessor;
@ -57,6 +64,7 @@ public class BerthCallsRangJobConfig extends BaseMultiStepJobConfig<BerthCallsDt
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -83,8 +91,13 @@ public class BerthCallsRangJobConfig extends BaseMultiStepJobConfig<BerthCallsDt
}
@Bean
@StepScope
public BerthCallsRangeReader berthCallsRangeReader() {
return new BerthCallsRangeReader(maritimeApiWebClient, batchDateService);
public BerthCallsRangeReader berthCallsRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
BerthCallsRangeReader reader = new BerthCallsRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<BerthCallsDto, BerthCallsEntity> createProcessor() {

19
src/main/java/com/snp/batch/jobs/movement/batch/config/CurrentlyAtRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.CurrentlyAtEntity;
import com.snp.batch.jobs.movement.batch.processor.CurrentlyAtProcessor;
import com.snp.batch.jobs.movement.batch.reader.CurrentlyAtRangeReader;
import com.snp.batch.jobs.movement.batch.writer.CurrentlyAtWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class CurrentlyAtRangeJobConfig extends BaseMultiStepJobConfig<CurrentlyA
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "CURRENTLY_AT_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -48,7 +54,8 @@ public class CurrentlyAtRangeJobConfig extends BaseMultiStepJobConfig<CurrentlyA
CurrentlyAtRangeReader currentlyAtRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.currentlyAtProcessor = currentlyAtProcessor;
@ -57,6 +64,7 @@ public class CurrentlyAtRangeJobConfig extends BaseMultiStepJobConfig<CurrentlyA
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -83,8 +91,13 @@ public class CurrentlyAtRangeJobConfig extends BaseMultiStepJobConfig<CurrentlyA
}
@Bean
@StepScope
public CurrentlyAtRangeReader currentlyAtReader() {
return new CurrentlyAtRangeReader(maritimeApiWebClient, batchDateService);
public CurrentlyAtRangeReader currentlyAtReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
CurrentlyAtRangeReader reader = new CurrentlyAtRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<CurrentlyAtDto, CurrentlyAtEntity> createProcessor() {

19
src/main/java/com/snp/batch/jobs/movement/batch/config/DestinationsRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.DestinationEntity;
import com.snp.batch.jobs.movement.batch.processor.DestinationProcessor;
import com.snp.batch.jobs.movement.batch.reader.DestinationRangeReader;
import com.snp.batch.jobs.movement.batch.writer.DestinationWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class DestinationsRangeJobConfig extends BaseMultiStepJobConfig<Destinati
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "DESTINATIONS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -49,7 +55,8 @@ public class DestinationsRangeJobConfig extends BaseMultiStepJobConfig<Destinati
DestinationRangeReader destinationRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.DestinationProcessor = DestinationProcessor;
@ -58,6 +65,7 @@ public class DestinationsRangeJobConfig extends BaseMultiStepJobConfig<Destinati
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -84,8 +92,13 @@ public class DestinationsRangeJobConfig extends BaseMultiStepJobConfig<Destinati
}
@Bean
@StepScope
public DestinationRangeReader destinationRangeReader() {
return new DestinationRangeReader(maritimeApiWebClient, batchDateService);
public DestinationRangeReader destinationRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
DestinationRangeReader reader = new DestinationRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<DestinationDto, DestinationEntity> createProcessor() {

19
src/main/java/com/snp/batch/jobs/movement/batch/config/ShipPortCallsRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.PortCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.PortCallsProcessor;
import com.snp.batch.jobs.movement.batch.reader.PortCallsRangeReader;
import com.snp.batch.jobs.movement.batch.writer.PortCallsWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class ShipPortCallsRangeJobConfig extends BaseMultiStepJobConfig<PortCall
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "PORT_CALLS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -48,7 +54,8 @@ public class ShipPortCallsRangeJobConfig extends BaseMultiStepJobConfig<PortCall
PortCallsRangeReader portCallsRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) {
super(jobRepository, transactionManager);
this.portCallsProcessor = portCallsProcessor;
@ -57,6 +64,7 @@ public class ShipPortCallsRangeJobConfig extends BaseMultiStepJobConfig<PortCall
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -71,8 +79,13 @@ public class ShipPortCallsRangeJobConfig extends BaseMultiStepJobConfig<PortCall
@Bean
@StepScope
public PortCallsRangeReader portCallsRangeReader() {
return new PortCallsRangeReader(maritimeApiWebClient, batchDateService);
public PortCallsRangeReader portCallsRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
PortCallsRangeReader reader = new PortCallsRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override

21
src/main/java/com/snp/batch/jobs/movement/batch/config/StsOperationRangeJobConfig.java
파일 보기

@ -1,12 +1,12 @@
package com.snp.batch.jobs.movement.batch.config;
import com.snp.batch.common.batch.config.BaseJobConfig;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.jobs.movement.batch.dto.StsOperationDto;
import com.snp.batch.jobs.movement.batch.entity.StsOperationEntity;
import com.snp.batch.jobs.movement.batch.processor.StsOperationProcessor;
import com.snp.batch.jobs.movement.batch.reader.StsOperationRangeReader;
import com.snp.batch.jobs.movement.batch.writer.StsOperationWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -38,6 +38,10 @@ public class StsOperationRangeJobConfig extends BaseMultiStepJobConfig<StsOperat
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "STS_OPERATION_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -51,7 +55,8 @@ public class StsOperationRangeJobConfig extends BaseMultiStepJobConfig<StsOperat
StsOperationRangeReader stsOperationRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.stsOperationProcessor = stsOperationProcessor;
@ -60,6 +65,7 @@ public class StsOperationRangeJobConfig extends BaseMultiStepJobConfig<StsOperat
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -82,14 +88,17 @@ public class StsOperationRangeJobConfig extends BaseMultiStepJobConfig<StsOperat
@Override
protected ItemReader<StsOperationDto> createReader() { // 타입 변경
// Reader 생성자 수정: ObjectMapper를 전달합니다.
return stsOperationRangeReader;
}
@Bean
@StepScope
public StsOperationRangeReader stsOperationRangeReader() {
// jobParameters 없으면 null 넘어오고 Reader에서 default 처리
return new StsOperationRangeReader(maritimeApiWebClient, batchDateService);
public StsOperationRangeReader stsOperationRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
StsOperationRangeReader reader = new StsOperationRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<StsOperationDto, StsOperationEntity> createProcessor() {

20
src/main/java/com/snp/batch/jobs/movement/batch/config/TerminalCallsRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.TerminalCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.TerminalCallsProcessor;
import com.snp.batch.jobs.movement.batch.reader.TerminalCallsRangeReader;
import com.snp.batch.jobs.movement.batch.writer.TerminalCallsWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class TerminalCallsRangeJobConfig extends BaseMultiStepJobConfig<Terminal
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "TERMINAL_CALLS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -49,7 +55,8 @@ public class TerminalCallsRangeJobConfig extends BaseMultiStepJobConfig<Terminal
TerminalCallsRangeReader terminalCallsRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.terminalCallsProcessor = terminalCallsProcessor;
@ -58,6 +65,7 @@ public class TerminalCallsRangeJobConfig extends BaseMultiStepJobConfig<Terminal
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -84,9 +92,13 @@ public class TerminalCallsRangeJobConfig extends BaseMultiStepJobConfig<Terminal
}
@Bean
@StepScope
public TerminalCallsRangeReader terminalCallsRangeReader() {
// jobParameters 없으면 null 넘어오고 Reader에서 default 처리
return new TerminalCallsRangeReader(maritimeApiWebClient, batchDateService);
public TerminalCallsRangeReader terminalCallsRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
TerminalCallsRangeReader reader = new TerminalCallsRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<TerminalCallsDto, TerminalCallsEntity> createProcessor() {

19
src/main/java/com/snp/batch/jobs/movement/batch/config/TransitsRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.movement.batch.entity.TransitsEntity;
import com.snp.batch.jobs.movement.batch.processor.TransitsProcessor;
import com.snp.batch.jobs.movement.batch.reader.TransitsRangeReader;
import com.snp.batch.jobs.movement.batch.writer.TransitsWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -36,6 +38,10 @@ public class TransitsRangeJobConfig extends BaseMultiStepJobConfig<TransitsDto,
private final WebClient maritimeApiWebClient;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "TRANSITS_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -48,7 +54,8 @@ public class TransitsRangeJobConfig extends BaseMultiStepJobConfig<TransitsDto,
TransitsRangeReader transitsRangeReader,
@Qualifier("maritimeServiceApiWebClient") WebClient maritimeApiWebClient,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService
BatchDateService batchDateService,
BatchApiLogService batchApiLogService
) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.transitsProcessor = TransitsProcessor;
@ -57,6 +64,7 @@ public class TransitsRangeJobConfig extends BaseMultiStepJobConfig<TransitsDto,
this.maritimeApiWebClient = maritimeApiWebClient;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -83,8 +91,13 @@ public class TransitsRangeJobConfig extends BaseMultiStepJobConfig<TransitsDto,
}
@Bean
@StepScope
public TransitsRangeReader transitsRangeReader() {
return new TransitsRangeReader(maritimeApiWebClient, batchDateService);
public TransitsRangeReader transitsRangeReaderanchorageCallsReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
TransitsRangeReader reader = new TransitsRangeReader(maritimeApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemProcessor<TransitsDto, TransitsEntity> createProcessor() {

72
src/main/java/com/snp/batch/jobs/movement/batch/reader/AnchorageCallsRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.AnchorageCallsDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,23 +16,31 @@ import java.util.Map;
@StepScope
public class AnchorageCallsRangeReader extends BaseApiReader<AnchorageCallsDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<AnchorageCallsDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "ANCHORAGE_CALLS_IMPORT_API";
}
public AnchorageCallsRangeReader(WebClient webClient, BatchDateService batchDateService) {
public AnchorageCallsRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@Override
protected String getReaderName() {
return "AnchorageCallsReader";
return "AnchorageCallsRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/AnchorageCalls";
}
protected String getApiKey() {
return "ANCHORAGE_CALLS_IMPORT_API";
}
@Override
@ -41,16 +49,6 @@ public class AnchorageCallsRangeReader extends BaseApiReader<AnchorageCallsDto>
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/AnchorageCalls";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
}
@Override
protected List<AnchorageCallsDto> fetchNextBatch() throws Exception {
// 1) 처음 호출이면 API 한 번 호출해서 전체 데이터를 가져온다
@ -94,36 +92,14 @@ public class AnchorageCallsRangeReader extends BaseApiReader<AnchorageCallsDto>
* @return API 응답
*/
private List<AnchorageCallsDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(AnchorageCallsDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<AnchorageCallsDto>>() {},
batchApiLogService
);
}
@Override

68
src/main/java/com/snp/batch/jobs/movement/batch/reader/BerthCallsRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.BerthCallsDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,16 +16,16 @@ import java.util.Map;
@StepScope
public class BerthCallsRangeReader extends BaseApiReader<BerthCallsDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<BerthCallsDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "BERTH_CALLS_IMPORT_API";
}
public BerthCallsRangeReader(WebClient webClient, BatchDateService batchDateService) {
public BerthCallsRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@ -33,21 +33,19 @@ public class BerthCallsRangeReader extends BaseApiReader<BerthCallsDto> {
protected String getReaderName() {
return "BerthCallsRangeReader";
}
@Override
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/BerthCalls";
}
protected String getApiKey() {
return "BERTH_CALLS_IMPORT_API";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
@ -93,36 +91,14 @@ public class BerthCallsRangeReader extends BaseApiReader<BerthCallsDto> {
* @return API 응답
*/
private List<BerthCallsDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(BerthCallsDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<BerthCallsDto>>() {},
batchApiLogService
);
}
@Override

72
src/main/java/com/snp/batch/jobs/movement/batch/reader/CurrentlyAtRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.CurrentlyAtDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,21 +16,29 @@ import java.util.Map;
@StepScope
public class CurrentlyAtRangeReader extends BaseApiReader<CurrentlyAtDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<CurrentlyAtDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "CURRENTLY_AT_IMPORT_API";
}
public CurrentlyAtRangeReader(WebClient webClient, BatchDateService batchDateService) {
public CurrentlyAtRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@Override
protected String getReaderName() {
return "CurrentlyAtReader";
return "CurrentlyAtRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/CurrentlyAt";
}
protected String getApiKey() {
return "CURRENTLY_AT_IMPORT_API";
}
@Override
@ -39,16 +47,6 @@ public class CurrentlyAtRangeReader extends BaseApiReader<CurrentlyAtDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/CurrentlyAt";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
}
@Override
protected List<CurrentlyAtDto> fetchNextBatch() throws Exception {
@ -92,36 +90,14 @@ public class CurrentlyAtRangeReader extends BaseApiReader<CurrentlyAtDto> {
* @return API 응답
*/
private List<CurrentlyAtDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(CurrentlyAtDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<CurrentlyAtDto>>() {},
batchApiLogService
);
}
@Override

72
src/main/java/com/snp/batch/jobs/movement/batch/reader/DestinationRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.DestinationDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,21 +16,29 @@ import java.util.Map;
@StepScope
public class DestinationRangeReader extends BaseApiReader<DestinationDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<DestinationDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "DESTINATIONS_IMPORT_API";
}
public DestinationRangeReader(WebClient webClient, BatchDateService batchDateService) {
public DestinationRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@Override
protected String getReaderName() {
return "DestinationsRangeReader";
return "DestinationRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/Destinations";
}
protected String getApiKey() {
return "DESTINATIONS_IMPORT_API";
}
@Override
@ -39,16 +47,6 @@ public class DestinationRangeReader extends BaseApiReader<DestinationDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/Destinations";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
}
@Override
protected List<DestinationDto> fetchNextBatch() throws Exception {
@ -90,36 +88,14 @@ public class DestinationRangeReader extends BaseApiReader<DestinationDto> {
* @return API 응답
*/
private List<DestinationDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(DestinationDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<DestinationDto>>() {},
batchApiLogService
);
}
@Override

66
src/main/java/com/snp/batch/jobs/movement/batch/reader/PortCallsRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.PortCallsDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,15 +16,16 @@ import java.util.Map;
@StepScope
public class PortCallsRangeReader extends BaseApiReader<PortCallsDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<PortCallsDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "PORT_CALLS_IMPORT_API";
}
public PortCallsRangeReader(WebClient webClient, BatchDateService batchDateService) {
public PortCallsRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@ -32,21 +33,19 @@ public class PortCallsRangeReader extends BaseApiReader<PortCallsDto> {
protected String getReaderName() {
return "PortCallsRangeReader";
}
@Override
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/PortCalls";
}
protected String getApiKey() {
return "PORT_CALLS_IMPORT_API";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
protected void resetCustomState() {
this.currentBatchIndex = 0;
this.allData = null;
}
@Override
@ -93,35 +92,14 @@ public class PortCallsRangeReader extends BaseApiReader<PortCallsDto> {
* @return API 응답
*/
private List<PortCallsDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(PortCallsDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<PortCallsDto>>() {},
batchApiLogService
);
}
@Override

65
src/main/java/com/snp/batch/jobs/movement/batch/reader/StsOperationRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.StsOperationDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,16 +16,16 @@ import java.util.Map;
@StepScope
public class StsOperationRangeReader extends BaseApiReader<StsOperationDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<StsOperationDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "STS_OPERATION_IMPORT_API";
}
public StsOperationRangeReader(WebClient webClient, BatchDateService batchDateService) {
public StsOperationRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@ -33,6 +33,14 @@ public class StsOperationRangeReader extends BaseApiReader<StsOperationDto> {
protected String getReaderName() {
return "StsOperationRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/StsOperations";
}
protected String getApiKey() {
return "STS_OPERATION_IMPORT_API";
}
@Override
protected void resetCustomState() {
@ -40,11 +48,6 @@ public class StsOperationRangeReader extends BaseApiReader<StsOperationDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/StsOperations";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
@ -92,36 +95,14 @@ public class StsOperationRangeReader extends BaseApiReader<StsOperationDto> {
* @return API 응답
*/
private List<StsOperationDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(StsOperationDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<StsOperationDto>>() {},
batchApiLogService
);
}
@Override

70
src/main/java/com/snp/batch/jobs/movement/batch/reader/TerminalCallsRangeReader.java
파일 보기

@ -2,12 +2,12 @@ package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.TerminalCallsDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -16,22 +16,30 @@ import java.util.Map;
@StepScope
public class TerminalCallsRangeReader extends BaseApiReader<TerminalCallsDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<TerminalCallsDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "TERMINAL_CALLS_IMPORT_API";
}
public TerminalCallsRangeReader(WebClient webClient, BatchDateService batchDateService) {
public TerminalCallsRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@Override
protected String getReaderName() {
return "TerminalCallsRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/TerminalCalls";
}
protected String getApiKey() {
return "TERMINAL_CALLS_IMPORT_API";
}
@Override
protected void resetCustomState() {
@ -39,16 +47,6 @@ public class TerminalCallsRangeReader extends BaseApiReader<TerminalCallsDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/TerminalCalls";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
}
@Override
protected List<TerminalCallsDto> fetchNextBatch() throws Exception {
@ -91,36 +89,14 @@ public class TerminalCallsRangeReader extends BaseApiReader<TerminalCallsDto> {
* @return API 응답
*/
private List<TerminalCallsDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(TerminalCallsDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<TerminalCallsDto>>() {},
batchApiLogService
);
}
@Override

64
src/main/java/com/snp/batch/jobs/movement/batch/reader/TransitsRangeReader.java
파일 보기

@ -1,10 +1,13 @@
package com.snp.batch.jobs.movement.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.movement.batch.dto.AnchorageCallsDto;
import com.snp.batch.jobs.movement.batch.dto.TransitsDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
@ -16,22 +19,30 @@ import java.util.Map;
@StepScope
public class TransitsRangeReader extends BaseApiReader<TransitsDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeServiceApiUrl;
private List<TransitsDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
protected String getApiKey() {
return "TRANSITS_IMPORT_API";
}
public TransitsRangeReader(WebClient webClient, BatchDateService batchDateService) {
public TransitsRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@Override
protected String getReaderName() {
return "TransitsRangeReader";
}
@Override
protected String getApiPath() {
return "/Movements/Transits";
}
protected String getApiKey() {
return "TRANSITS_IMPORT_API";
}
@Override
protected void resetCustomState() {
@ -39,11 +50,6 @@ public class TransitsRangeReader extends BaseApiReader<TransitsDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/Movements/Transits";
}
@Override
protected String getApiBaseUrl() {
return "https://webservices.maritime.spglobal.com";
@ -91,36 +97,14 @@ public class TransitsRangeReader extends BaseApiReader<TransitsDto> {
* @return API 응답
*/
private List<TransitsDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
return webClient.get()
.uri(getApiPath(), uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("startDate", params.get("fromDate"))
.queryParam("stopDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToFlux(TransitsDto.class)
.collectList()
.block();
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey(), "startDate", "stopDate");
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<TransitsDto>>() {},
batchApiLogService
);
}
@Override

27
src/main/java/com/snp/batch/jobs/pscInspection/batch/config/PscInspectionJobConfig.java
파일 보기

@ -6,10 +6,12 @@ import com.snp.batch.jobs.pscInspection.batch.entity.PscInspectionEntity;
import com.snp.batch.jobs.pscInspection.batch.processor.PscInspectionProcessor;
import com.snp.batch.jobs.pscInspection.batch.reader.PscApiReader;
import com.snp.batch.jobs.pscInspection.batch.writer.PscInspectionWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
@ -19,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -31,9 +34,14 @@ public class PscInspectionJobConfig extends BaseMultiStepJobConfig<PscInspection
private final PscInspectionProcessor pscInspectionProcessor;
private final PscInspectionWriter pscInspectionWriter;
private final PscApiReader pscApiReader;
private final JdbcTemplate jdbcTemplate;
private final WebClient maritimeApiWebClient;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.ship-api.url}")
private String maritimeApiUrl;
protected String getApiKey() {return "PSC_IMPORT_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -43,15 +51,19 @@ public class PscInspectionJobConfig extends BaseMultiStepJobConfig<PscInspection
PlatformTransactionManager transactionManager,
PscInspectionProcessor pscInspectionProcessor,
PscInspectionWriter pscInspectionWriter,
PscApiReader pscApiReader,
JdbcTemplate jdbcTemplate,
BatchDateService batchDateService,
@Qualifier("maritimeApiWebClient") WebClient maritimeApiWebClient) { // ObjectMapper 주입 추가
@Qualifier("maritimeApiWebClient") WebClient maritimeApiWebClient,
BatchApiLogService batchApiLogService) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.pscInspectionProcessor = pscInspectionProcessor;
this.pscInspectionWriter = pscInspectionWriter;
this.pscApiReader = pscApiReader;
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.maritimeApiWebClient = maritimeApiWebClient;
this.batchApiLogService = batchApiLogService;
}
@ -74,9 +86,20 @@ public class PscInspectionJobConfig extends BaseMultiStepJobConfig<PscInspection
.build();
}
@Bean
@StepScope
public PscApiReader pscApiReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
PscApiReader reader = new PscApiReader(maritimeApiWebClient, jdbcTemplate, batchDateService, batchApiLogService, maritimeApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemReader<PscInspectionDto> createReader() {
return new PscApiReader(maritimeApiWebClient, jdbcTemplate, batchDateService);
return pscApiReader;
}
@Override

96
src/main/java/com/snp/batch/jobs/pscInspection/batch/reader/PscApiReader.java
파일 보기

@ -1,16 +1,15 @@
package com.snp.batch.jobs.pscInspection.batch.reader;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.pscInspection.batch.dto.PscApiResponseDto;
import com.snp.batch.jobs.pscInspection.batch.dto.PscInspectionDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.http.HttpStatusCode;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.LocalDate;
import java.time.LocalDateTime;
@ -22,20 +21,20 @@ import java.util.Map;
@Slf4j
@StepScope
public class PscApiReader extends BaseApiReader<PscInspectionDto> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeApiUrl;
private List<PscInspectionDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
protected String getApiKey() {
return "PSC_IMPORT_API";
}
public PscApiReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService) {
public PscApiReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeApiUrl) {
super(webClient);
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeApiUrl = maritimeApiUrl;
enableChunkMode(); // ✨ Chunk 모드 활성화
}
@ -43,6 +42,14 @@ public class PscApiReader extends BaseApiReader<PscInspectionDto> {
protected String getReaderName() {
return "PscApiReader";
}
@Override
protected String getApiPath() {
return "/MaritimeWCF/PSCService.svc/RESTFul/GetPSCDataByLastUpdateDateRange";
}
protected String getApiKey() {
return "PSC_IMPORT_API";
}
@Override
protected void resetCustomState() {
@ -50,11 +57,6 @@ public class PscApiReader extends BaseApiReader<PscInspectionDto> {
this.allData = null;
}
@Override
protected String getApiPath() {
return "/MaritimeWCF/PSCService.svc/RESTFul/GetPSCDataByLastUpdateDateRange";
}
@Override
protected List<PscInspectionDto> fetchNextBatch() {
@ -94,70 +96,30 @@ public class PscApiReader extends BaseApiReader<PscInspectionDto> {
}
private List<PscInspectionDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithoutTimeParams(getApiKey());
String url = getApiPath();
String json = webClient.get()
.uri(url, uriBuilder -> uriBuilder
.queryParam("shipsCategory", params.get("shipsCategory"))
.queryParam("fromYear", params.get("fromYear"))
.queryParam("fromMonth", params.get("fromMonth"))
.queryParam("fromDay", params.get("fromDay"))
.queryParam("toYear", params.get("toYear"))
.queryParam("toMonth", params.get("toMonth"))
.queryParam("toDay", params.get("toDay"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(String.class)
.block();
if (json == null || json.isBlank()) {
log.warn("[PSC] API 응답 없음");
return Collections.emptyList();
// 1. 단일 객체 응답 API 호출
PscApiResponseDto response = executeSingleApiCall(
maritimeApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<PscApiResponseDto>() {},
batchApiLogService,
res -> res.getInspections() != null ? (long) res.getInspections().size() : 0L // 람다 적용
);
// 2. Inspections Array 데이터 추출
if (response != null && response.getInspections() != null) {
log.info("[{}] PSC 데이터 추출 성공 - 건수: {}", getReaderName(), response.getInspections().size());
return response.getInspections();
}
try {
ObjectMapper mapper = new ObjectMapper();
PscApiResponseDto resp = mapper.readValue(json, PscApiResponseDto.class);
if (resp.getInspections() == null) {
log.warn("[PSC] inspections 필드 없음");
return Collections.emptyList();
}
return resp.getInspections();
} catch (Exception e) {
log.error("[PSC] JSON 파싱 실패: {}", e.getMessage());
return Collections.emptyList();
}
}
@Override
protected void afterFetch(List<PscInspectionDto> data) {
if (data == null) {
int totalBatches = (int) Math.ceil((double) allData.size() / batchSize);
log.info("[{}] 전체 {} 개 배치 처리 완료", getReaderName(), totalBatches);
log.info("[{}] 총 {} 개의 IMO 번호에 대한 API 호출 종료",
getReaderName(), allData.size());
}
}

19
src/main/java/com/snp/batch/jobs/risk/batch/config/RiskImportRangeJobConfig.java
파일 보기

@ -6,6 +6,7 @@ import com.snp.batch.jobs.risk.batch.entity.RiskEntity;
import com.snp.batch.jobs.risk.batch.processor.RiskDataProcessor;
import com.snp.batch.jobs.risk.batch.reader.RiskDataRangeReader;
import com.snp.batch.jobs.risk.batch.writer.RiskDataWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
@ -20,6 +21,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -35,6 +37,10 @@ public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, Ri
private final RiskDataRangeReader riskDataRangeReader;
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.webservice-api.url}")
private String maritimeServiceApiUrl;
protected String getApiKey() {return "RISK_IMPORT_API";}
protected String getBatchUpdateSql() {
@ -53,7 +59,8 @@ public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, Ri
JdbcTemplate jdbcTemplate,
@Qualifier("maritimeServiceApiWebClient")WebClient maritimeServiceApiWebClient,
RiskDataRangeReader riskDataRangeReader,
BatchDateService batchDateService) {
BatchDateService batchDateService,
BatchApiLogService batchApiLogService) {
super(jobRepository, transactionManager);
this.maritimeServiceApiWebClient = maritimeServiceApiWebClient;
this.riskDataProcessor = riskDataProcessor;
@ -61,6 +68,7 @@ public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, Ri
this.jdbcTemplate = jdbcTemplate;
this.riskDataRangeReader = riskDataRangeReader;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -88,8 +96,13 @@ public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, Ri
}
@Bean
@StepScope
public RiskDataRangeReader riskDataRangeReader() {
return new RiskDataRangeReader(maritimeServiceApiWebClient, jdbcTemplate, batchDateService);
public RiskDataRangeReader riskDataRangeReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
RiskDataRangeReader reader = new RiskDataRangeReader(maritimeServiceApiWebClient, jdbcTemplate, batchDateService, batchApiLogService, maritimeServiceApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override

46
src/main/java/com/snp/batch/jobs/risk/batch/reader/RiskDataRangeReader.java
파일 보기

@ -2,13 +2,12 @@ package com.snp.batch.jobs.risk.batch.reader;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.jobs.risk.batch.dto.RiskDto;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpStatusCode;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
@ -18,15 +17,19 @@ public class RiskDataRangeReader extends BaseApiReader<RiskDto> {
private final JdbcTemplate jdbcTemplate;
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private List<RiskDto> allData;
private int currentBatchIndex = 0;
private final int batchSize = 5000;
private String fromDate;
private String toDate;
public RiskDataRangeReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService) {
String maritimeServiceApiUrl;
public RiskDataRangeReader(WebClient webClient, JdbcTemplate jdbcTemplate, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
super(webClient);
this.jdbcTemplate = jdbcTemplate;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
enableChunkMode();
}
@ -100,35 +103,14 @@ public class RiskDataRangeReader extends BaseApiReader<RiskDto> {
private List<RiskDto> callApiWithBatch() {
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
// log.info("[{}] 요청 날짜 범위: {} → {}", getReaderName(), params.get("fromDate"), params.get("toDate"));
String url = getApiPath();
return webClient.get()
.uri(url, uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("fromDate", params.get("fromDate"))
.queryParam("toDate", params.get("toDate"))
.build())
.retrieve()
// 1. 에러 상태 코드 감지 (4xx, 5xx)
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(new ParameterizedTypeReference<List<RiskDto>>() {})
.block();
// 부모 클래스의 공통 모듈 호출 (단 한 줄로 처리 가능)
return executeListApiCall(
maritimeServiceApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<List<RiskDto>>() {},
batchApiLogService
);
}

30
src/main/java/com/snp/batch/jobs/shipdetail/batch/config/ShipDetailUpdateJobConfig.java
파일 보기

@ -7,10 +7,12 @@ import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailUpdate;
import com.snp.batch.jobs.shipdetail.batch.processor.ShipDetailDataProcessor;
import com.snp.batch.jobs.shipdetail.batch.reader.ShipDetailUpdateDataReader;
import com.snp.batch.jobs.shipdetail.batch.writer.ShipDetailDataWriter;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
@ -20,6 +22,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
@ -32,10 +35,15 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
private final ShipDetailDataProcessor shipDetailDataProcessor;
private final ShipDetailDataWriter shipDetailDataWriter;
private final ShipDetailUpdateDataReader shipDetailUpdateDataReader;
private final JdbcTemplate jdbcTemplate;
private final WebClient maritimeApiWebClient;
private final ObjectMapper objectMapper; // ObjectMapper 주입 추가
private final BatchDateService batchDateService;
private final BatchApiLogService batchApiLogService;
@Value("${app.batch.ship-api.url}")
private String maritimeApiUrl;
protected String getApiKey() {return "SHIP_DETAIL_UPDATE_API";}
protected String getBatchUpdateSql() {
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
@ -46,17 +54,21 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
PlatformTransactionManager transactionManager,
ShipDetailDataProcessor shipDetailDataProcessor,
ShipDetailDataWriter shipDetailDataWriter,
ShipDetailUpdateDataReader shipDetailUpdateDataReader,
JdbcTemplate jdbcTemplate,
@Qualifier("maritimeApiWebClient") WebClient maritimeApiWebClient,
ObjectMapper objectMapper,
BatchDateService batchDateService) { // ObjectMapper 주입 추가
BatchDateService batchDateService,
BatchApiLogService batchApiLogService) { // ObjectMapper 주입 추가
super(jobRepository, transactionManager);
this.shipDetailDataProcessor = shipDetailDataProcessor;
this.shipDetailDataWriter = shipDetailDataWriter;
this.shipDetailUpdateDataReader = shipDetailUpdateDataReader;
this.jdbcTemplate = jdbcTemplate;
this.maritimeApiWebClient = maritimeApiWebClient;
this.objectMapper = objectMapper; // ObjectMapper 초기화
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
}
@Override
@ -77,10 +89,20 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
.build();
}
@Bean
@StepScope
public ShipDetailUpdateDataReader shipDetailUpdateDataReader(
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
@Value("#{stepExecution.id}") Long stepExecutionId
) {
ShipDetailUpdateDataReader reader = new ShipDetailUpdateDataReader(maritimeApiWebClient, jdbcTemplate, objectMapper, batchDateService, batchApiLogService, maritimeApiUrl);
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
return reader;
}
@Override
protected ItemReader<ShipDetailComparisonData> createReader() { // 타입 변경
// Reader 생성자 수정: ObjectMapper를 전달합니다.
return new ShipDetailUpdateDataReader(maritimeApiWebClient, jdbcTemplate, objectMapper, batchDateService);
return shipDetailUpdateDataReader;
}
@Override
@ -95,7 +117,7 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
@Override
protected int getChunkSize() {
return 20; // API에서 100개씩 가져오므로 chunk도 100으로 설정
return 50; // API에서 100개씩 가져오므로 chunk도 100으로 설정
}
@Bean(name = "ShipDetailUpdateJob")

106
src/main/java/com/snp/batch/jobs/shipdetail/batch/reader/ShipDetailUpdateDataReader.java
파일 보기

@ -5,12 +5,12 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.reader.BaseApiReader;
import com.snp.batch.common.util.JsonChangeDetector;
import com.snp.batch.jobs.shipdetail.batch.dto.*;
import com.snp.batch.service.BatchApiLogService;
import com.snp.batch.service.BatchDateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.HttpStatusCode;
import reactor.core.publisher.Mono;
import java.util.*;
import java.util.stream.Collectors;
@ -18,23 +18,25 @@ import java.util.stream.Collectors;
@Slf4j
public class ShipDetailUpdateDataReader extends BaseApiReader<ShipDetailComparisonData> {
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
private final BatchApiLogService batchApiLogService;
private final String maritimeApiUrl;
private final JdbcTemplate jdbcTemplate;
private final ObjectMapper objectMapper;
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
protected String getApiKey() {return "SHIP_DETAIL_UPDATE_API";}
// 배치 처리 상태
private List<String> allImoNumbers;
// DB 해시값을 저장할 맵
private Map<String, String> dbMasterHashes;
private int currentBatchIndex = 0;
private final int batchSize = 20;
public ShipDetailUpdateDataReader(WebClient webClient, JdbcTemplate jdbcTemplate, ObjectMapper objectMapper,BatchDateService batchDateService) {
private final int batchSize = 50;
public ShipDetailUpdateDataReader(WebClient webClient, JdbcTemplate jdbcTemplate, ObjectMapper objectMapper,BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeApiUrl) {
super(webClient);
this.jdbcTemplate = jdbcTemplate;
this.objectMapper = objectMapper;
this.batchDateService = batchDateService;
this.batchApiLogService = batchApiLogService;
this.maritimeApiUrl = maritimeApiUrl;
enableChunkMode(); // ✨ Chunk 모드 활성화
}
@ -42,6 +44,12 @@ public class ShipDetailUpdateDataReader extends BaseApiReader<ShipDetailComparis
protected String getReaderName() {
return "ShipDetailUpdateDataReader";
}
protected String getShipUpdateApiPath(){ return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipChangesByLastUpdateDateRange"; }
@Override
protected String getApiPath() {
return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipsByIHSLRorIMONumbersAll";
}
protected String getApiKey() {return "SHIP_DETAIL_UPDATE_API";}
@Override
protected void resetCustomState() {
@ -50,15 +58,6 @@ public class ShipDetailUpdateDataReader extends BaseApiReader<ShipDetailComparis
this.dbMasterHashes = null;
}
@Override
protected String getApiPath() {
return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipsByIHSLRorIMONumbersAll";
}
protected String getShipUpdateApiPath(){
return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipChangesByLastUpdateDateRange";
}
private static final String FETCH_ALL_HASHES_QUERY =
"SELECT imo_number, ship_detail_hash FROM snp_data.ship_detail_hash_json ORDER BY imo_number";
@ -220,67 +219,30 @@ public class ShipDetailUpdateDataReader extends BaseApiReader<ShipDetailComparis
* @return API 응답
*/
private ShipDetailApiResponse callApiWithBatch(String imoNumbers) {
String url = getApiPath() + "?IMONumbers=" + imoNumbers;
Map<String, String> params = new HashMap<>();
params.put("IMONumbers", imoNumbers);
return webClient.get()
.uri(url)
.retrieve()
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(ShipDetailApiResponse.class)
.block();
return executeSingleApiCall(
maritimeApiUrl,
getApiPath(),
params,
new ParameterizedTypeReference<ShipDetailApiResponse>() {},
batchApiLogService,
res -> res.getShipResult() != null ? (long) res.getShipResult().size() : 0L // 람다 적용
);
}
private ShipUpdateApiResponse callShipUpdateApi(){
// 1. BatchDateService를 통해 동적 날짜 파라미터 맵 조회
Map<String, String> params = batchDateService.getDateRangeWithoutTimeParams(getApiKey());
String url = getShipUpdateApiPath();
return webClient.get()
.uri(url, uriBuilder -> uriBuilder
// 맵에서 파라미터 값을 동적으로 가져와 세팅
.queryParam("shipsCategory", params.get("shipsCategory"))
.queryParam("fromYear", params.get("fromYear"))
.queryParam("fromMonth", params.get("fromMonth"))
.queryParam("fromDay", params.get("fromDay"))
.queryParam("toYear", params.get("toYear"))
.queryParam("toMonth", params.get("toMonth"))
.queryParam("toDay", params.get("toDay"))
.build())
.retrieve()
.onStatus(HttpStatusCode::isError, clientResponse ->
clientResponse.bodyToMono(String.class) // 에러 바디를 문자열로 읽음
.flatMap(errorBody -> {
// 2. 로그에 상태 코드와 에러 메세지 출력
log.error("[{}] API 호출 오류 발생!", getReaderName());
log.error("[{}] ERROR CODE: {}, REASON: {}",
getReaderName(),
clientResponse.statusCode(),
errorBody);
// 3. 상위로 예외 던지기 (배치 중단을 원할 경우)
return Mono.error(new RuntimeException(
String.format("API 호출 실패 (%s): %s", clientResponse.statusCode(), errorBody)
));
})
)
.bodyToMono(ShipUpdateApiResponse.class)
.block();
return executeSingleApiCall(
maritimeApiUrl,
getShipUpdateApiPath(),
params,
new ParameterizedTypeReference<ShipUpdateApiResponse>() {},
batchApiLogService,
res -> res.getShips() != null ? (long) res.getShips().size() : 0L // 람다 적용
);
}
private List<String> extractUpdateImoNumbers(ShipUpdateApiResponse response) {

33
src/main/java/com/snp/batch/service/BatchApiLogService.java Normal file
파일 보기

@ -0,0 +1,33 @@
package com.snp.batch.service;
import com.snp.batch.global.model.BatchApiLog;
import com.snp.batch.global.repository.BatchApiLogRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
@Service
@RequiredArgsConstructor
@Slf4j
public class BatchApiLogService {
private final BatchApiLogRepository batchApiLogRepository;
/**
* 비동기로 API 로그를 저장합니다.
* propagation = Propagation.REQUIRES_NEW 를 사용하여
* 메인 배치가 실패(Rollback)하더라도 로그는 저장되도록 설정합니다.
*/
@Async("apiLogExecutor") // 설정한 스레드 풀 사용
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void saveLog(BatchApiLog logEntry) {
try {
batchApiLogRepository.save(logEntry);
} catch (Exception e) {
// 로그 저장 실패가 배치를 중단시키지 않도록 여기서 예외 처리
log.error("API 로그 저장 실패: {}", e.getMessage());
}
}
}

24
src/main/java/com/snp/batch/service/BatchDateService.java
파일 보기

@ -82,6 +82,30 @@ public class BatchDateService {
});
}
public Map<String, String> getDateRangeWithTimezoneParams(String apiKey, String dateParam1, String dateParam2) {
return repository.findDateRangeByApiKey(apiKey)
.map(projection -> {
Map<String, String> params = new HashMap<>();
// 'Z'를 문자열 리터럴이 아닌 실제 타임존 기호(X)로 처리
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSX");
// 한국 시간을 UTC로 변환하는 헬퍼 메소드 (아래 정의)
params.put(dateParam1, formatToUtc(projection.getRangeFromDate() != null ?
projection.getRangeFromDate() : projection.getLastSuccessDate(), formatter));
LocalDateTime toDateTime = projection.getRangeToDate() != null ?
projection.getRangeToDate() : LocalDateTime.now();
params.put(dateParam2, formatToUtc(toDateTime, formatter));
return params;
})
.orElseGet(() -> {
log.warn("해당 apiKey에 대한 데이터를 찾을 수 없습니다: {}", apiKey);
return new HashMap<>();
});
}
// 한국 시간(LocalDateTime)을 UTC 문자열로 변환하는 로직
private String formatToUtc(LocalDateTime localDateTime, DateTimeFormatter formatter) {
if (localDateTime == null) return null;