Skip to content

Commit 2d924c8

Browse files
oOccasiooOccasio
authored
✨ Feature - 비동기 Job 스케줄링 AI 호출을 구현한다
✨ Feature - 비동기 Job 스케줄링 AI 호출을 구현한다
2 parents ffcfed6 + 0525314 commit 2d924c8

11 files changed

Lines changed: 286 additions & 6 deletions

File tree

‎src/main/java/sopt/comfit/global/constants/Constants.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ public class Constants {
1212
public static final AntPathMatcher PATH_MATCHER = new AntPathMatcher();
1313
public static final String BASE_URL =
1414
"https://bucket-com-fit-server.s3.ap-northeast-2.amazonaws.com/";
15+
public static final String JOB_QUEUE_KEY = "report:job:queue";
1516

1617
public static List<String> NO_NEED_AUTH = List.of(
1718
"/swagger",

‎src/main/java/sopt/comfit/report/controller/AIReportController.java‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,8 @@ public GetReportCompanyResponseDto getReportCompany(@PathVariable Long companyId
4949
return aiReportFacade.getReportCompany(companyId);
5050
}
5151

52-
@Override
52+
@PostMapping("/match/sync")
53+
@SecurityRequirement(name = "JWT")
5354
public AIReportResponseDto matchExperience(@LoginUser Long userId,
5455
@Valid @RequestBody MatchExperienceRequestDto requestDto){
5556
return aiReportFacade.matchExperience(MatchExperienceCommandDto.of(userId, requestDto));
@@ -76,10 +77,17 @@ public Mono<AIReportResponseDto> matchExperienceWebfluxParallel(@LoginUser Long
7677
return aiReportFacade.matchExperienceParallel(MatchExperienceCommandDto.of(userId, requestDto));
7778
}
7879

79-
@PostMapping("/match/sync/parallel")
80-
@SecurityRequirement(name = "JWT")
80+
@Override
8181
public AIReportResponseDto matchExperienceVirtualThread(@LoginUser Long userId,
8282
@RequestBody MatchExperienceRequestDto requestDto){
8383
return aiReportFacade.matchExperienceVirtualThread(MatchExperienceCommandDto.of(userId, requestDto));
8484
}
85+
86+
@PostMapping("/match/async/jobs")
87+
@SecurityRequirement(name = "JWT")
88+
public Long matchExperienceJob(@LoginUser Long userId,
89+
@RequestBody MatchExperienceRequestDto requestDto){
90+
return aiReportFacade.matchExperienceJob(MatchExperienceCommandDto.of(userId, requestDto));
91+
}
92+
8593
}

‎src/main/java/sopt/comfit/report/controller/AIReportSwagger.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,8 @@ public interface AIReportSwagger {
4848
@PostMapping
4949
@ResponseStatus(HttpStatus.CREATED)
5050
@SecurityRequirement(name = "JWT")
51-
AIReportResponseDto matchExperience(@LoginUser Long userId,
52-
@Valid @RequestBody MatchExperienceRequestDto requestDto);
51+
AIReportResponseDto matchExperienceVirtualThread(@LoginUser Long userId,
52+
@Valid @RequestBody MatchExperienceRequestDto requestDto);
5353

5454
@Operation(
5555
summary = "AI_Report 리스트 조회/ 검색",

‎src/main/java/sopt/comfit/report/domain/AIReportJob.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package sopt.comfit.report.domain;
2+
3+
import jakarta.persistence.*;
4+
import lombok.AccessLevel;
5+
import lombok.Builder;
6+
import lombok.Getter;
7+
import lombok.NoArgsConstructor;
8+
import sopt.comfit.global.base.BaseTimeEntity;
9+
10+
@Entity
11+
@NoArgsConstructor(access = AccessLevel.PROTECTED)
12+
@Getter
13+
@Table(name = "ai_report_job")
14+
public class AIReportJob extends BaseTimeEntity {
15+
@Id
16+
@GeneratedValue(strategy = GenerationType.IDENTITY)
17+
private Long id;
18+
19+
@Column(nullable = false)
20+
private Long userId;
21+
22+
@Column(nullable = false)
23+
private Long experienceId;
24+
25+
@Column(nullable = false)
26+
private Long companyId;
27+
28+
@Column(nullable = false, columnDefinition = "TEXT")
29+
private String description;
30+
31+
@Column(nullable = false)
32+
@Enumerated(EnumType.STRING)
33+
private EJobStatus status = EJobStatus.PENDING;
34+
35+
@Builder(access = AccessLevel.PRIVATE)
36+
private AIReportJob(
37+
final Long userId,
38+
final Long experienceId,
39+
final Long companyId,
40+
final String description,
41+
final EJobStatus status
42+
){
43+
this.userId = userId;
44+
this.experienceId = experienceId;
45+
this.companyId = companyId;
46+
this.description = description;
47+
this.status = status;
48+
}
49+
50+
public static AIReportJob create(final Long userId,
51+
final Long experienceId,
52+
final Long companyId,
53+
final String description) {
54+
return new AIReportJob(
55+
userId,
56+
experienceId,
57+
companyId,
58+
description,
59+
EJobStatus.PENDING);
60+
}
61+
62+
public void startProcessing() {
63+
this.status = EJobStatus.PROCESSING;
64+
}
65+
66+
public void complete() {
67+
this.status = EJobStatus.COMPLETED;
68+
}
69+
70+
public void fail() {
71+
this.status = EJobStatus.FAILED;
72+
}
73+
}

‎src/main/java/sopt/comfit/report/domain/AIReportJobRepository.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package sopt.comfit.report.domain;
2+
3+
import org.springframework.data.jpa.repository.JpaRepository;
4+
5+
public interface AIReportJobRepository extends JpaRepository<AIReportJob,Long> {
6+
}

‎src/main/java/sopt/comfit/report/domain/EJobStatus.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package sopt.comfit.report.domain;
2+
3+
import lombok.AllArgsConstructor;
4+
import lombok.Getter;
5+
6+
@AllArgsConstructor
7+
@Getter
8+
public enum EJobStatus {
9+
PENDING("수정"),
10+
PROCESSING("작업중"),
11+
COMPLETED("완료"),
12+
FAILED("실패");
13+
14+
private final String description;
15+
}

‎src/main/java/sopt/comfit/report/dto/command/MatchExperienceCommandDto.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,4 +15,8 @@ public record MatchExperienceCommandDto(
1515
public static MatchExperienceCommandDto of(Long userId, MatchExperienceRequestDto request) {
1616
return new MatchExperienceCommandDto(userId, request.companyId(), request.experienceId(), request.jobDescription());
1717
}
18+
19+
public static MatchExperienceCommandDto of(Long userId, Long companyId, Long experienceId, String jobDescription) {
20+
return new MatchExperienceCommandDto(userId, companyId, experienceId, jobDescription);
21+
}
1822
}

‎src/main/java/sopt/comfit/report/exception/AIReportErrorCode.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@ public enum AIReportErrorCode implements ErrorCode {
1818
AI_RATE_LIMITED(HttpStatus.TOO_MANY_REQUESTS, "AI_429_001", "AI 요청 한도 초과"),
1919
AI_SERVER_ERROR(HttpStatus.BAD_GATEWAY, "AI_502_002", "AI 서버 오류"),
2020
AI_CALL_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "AI_500_003", "AI 병렬 호출 실패"),
21-
AI_RESPONSE_REQUIRED_FIELD_OMIT(HttpStatus.INTERNAL_SERVER_ERROR, "AI_500_004", "AI 응답이 필수 응답값을 누락했습니다.")
21+
AI_RESPONSE_REQUIRED_FIELD_OMIT(HttpStatus.INTERNAL_SERVER_ERROR, "AI_500_004", "AI 응답이 필수 응답값을 누락했습니다."),
22+
REPORT_JOB_NOT_FOUND(HttpStatus.NOT_FOUND, "AI_404_002", "해당하는 Job을 찾을 수 없습니다."),
23+
JOB_QUEUE_FULL(HttpStatus.TOO_MANY_REQUESTS,"AI_429_001", "JOB QUEUE가 꽉찼습니다")
2224
;
2325
private final HttpStatus status;
2426
private final String prefix;

‎src/main/java/sopt/comfit/report/job/AIReportJobService.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package sopt.comfit.report.job;
2+
3+
import lombok.RequiredArgsConstructor;
4+
import org.springframework.data.redis.core.StringRedisTemplate;
5+
import org.springframework.stereotype.Service;
6+
import org.springframework.transaction.annotation.Transactional;
7+
import sopt.comfit.global.constants.Constants;
8+
import sopt.comfit.global.exception.BaseException;
9+
import sopt.comfit.report.domain.AIReportJob;
10+
import sopt.comfit.report.domain.AIReportJobRepository;
11+
import sopt.comfit.report.exception.AIReportErrorCode;
12+
13+
@Service
14+
@RequiredArgsConstructor
15+
public class AIReportJobService {
16+
17+
private final AIReportJobRepository reportJobRepository;
18+
private final StringRedisTemplate redisTemplate;
19+
20+
@Transactional
21+
public Long createJob(Long userId, Long experienceId, Long companyId, String jobDescription) {
22+
23+
Long queueSize = redisTemplate.opsForList().size(Constants.JOB_QUEUE_KEY);
24+
25+
if (queueSize != null && queueSize > 200) {
26+
throw BaseException.type(AIReportErrorCode.JOB_QUEUE_FULL);
27+
}
28+
AIReportJob job = AIReportJob.create(userId, experienceId, companyId, jobDescription);
29+
reportJobRepository.save(job);
30+
31+
redisTemplate.opsForList().leftPush(Constants.JOB_QUEUE_KEY, String.valueOf(job.getId()));
32+
33+
return job.getId();
34+
}
35+
36+
@Transactional
37+
public void startProcessing(Long jobId) {
38+
AIReportJob job = findJob(jobId);
39+
job.startProcessing();
40+
}
41+
42+
@Transactional
43+
public void complete(Long jobId) {
44+
AIReportJob job = findJob(jobId);
45+
job.complete();
46+
}
47+
48+
@Transactional
49+
public void fail(Long jobId) {
50+
AIReportJob job = findJob(jobId);
51+
job.fail();
52+
}
53+
54+
@Transactional(readOnly = true)
55+
public AIReportJob findJob(Long jobId) {
56+
return reportJobRepository.findById(jobId)
57+
.orElseThrow(() -> BaseException.type(AIReportErrorCode.REPORT_JOB_NOT_FOUND));
58+
}
59+
}

‎src/main/java/sopt/comfit/report/job/AIReportJobWorker.java‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
package sopt.comfit.report.job;
2+
3+
import jakarta.annotation.PostConstruct;
4+
import jakarta.annotation.PreDestroy;
5+
import lombok.RequiredArgsConstructor;
6+
import lombok.extern.slf4j.Slf4j;
7+
import org.springframework.data.redis.core.StringRedisTemplate;
8+
import org.springframework.stereotype.Component;
9+
import sopt.comfit.global.constants.Constants;
10+
import sopt.comfit.report.domain.AIReportJob;
11+
import sopt.comfit.report.dto.command.MatchExperienceCommandDto;
12+
import sopt.comfit.report.infra.dto.PreparedDataDto;
13+
import sopt.comfit.report.infra.prompt.AIReportParallelPromptBuilder;
14+
import sopt.comfit.report.infra.service.RetryableAiCallerService;
15+
import sopt.comfit.report.service.AIReportCommandService;
16+
import sopt.comfit.report.service.AIReportQueryService;
17+
18+
import java.time.Duration;
19+
import java.util.ArrayList;
20+
import java.util.List;
21+
22+
@Slf4j
23+
@Component
24+
@RequiredArgsConstructor
25+
public class AIReportJobWorker {
26+
27+
private final StringRedisTemplate redisTemplate;
28+
private final AIReportJobService reportJobService;
29+
private final AIReportQueryService aiReportQueryService;
30+
private final AIReportCommandService aiReportCommandService;
31+
private final RetryableAiCallerService aiCaller;
32+
33+
List<Thread> workers = new ArrayList<>();
34+
35+
@PostConstruct
36+
public void startWorker() {
37+
int workerCount = 3;
38+
workers = new ArrayList<>();
39+
for (int i = 0; i < workerCount; i++) {
40+
Thread t = Thread.ofVirtual()
41+
.name("report-job-worker-" + i)
42+
.start(this::listen);
43+
workers.add(t);
44+
}
45+
log.info("ReportJobWorker {}개 시작", workerCount);
46+
}
47+
48+
@PreDestroy
49+
public void stopWorker() {
50+
workers.forEach(Thread::interrupt);
51+
log.info("ReportJobWorker 종료");
52+
}
53+
54+
private void listen() {
55+
while (!Thread.currentThread().isInterrupted()) {
56+
try {
57+
String jobId = redisTemplate.opsForList()
58+
.rightPop(Constants.JOB_QUEUE_KEY, Duration.ofSeconds(30));
59+
60+
if (jobId == null) continue;
61+
62+
processJob(Long.parseLong(jobId));
63+
} catch (Exception e) {
64+
log.error("Worker 루프 에러", e);
65+
}
66+
}
67+
}
68+
69+
private void processJob(Long jobId) {
70+
reportJobService.startProcessing(jobId);
71+
72+
try {
73+
MatchExperienceCommandDto command = buildCommand(jobId);
74+
PreparedDataDto data = aiReportQueryService.prepareData(command);
75+
76+
String perspectivesJson = aiCaller.callSyncWithField(
77+
AIReportParallelPromptBuilder.buildPerspective(data),
78+
"Perspectives", "perspectives");
79+
80+
String mergedJson = aiCaller.callParallelWithVirtualThread(data, perspectivesJson);
81+
82+
aiReportCommandService.parseAndSave(mergedJson, data.experience(),
83+
data.company(), command.jobDescription());
84+
85+
reportJobService.complete(jobId);
86+
log.info("Job 처리 완료 - jobId: {}", jobId);
87+
} catch (Exception e) {
88+
log.error("Job 처리 실패 - jobId: {}", jobId, e);
89+
reportJobService.fail(jobId);
90+
}
91+
}
92+
93+
private MatchExperienceCommandDto buildCommand(Long jobId) {
94+
AIReportJob job = reportJobService.findJob(jobId);
95+
return new MatchExperienceCommandDto(
96+
job.getUserId(),
97+
job.getCompanyId(),
98+
job.getExperienceId(),
99+
job.getDescription());
100+
}
101+
}

0 commit comments

Comments
 (0)