[Java] CompletableFuture 비동기 작업

xml 형식의 서울 실시간 도시 데이터를 가져와서 파싱하고 DB에 저장하는 데까지 442초(7분 22초) 소요됐다.

한 지역의 데이터를 파싱하는 데 3~10초 가량 소요됐고, 총 115곳을 가져와야하므로 약 7-8분 소요되는 것이었다. 

워낙 가져와야할 데이터가 많아 애초에 주소창에 xml이 존재하는 url을 요청했을 때 3~5초 정도 지연됨이 발생하긴 했다. 

꽤 오랜 시간이 걸려 성능을 개선시키기 위해 여러 방식들을 시도해봤다...

 

CompletableFuture 란?

Java 8에서 추가된 클래스로 비동기 연산을 처리하고 완료된 결과나 예외를 처리한다. 

기존에 있던 Java5의 Future 인터페이스의 한계점을 개선시킨 것이다. 

🔹 Future란 ?

👉 비동기 작업의 결과를 나타내는 데 사용되며 완료 상태를 확인하거나 작업 결과를 얻는 기능만 제공

🔹 Future의 한계점 

• 외부에서 완료시킬 수 없음 ➔ get()의 타임아웃 설정으로만 완료 가능 
• 블로킹 메소드 호출 : get() 메서드 호출 시 작업이 완료될 때까지 대기
• 작업 조합의 어려움
• 예외 처리 한계

Future 인터페이스를 확장하여 외부에서 완료시킬 수 있으며 작업 콜백, 예외 처리, 비동기적으로 여러 작업을 조합하는 기능 등을 제공하여 보다 효율적인 비동기 프로그래밍을 지원한다. 

---

🔹 비동기 작업 실행 메서드 

• runAsync : 반환값이 없는 경우

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 비동기 작업 수행
});

• supplyAsync : 반환값이 있는 경우 

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 비동기 작업 수행 후 결과 반환
    return "작업 결과";
});

 

🔹 작업 콜백 

• thenApply : 비동기 작업이 완료된 후에 해당 결과를 가공하거나 변환

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
        .thenApply(result -> result + " World");

 

• thenAccept : 비동기 작업이 완료된 후에 해당 결과를 소비

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
        .thenAccept(result -> System.out.println("Result: " + result));

• thenRun : 비동기 작업이 완료된 후에 추가적인 작업 수행

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "Hello")
        .thenRun(() -> System.out.println("Additional task after completion"));

 

 

🔹 작업 조합 

• thenCompose : 이전 작업의 결과에 따라 다음 작업 수행하는데, 다음 작업은 이전 작업의 결과를 입력으로 받음

public class Main {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
                .thenCompose(result -> CompletableFuture.supplyAsync(() -> result + " World"));

        future.thenAccept(System.out::println);
    }
}

• thenCombine : 두 개의 CompletableFuture가 모두 완료된 후에 두 작업의 결과를 조합하여 새로운 결과 생성

public class Main {
    public static void main(String[] args) {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

        CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (result1, result2) -> result1 + " " + result2);

        combinedFuture.thenAccept(System.out::println);
    }
}

• allOf : 여러 개의 CompletableFuture가 모두 완료될 때까지 기다렸다가 모든 작업이 완료되면 새로운 CompletableFuture 반환

public class Main {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

        CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(future1, future2);

        // 모든 CompletableFuture가 완료될 때까지 기다린 후에 실행
        allOfFuture.get();

        System.out.println("All futures completed");
    }
}

• anyOf : 여러 개의 CompletableFuture 중에서 하나라도 완료될 때까지 기다린 후에 해당 작업의 결과 반환

public class Main {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

        CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2);

        // 가장 먼저 완료된 CompletableFuture의 결과 출력
        System.out.println(anyOfFuture.get());
    }
}

 

 

 

CompletableFuture의 non-blocking과 blocking

CompletableFuture의 장점은 다양한 task를 blocking하지 않고 조합할 수 있다.

조합 외에도 supplyAsync(), thenApply(), thenCompose() 등을 통해서 non-blocking으로 수행하는 것을 확인할 수 있다. 

---

그렇다면 blocking은 언제 발생하는 것일까?

결론부터 말하면, 

• get()
• join()

등의 메서드를 수행할 때이다. 

 

get(), join() 메서드는 CompletableFuture가 완료될 때까지 호출한 스레드가 블로킹되어 작업이 완료되기를 기다린다.

 

 

 

 

CompletableFuture - 3분 소요

아래와 같은 코드를 돌려보니 3분 가량 소요됐다. 

