Quartz, Spring Batch, AWS SQS 로직 순서

러블리소피·2024년 12월 27일

Quartz, Spring Batch, AWS SQS

목록 보기

4/11

1. Quartz: 작업 스케줄링

Quartz는 특정 시간(또는 주기)에 작업을 실행할 Trigger와 Job을 설정합니다.
작업이 실행되면 Quartz는 미리 정의된 Job 클래스를 호출합니다.
이 Job 클래스에서 Spring Batch Job을 실행하거나, AWS SQS와의 통신을 시작할 수 있습니다.

2. Spring Batch: 데이터 처리

Quartz에서 호출된 작업(Job)은 Spring Batch의 JobLauncher를 사용해 배치 작업을 실행합니다.
Spring Batch는 Reader, Processor, Writer 단계를 통해 데이터를 처리합니다.

Reader: 데이터를 읽어들임 (예: 데이터베이스, 파일, SQS 등).
Processor: 데이터를 가공하거나 변환.
Writer: 처리된 데이터를 저장하거나 전송 (예: SQS 메시지 전송).

데이터 처리 중 발생한 로그와 상태는 Quartz 로그나 Spring Batch의 Job Execution Repository에 기록됩니다.

3. AWS SQS: 메시지 전달 및 처리

Spring Batch의 Writer 또는 Quartz 작업 중 AWS SQS에 메시지를 전송합니다.
메시지는 특정 큐로 전달되고 대기 상태가 됩니다.
다른 서비스(또는 애플리케이션)는 SQS 큐에서 메시지를 읽고 처리합니다.
AWS SDK 또는 메시지 소비자(Consumer)를 사용해 메시지를 폴링(Polling)하거나 구독(Subscription) 방식으로 처리.
메시지 처리 결과는 필요에 따라 다시 데이터베이스나 로그 시스템에 저장되거나, 다른 큐로 전달됩니다.

로직 흐름의 예시

1. Quartz 기반 스케줄링

@DisallowConcurrentExecution
public class ScheduledJob implements Job {
    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job springBatchJob;

    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        try {
            JobParameters jobParameters = new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())
                    .toJobParameters();
            jobLauncher.run(springBatchJob, jobParameters);
        } catch (Exception e) {
            throw new JobExecutionException("Error executing job", e);
        }
    }
}

2. Spring Batch Job 설정

@Configuration
public class BatchConfig {
    @Bean
    public Job exampleJob(JobBuilderFactory jobBuilderFactory, Step step1) {
        return jobBuilderFactory.get("exampleJob")
                .start(step1)
                .build();
    }

    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<String> reader,
                      ItemProcessor<String, String> processor, ItemWriter<String> writer) {
        return stepBuilderFactory.get("step1")
                .<String, String>chunk(10)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public ItemWriter<String> writer() {
        return items -> {
            for (String item : items) {
                // AWS SQS로 메시지 전송
                sendToSQS(item);
            }
        };
    }

    private void sendToSQS(String message) {
        // AWS SQS 클라이언트를 사용해 메시지 전송
        AmazonSQS sqsClient = AmazonSQSClientBuilder.defaultClient();
        sqsClient.sendMessage("https://sqs.<region>.amazonaws.com/<account-id>/<queue-name>", message);
    }
}

3. AWS SQS 메시지 소비자

public class SQSConsumer {
    public void pollMessages() {
        AmazonSQS sqsClient = AmazonSQSClientBuilder.defaultClient();
        ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest()
                .withQueueUrl("https://sqs.<region>.amazonaws.com/<account-id>/<queue-name>")
                .withMaxNumberOfMessages(10);
        
        List<Message> messages = sqsClient.receiveMessage(receiveMessageRequest).getMessages();
        for (Message message : messages) {
            // 메시지 처리
            processMessage(message.getBody());

            // 메시지 삭제
            sqsClient.deleteMessage(new DeleteMessageRequest()
                    .withQueueUrl("https://sqs.<region>.amazonaws.com/<account-id>/<queue-name>")
                    .withReceiptHandle(message.getReceiptHandle()));
        }
    }

    private void processMessage(String message) {
        System.out.println("Processing message: " + message);
        // 메시지 처리 로직 구현
    }
}