@RequiredArgsConstructor
@Service
@Slf4j
public class ApiScheduler {

	private final AreaRepository areaRepository;
	private final CityDataRepository cityDataRepository;

	@Value("${seoul.open.api.url}")
	private StringBuilder url;

	@Transactional
	@Scheduled(cron = "* * * * * *")
	public void call() {
		List<Area> areas = areaRepository.findAll();
		List<CompletableFuture<CityData>> futures = areas.stream()
			.map(this::fetchCityDataAsync)
			.collect(Collectors.toList());
            
		List<CityData> cityDataList = futures.stream()
			.map(CompletableFuture::join)
			.collect(Collectors.toList());
            
		cityDataRepository.deleteAll();
		cityDataRepository.saveAll(cityDataList);
	}

	private CompletableFuture<CityData> fetchCityDataAsync(Area area) {
		return CompletableFuture.supplyAsync(() -> {
			try {
				log.info("areaId: {} areaName: {}", area.getAreaId(), area.getAreaName());
				String apiUrl = url + ("/" + urlEncoding(area.getAreaName()));
				Document document = DocumentBuilderFactory.newInstance().newDocumentBuilder().parse(apiUrl);
				return parseCityData(document);
			} catch (SAXException | IOException | ParserConfigurationException e) {
				log.error("error fetching citydata for areaname {}", area.getAreaName(), e);
				return null;
			}
		});
	}

	private CityData parseCityData(Document document) {
		return CityData cityData = CityData.builder()
                .areaName(getElement(documentInfo, "AREA_NM"))
                .areaCongestionLevel(getElement(documentInfo, "AREA_CONGEST_LVL"))
                .areaCongestionMessage(getElement(documentInfo, "AREA_CONGEST_MSG"))
                .pplUpdateTime(getElement(documentInfo, "PPLTN_TIME"))
                .forecastPopulation(getElement(documentInfo, "FCST_PPLTN"))
                .forecastCongestionLevel(getElement(documentInfo, "FCST_CONGEST_LVL"))
                .temperature(getElement(documentInfo, "TEMP"))
                .maxTemperature(getElement(documentInfo, "MAX_TEMP"))
                .minTemperature(getElement(documentInfo, "MIN_TEMP"))
                .pm25Index(getElement(documentInfo, "PM25_INDEX"))
                .pm10Index(getElement(documentInfo, "PM10_INDEX"))
                .pcpMsg(getElement(documentInfo, "PCP_MSG"))
                .weatherTime(getElement(documentInfo,"WEATHER_TIME"))
                .culturalEventName(getElement(documentInfo, "EVENT_NM"))
                .culturalEventPeriod(getElement(documentInfo, "EVENT_PERIOD"))
                .culturalEventPlace(getElement(documentInfo, "EVENT_PLACE"))
                .culturalEventUrl(getElement(documentInfo, "URL"))
                .build();
	}

	private String getElementText(Document document, String tag) {
		// return document.getElementsByTagName(tag).item(0).getTextContent();
		// Document 객체에서 태그 이름에 해당하는 요소 가져오기
		NodeList nodeList = document.getElementsByTagName(tag);

		// 가져온 요소가 비어 있는지 확인
		if (nodeList.getLength() > 0) {
			// 첫 번째로 발견된 요소의 텍스트 내용을 반환
			return nodeList.item(0).getTextContent();
		} else {
			// 해당 태그가 없을 경우
			return "NO TAG";
		}
	}
}

 

 

 

 

CompletableFuture & 엔티티 분류하기 - 60~80초 소요

날씨, 인구, 문화행사 엔티티를 각각 분류했고, 필요한 데이터들을 각 엔티티 필드로 생성했다. 

실행시켜보니 60~80초 정도 소요되었다 !

 

 

😇

솔직히 60~80초 가량 소요되는 것도 많이 걸리는 것이라 생각된다. 

7-8분 소요되는 것을 그래도 1분대로 줄였으니 뿌듯하다만....

코드가 훨씬 길어졌고, 중복되는 내용도 있어 코드 자체로는 썩 마음에 들지 않는다. 

WebClient로 API를 호출하여 처리하는 방법 등 다른 해결책을 추후에 적용시켜 좀더 알아봐야겠다..😇

 

 

 

 

 

참고자료

https://11st-tech.github.io/2024/01/04/completablefuture/#%EB%B3%91%EB%A0%AC%EC%B2%98%EB%A6%AC

https://mangkyu.tistory.com/263

https://www.baeldung.com/java-completablefuture

https://www.baeldung.com/java-completablefuture-non-blocking