Spring Batch 기반 블로그 조회수 누적 및 집계 처리하기

Processing Blog Post Hit Count with Spring Batch

이 블로그 기획 초기부터 조회수 기능은 중요한 요소로 고려되었습니다. 구현 과정에서 단순히 페이지 로드를 조회로 정의할지, 더 다양한 조건을 포함할지 고민이 있었지만, 과거 조회수를 조작하던 시절의 감성을 살려 새로고침이나 뒤로 가기 등 모든 접근을 조회수로 처리하기로 결정했습니다. 이는 단순 조회수 집계를 넘어, 빈번한 이벤트가 시스템에 미치는 영향을 실험하려는 목적도 있었습니다.

유저 유입이 적은 상황에서도 이 정책은 빈번한 이벤트로 서버와 데이터베이스에 부담을 줄 가능성이 있습니다. 이를 해결하기 위해  Redis를 도입해 조회 데이터를 실시간으로 저장하는 방식을 채택했고, 데이터를 효과적으로 활용하기 위해  Spring Batch를 사용해 Redis에 저장된 데이터를 배치로 처리하는 전략을 추가했습니다.

이번 글에서는 이러한 조회수 누적 및 집계 처리 작업을 구현하는 과정을 정리해 보았습니다.

Prerequisites

이 글은  Spring Boot 3.3.2와  Java 21을 기반으로 작성되었으며,  MySQL과  Redis 환경이 구축되어 있어야 합니다. 다음과 같이 build.gradle 파일에 필요한 의존성들을 추가하여 Spring Boot 프로젝트를 구성합니다:

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    implementation 'org.springframework.boot:spring-boot-starter-security'
    implementation 'org.springframework.boot:spring-boot-starter-validation'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    developmentOnly 'org.springframework.boot:spring-boot-devtools'
    runtimeOnly 'com.mysql:mysql-connector-j'
    annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

• Java 21
• Spring Boot 3.3.2
• Gradle 8
• MySQL 8
• Dependencies
  • Spring Web
  • Spring Data JPA
  • MySQL Driver
  • Validation
  • Lombok

Storing Real-Time Data in Redis

현재 블로그의 프론트엔드는  Next.js로 구성되어 있습니다. 사용자가 특정 게시글에 진입하면 Next.js 클라이언트 컴포넌트에서 해당 이벤트를 감지하고, 관련 데이터를 서버로 전송하게 됩니다. 이번 단계에서는 이렇게 서버가 수신한 데이터를 Redis에 저장하여 실시간 데이터 처리를 빠르게 처리하는 구조를 구현합니다.

Integrating Spring Data Redis

실시간으로 조회 데이터를 Redis에 저장하기 위해 Spring Data Redis를 사용하였습니다. Spring Data Redis는 Redis와의 상호작용을 간편하게 처리할 수 있는 인터페이스를 제공하고 있습니다. 먼저, 프로젝트에 필요한 Redis 의존성을 build.gradle 파일에 추가합니다:

dependencies {
+   implementation 'org.springframework.boot:spring-boot-starter-data-redis'

    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    implementation 'org.springframework.boot:spring-boot-starter-security'
    implementation 'org.springframework.boot:spring-boot-starter-validation'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    developmentOnly 'org.springframework.boot:spring-boot-devtools'
    runtimeOnly 'com.mysql:mysql-connector-j'
    annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

의존성을 추가한 후, application.yml 파일에 Redis 연결 정보를 설정합니다. 이 설정을 기반으로 Spring Boot는 자동으로 RedisTemplate을 생성하며, 이를 사용해 데이터를 손쉽게 저장할 수 있게 됩니다.

spring:
  data:
    redis:
      host: localhost
      port: 6379
      password: ${REDIS_PASSWORD}
      repositories:
        enabled: false

Spring Boot는 기본적으로 RedisTemplate을 자동으로 제공하지만, 필요하다면 커스터마이징할 수도 있습니다. 여기서는 기본 설정을 사용해 Redis와 상호작용하는 로직을 구현하겠습니다.

Recording Post Hit Events

사용자가 특정 블로그 포스트에 진입할 때마다 클라이언트는 postId, pathname, ip와 같은 데이터를 서버로 전송하게 됩니다:

{
  "postId": "85173281782414024",
  "pathname": "all-new-design",
  "ip": "127.0.0.1"
}

서버에서는 이 데이터를 수신한 후 기본적인 유효성 검사를 진행합니다:

@Slf4j
public record StoreHitEventRequest(
        @NotNull String id,
        @NotBlank String pathname,
        @NotBlank String ip) {

    public StoreHitEventRequest {
        validate();
    }
    ...
}

유효성 검사를 통과한 데이터는 Redis에 저장하기 전에 적절한 형태로 가공하며, 이때 TTL 값도 함께 설정합니다. 이 TTL 값은 일정 시간이 지나면 Redis에 저장된 데이터가 자동으로 삭제되도록 하는 역할을 합니다. Redis에 데이터를 저장할 때 TTL(Time To Live) 값을 설정하는 것은 필수적입니다. 배치 서버에 예기치 못한 문제가 발생해 데이터가 계속 쌓이게 되면, Redis를 사용하는 다른 주요 서비스에 영향을 줄 수 있기 때문입니다. 이를 방지하기 위해 TTL 값을 설정하여 데이터가 자동으로 삭제되도록 하고, 시스템 자원을 효율적으로 관리합니다. 여기서는 TTL 값을 1일로 설정해, 하루가 지나면 데이터가 자동으로 정리되도록 했습니다.

@Slf4j
public record HitEventRedis(
        Long id,
        Long postId,
        String ipAddress,
        LocalDateTime hitAt,
        Long ttl) {

    public static HitEventRedis from(HitEvent domain) {
        return new HitEventRedis(
                domain.getHitEventId().id(),
                domain.getPostId(),
                domain.getIpAddress(),
                domain.getHitAt(),
                86400L);
    }
}

마지막으로, RedisTemplate을 사용해 Redis에 데이터를 저장하는 로직을 구현합니다. Redis는 여러 직렬화 및 역직렬화 방식을 지원하지만, 여기서는 ObjectMapper를 사용해 객체를 JSON 문자열로 변환해 저장했습니다. JSON 문자열을 사용하는 이유는, 여러 서버 간에 데이터를 쉽게 공유하고 역직렬화할 수 있기 때문입니다. 또한, JSON 형식은 시스템 간의 호환성이 높아 관리와 디버깅에도 유리합니다.

@Slf4j
@RequiredArgsConstructor
@PersistenceAdapter
public class HitPersistenceAdapter implements StoreHitEventPort {

    private static final String HIT_EVENT_KEY_PREFIX = "hits::";

    private final RedisTemplate<String, String> redisTemplate;
    private final ObjectMapper objectMapper;

    @Override
    @Transactional
    public HitEvent persist(HitEvent hitEvent) {
        try {
            var hitEventRedis = HitEventRedis.from(hitEvent);
            var key = HIT_EVENT_KEY_PREFIX + hitEventRedis.id();
            var value = objectMapper.writeValueAsString(hitEventRedis);
            redisTemplate.opsForValue().set(key, value, hitEventRedis.ttl(), SECONDS);

            log.info("persist: Successfully persisted HitEvent - hitEventId={}, postId={}",
                    hitEvent.getHitEventId().id(),
                    hitEvent.getPostId());
        } catch (JsonProcessingException e) {
            log.error("persist: Failed to serialize HitEvent for Redis - errorMessage={}", e.getMessage());
        } catch (Exception redisException) {
            log.error("persist: Failed to persist HitEvent in Redis - errorMessage={}", redisException.getMessage());
        }
        return hitEvent;
    }
}

• RedisTemplate<K, V>:
  • Redis와의 상호작용을 간편하게 해주는 Spring의 템플릿 클래스입니다.
  • 여기서는 키와 값을 모두 String 타입으로 사용하는 RedisTemplate을 사용해, 데이터를 직렬화된 문자열 형태로 저장하고 조회합니다.
  • RedisTemplate을 통해 Redis에 데이터를 저장, 조회, 삭제 등의 작업을 편리하게 수행할 수 있습니다.
  • Spring Data Redis에서 기본적으로 제공하는 CRUD Repository도 있지만, RedisTemplate을 활용하면 더 세밀하고 맞춤형으로 Redis와 상호작용할 수 있습니다.
• writeValueAsString():
  • ObjectMapper 클래스에서 제공하는 기능으로, Java 객체를 JSON 문자열로 직렬화합니다.
  • 여기서는 HitEventRedis 객체를 JSON 형식의 문자열로 변환하여 Redis에 저장할 수 있도록 준비하는 역할을 합니다.
  • 직렬화된 JSON 데이터는 Redis에 쉽게 저장되고, 다른 시스템과 호환성이 높아집니다.
• opsForValue():
  • RedisTemplate 클래스에서 값을 단순한 키-값 형태로 다룰 수 있도록 하는 API를 제공합니다.
  • 기본적인 키-값 작업을 처리하는 데 사용되며, 여기서는 set() 메서드를 호출하여 특정 키에 대해 직렬화된 JSON 데이터를 Redis에 저장합니다.

Testing Data Storage Performance with K6

조회 이벤트 저장 API를 구현한 후, Redis와 MySQL에 데이터를 저장할 때의 성능 차이를 비교해 보겠습니다. 이를 위해 K6를 사용해 동일한 데이터를 두 데이터베이스에 저장하는 부하 테스트를 진행했습니다. 100명의 가상 사용자가 1분 동안 지속적으로 API 요청을 보내며 성능을 측정하였고, 이제 그 결과를 살펴보겠습니다.

먼저 MySQL에 데이터를 저장했을 때의 테스트 결과는 다음과 같습니다:

          /\      |‾‾| /‾‾/   /‾‾/
     /\  /  \     |  |/  /   /  /
    /  \/    \    |     (   /   ‾‾\
   /          \   |  |\  \ |  (‾)  |
  / __________ \  |__| \__\ \_____/ .io

     execution: local
        script: store-hit-events-load-test.js
        output: -

     scenarios: (100.00%) 1 scenario, 100 max VUs, 1m30s max duration (incl. graceful stop):
              * default: 100 looping VUs for 1m0s (gracefulStop: 30s)


     ✓ status is 200

     checks.........................: 100.00% ✓ 209133      ✗ 0
     data_received..................: 94 MB   1.6 MB/s
     data_sent......................: 38 MB   631 kB/s
     http_req_blocked...............: avg=6.58µs   min=0s     med=1µs     max=18.76ms  p(90)=2µs     p(95)=3µs
     http_req_connecting............: avg=4.27µs   min=0s     med=0s      max=13.69ms  p(90)=0s      p(95)=0s
     http_req_duration..............: avg=28.6ms   min=1.68ms med=25.41ms max=837.82ms p(90)=40.85ms p(95)=49.67ms
       { expected_response:true }...: avg=28.6ms   min=1.68ms med=25.41ms max=837.82ms p(90)=40.85ms p(95)=49.67ms
     http_req_failed................: 0.00%   ✓ 0           ✗ 209133
     http_req_receiving.............: avg=100.97µs min=6µs    med=77µs    max=43.51ms  p(90)=168µs   p(95)=234µs
     http_req_sending...............: avg=11.19µs  min=2µs    med=7µs     max=15.4ms   p(90)=13µs    p(95)=19µs
     http_req_waiting...............: avg=28.49ms  min=1.48ms med=25.31ms max=837.79ms p(90)=40.72ms p(95)=49.57ms
     http_reqs......................: 209133  3484.018774/s
     iteration_duration.............: avg=28.68ms  min=1.72ms med=25.49ms max=837.88ms p(90)=40.95ms p(95)=49.75ms
     iterations.....................: 209133  3484.018774/s
     successful_requests............: 209133  3484.018774/s
     vus............................: 100     min=100       max=100
     vus_max........................: 100     min=100       max=100


running (1m00.0s), 000/100 VUs, 209133 complete and 0 interrupted iterations
default ✓ [======================================] 100 VUs  1m0s

이와 동일한 테스트를 Redis에 적용했을 때의 결과는 아래와 같습니다:

          /\      |‾‾| /‾‾/   /‾‾/
     /\  /  \     |  |/  /   /  /
    /  \/    \    |     (   /   ‾‾\
   /          \   |  |\  \ |  (‾)  |
  / __________ \  |__| \__\ \_____/ .io

     execution: local
        script: store-hit-events-load-test.js
        output: -

     scenarios: (100.00%) 1 scenario, 100 max VUs, 1m30s max duration (incl. graceful stop):
              * default: 100 looping VUs for 1m0s (gracefulStop: 30s)


     ✓ status is 200

     checks.........................: 100.00% ✓ 397961      ✗ 0
     data_received..................: 180 MB  3.0 MB/s
     data_sent......................: 72 MB   1.2 MB/s
     http_req_blocked...............: avg=4.32µs  min=0s       med=1µs     max=7.61ms   p(90)=1µs     p(95)=2µs
     http_req_connecting............: avg=2.81µs  min=0s       med=0s      max=4.89ms   p(90)=0s      p(95)=0s
     http_req_duration..............: avg=15.02ms min=645µs    med=12.54ms max=539.19ms p(90)=28.81ms p(95)=36.9ms
       { expected_response:true }...: avg=15.02ms min=645µs    med=12.54ms max=539.19ms p(90)=28.81ms p(95)=36.9ms
     http_req_failed................: 0.00%   ✓ 0           ✗ 397961
     http_req_receiving.............: avg=1.4ms   min=6µs      med=84µs    max=426.49ms p(90)=6.13ms  p(95)=9.66ms
     http_req_sending...............: avg=5.06µs  min=1µs      med=4µs     max=2.49ms   p(90)=8µs     p(95)=12µs
     http_req_tls_handshaking.......: avg=0s      min=0s       med=0s      max=0s       p(90)=0s      p(95)=0s
     http_req_waiting...............: avg=13.61ms min=564µs    med=11.44ms max=539.12ms p(90)=26.45ms p(95)=34.11ms
     http_reqs......................: 397961  6631.785942/s
     iteration_duration.............: avg=15.07ms min=673.33µs med=12.58ms max=539.22ms p(90)=28.85ms p(95)=36.95ms
     iterations.....................: 397961  6631.785942/s
     successful_requests............: 397961  6631.785942/s
     vus............................: 100     min=100       max=100
     vus_max........................: 100     min=100       max=100


running (1m00.0s), 000/100 VUs, 397961 complete and 0 interrupted iterations
default ✓ [======================================] 100 VUs  1m0s

MySQL의 경우 평균 응답 시간은 28.6ms로 측정되었으며, Redis는 15.02ms로 MySQL에 비해 두 배 가까이 빠른 응답 속도를 보였습니다. Redis는 실시간 데이터 처리에 최적화되어 있어, 초당 처리 요청 수 역시 MySQL보다 두 배 이상 높은 성능을 기록했습니다:

환경	총 요청 수	평균 응답 시간	초당 처리 요청 수	데이터 수신량	데이터 전송량
MySQL	209,133	28.6ms	3,484 req/s	94MB	38MB
Redis	397,961	15.02ms	6,631 req/s	180MB	72MB

결과적으로 Redis를 사용하면 블로그 게시글 조회수를 실시간으로 기록할 수 있으며, 구현 과정도 간편하게 처리할 수 있었습니다. 이러한 장점 덕분에 Redis는 실시간 트래픽이 많거나 대규모 데이터를 빠르게 처리해야 하는 상황에서 특히 효과적입니다. 이제 Redis에 저장된 데이터를 Spring Batch를 활용해 집계하는 방법을 살펴보겠습니다.

Batch Processing with Spring Batch

Spring Batch는 대용량 데이터 처리에 적합한 배치 작업을 효율적으로 관리할 수 있는 프레임워크입니다. 다양한 인터페이스와 유연한 구성을 제공하여 데이터 처리를 쉽게 스케줄링하고 관리할 수 있도록 합니다. 이번 단계에서는 Redis에 저장된 조회 이벤트 데이터를 MySQL 데이터베이스에 저장하고, 이를 합산하여 조회 수로 반영하는 배치 작업을 구현해 보겠습니다.

Setting Up Spring Batch Environment

먼저, Spring Batch를 사용하기 위해 필요한 의존성을 추가해야 합니다. 배치 작업은 한 번 구현되면 큰 변경 없이 주기적으로 실행되는 경우가 많습니다. 따라서, 이러한 배치 작업은 별도의 프로젝트로 구현하는 것이 개발 및 운영 측면에서 유리할 수 있습니다.

dependencies {
+   implementation 'org.springframework.boot:spring-boot-starter-batch'
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'

    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    implementation 'org.springframework.boot:spring-boot-starter-security'
    implementation 'org.springframework.boot:spring-boot-starter-validation'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    developmentOnly 'org.springframework.boot:spring-boot-devtools'
    runtimeOnly 'com.mysql:mysql-connector-j'
    annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

그리고 @SpringBootApplication 클래스에 Spring Batch 관련 애노테이션을 추가해 배치 작업과 스케줄링을 활성화합니다:

@SpringBootApplication
@ConfigurationPropertiesScan
@EnableScheduling
@EnableBatchProcessing
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

• @EnableBatchProcessing: Spring Batch 기능을 활성화하는 애노테이션입니다. 이 애노테이션을 통해 배치 작업을 정의하고 실행할 수 있는 다양한 설정을 자동으로 구성합니다.
• @EnableScheduling: Spring의 스케줄링 기능을 활성화하는 애노테이션입니다. 이를 통해 스케줄링된 작업이 일정한 간격으로 자동 실행되도록 설정할 수 있습니다.

Spring Batch를 추가한 후, 배치 작업의 환경 설정과 구성을 단계 별로 구현해나가겠습니다.

Fetching Redis Data and Storing in Database

Spring Batch에서 ItemReader는 데이터를 읽어오는 역할을 담당하며 다양한 구현체를 제공하지만, Redis에서 데이터를 읽어오는 구현체는 기본적으로 제공되지 않습니다. 따라서 아래와 같이 Redis 데이터를 조회하여 처리하는 ItemReader를 직접 구현해야 합니다:

@Slf4j
@RequiredArgsConstructor
@Component
public class HitEventRedisItemReader implements ItemReader<HitEventItem> {

    private static final String HIT_EVENT_KEY_PATTERN = "hits::";

    private final RedisTemplate<String, String> redisTemplate;
    private final ObjectMapper objectMapper;
    private Iterator<String> keyIterator;

    @Override
    public HitEventItem read() throws Exception {
        if (keyIterator == null || !keyIterator.hasNext()) {
            Set<String> keys = redisTemplate.keys(HIT_EVENT_KEY_PATTERN);
            if (keys == null || keys.isEmpty()) return null;

            keyIterator = keys.iterator();
        }

        if (keyIterator.hasNext()) {
            String key = keyIterator.next();
            String value = redisTemplate.opsForValue().get(key);
            if (value != null) {
                var hitEventItem = objectMapper.readValue(value, HitEventItem.class);

                redisTemplate.delete(key);
                return hitEventItem;
            }
        }

        return null;
    }
}

• ItemReader<T>: Spring Batch의 인터페이스로, 데이터를 읽어오는 역할을 합니다. 여기서는 인터페이스를 직접 구현하여 Redis에 저장된 데이터를 하나씩 가져와서 처리합니다.
• opsForValue().get(key): RedisTemplate을 통해 Redis에서 값을 가져옵니다. 여기서는 Redis에 저장된 JSON 데이터를 키를 사용해 조회하고, 이를 Java 객체로 역직렬화합니다.

위의 ItemReader는 Redis에 저장된 데이터를 Java 객체로 변환해 다음 단계로 전달하는 역할을 합니다. 전달된 데이터는 ItemProcessor에서 추가적인 가공을 거쳐, MySQL 테이블에 저장하기 적합한 형태로 변환됩니다. ItemProcessor는 데이터를 가공하고 필터링하는 중간 단계로서 중요한 역할을 수행합니다.

@Slf4j
@RequiredArgsConstructor
@Component
public class HitEventItemProcessor implements ItemProcessor<HitEventItem, HitEvent> {

    @Override
    public HitEvent process(@NonNull HitEventItem item) throws Exception {
        return HitEvent.from(item);
    }
}

• ItemProcessor<I, O>: 읽어온 데이터를 가공하는 인터페이스입니다. 여기서는 Redis에서 읽어온 HitEventItem 객체를 MySQL에 저장할 수 있도록 HitEvent 객체로 변환하는 작업을 수행합니다.

ItemProcessor에서 가공된 데이터는 ItemWriter를 통해 데이터베이스에 저장됩니다. ItemWriter는 배치 작업의 최종 단계로, 가공된 데이터를 외부 저장소로 전달하는 역할을 합니다. Spring Batch는 다양한 ItemWriter 구현체를 제공하며, 이 프로젝트에서는 대량의 데이터를 효율적으로 처리하기 위해 JdbcBatchItemWriter를 사용해 조회 이벤트를 MySQL에 일괄 저장하도록 구현하였습니다:

@Slf4j
@RequiredArgsConstructor
@Configuration
public class HitEventItemWriterConfig {

    private final DataSource dataSource;

    @Bean
    public JdbcBatchItemWriter<HitEvent> hitEventItemWriter() {
        return new JdbcBatchItemWriterBuilder<HitEvent>()
                .sql("""
                     insert into HIT_EVENTS (id, post_id, ip_address, hit_at, created_at, updated_at)
                     values (:id, :postId, :ipAddress, :hitAt, :createdAt, :updatedAt)
                     """)
                .itemSqlParameterSourceProvider(this::createSqlParameterSource)
                .dataSource(dataSource)
                .build();
    }

    private MapSqlParameterSource createSqlParameterSource(HitEvent hitEvent) {
        return new MapSqlParameterSource()
                .addValue("id", hitEvent.id())
                .addValue("postId", hitEvent.postId())
                .addValue("ipAddress", hitEvent.ipAddress())
                .addValue("hitAt", hitEvent.hitAt())
                .addValue("createdAt", hitEvent.createdAt())
                .addValue("updatedAt", hitEvent.updatedAt());
    }
}

• ItemWriter<T>:
  • Spring Batch에서 데이터를 최종적으로 처리하고 저장하는 역할을 담당하는 인터페이스입니다.
  • 배치 작업이 데이터를 ItemReader와 ItemProcessor를 통해 읽고 가공한 후, ItemWriter는 이 데이터를 목적지에 저장하는 데 사용됩니다. 목적지는 데이터베이스, 파일, 메시지 큐 등 다양할 수 있습니다.
  • ItemWriter는 데이터를 한 번에 여러 개씩 처리할 수 있으며, 배치 처리를 위해 데이터 일괄 저장을 지원합니다.
• JdbcBatchItemWriter<T>:
  • 데이터베이스에 대량의 데이터를 효율적으로 저장하는 데 사용되는 ItemWriter의 구현체입니다.
  • Spring의 JDBC 지원을 활용하여 배치 처리 중에 여러 데이터베이스 레코드를 한 번에 삽입하거나 갱신할 수 있습니다.
  • SQL을 사용하여 데이터를 데이터베이스로 배치로 전송하며, 성능을 최적화하기 위해 대량의 데이터를 한 번에 처리하는 것이 가능합니다.
• JdbcBatchItemWriterBuilder<T>: JdbcBatchItemWriter를 더 쉽게 구성하기 위한 빌더 클래스입니다. 이 빌더는 JdbcBatchItemWriter의 SQL 쿼리와 매개변수 제공자 등을 간편하게 설정할 수 있는 다양한 메서드를 제공합니다. 이 빌더를 사용하면 코드를 더 간결하고 명확하게 작성할 수 있으며, SQL 쿼리와 데이터를 손쉽게 설정할 수 있습니다.
  • sql(): 이 메서드는 JdbcBatchItemWriter가 실행할 SQL 쿼리를 설정합니다. 여기서는 데이터를 데이터베이스에 삽입하기 위해 INSERT 쿼리를 지정하고 있으며, 각 필드 값은 매개변수로 설정됩니다. 이 SQL 쿼리는 배치로 처리될 각 데이터 항목에 대해 실행됩니다.
  • itemSqlParameterSourceProvider(): 이 메서드는 SQL 쿼리에서 사용될 매개변수를 제공하는 SqlParameterSourceProvider를 설정합니다. JdbcBatchItemWriter가 데이터를 SQL 쿼리에 삽입할 때 각 필드의 값을 매개변수로 전달해야 하는데, itemSqlParameterSourceProvider()는 이를 제공하는 역할을 합니다. 이 코드에서는 각 HitEvent 객체의 필드 값을 MapSqlParameterSource를 통해 SQL 쿼리에 매핑하는 작업을 수행하고 있습니다.
  • MapSqlParameterSource: SQL 쿼리에 전달할 매개변수 값을 지정하는 데 사용되는 클래스입니다. 이 객체는 키-값 쌍으로 구성되며, SQL 쿼리의 각 매개변수에 대응되는 값을 제공할 수 있습니다. 여기서는 HitEvent 객체의 필드 값들을 매개변수로 설정하여 SQL 쿼리에 삽입하고 있습니다. 예를 들어, addValue() 메서드를 사용해 id, postId, ipAddress 등의 값을 쿼리에 전달합니다.
  • dataSource(): JdbcBatchItemWriter가 데이터를 저장할 데이터베이스 연결을 설정하는 메서드입니다. DataSource 객체는 데이터베이스와의 연결을 관리하며, 배치 작업 중에 데이터를 삽입하거나 갱신할 때 사용됩니다. 여기서는 스프링이 관리하는 DataSource 빈을 주입받아 사용합니다.

지금까지 구현한 ItemReader, ItemProcessor, ItemWriter를 하나의 Step으로 결합할 수 있습니다. Step은 Spring Batch에서 배치 작업을 구성하는 단위로, 특정 작업의 흐름을 정의하는 역할을 합니다. 이 Step은 Redis에서 데이터를 읽어오고, 필요한 처리를 거쳐 최종적으로 MySQL에 저장하는 전체 과정을 담당합니다. 배치 작업은 여러 Step으로 구성될 수 있으며, 각 Step은 독립적으로 실행됩니다.

@Slf4j
@Configuration
public class HitStatsBatchJobConfig {

    private final JobRepository jobRepository;
    private final PlatformTransactionManager transactionManager;
    private final HitEventRedisItemReader hitEventRedisItemReader;
    private final HitEventItemProcessor hitEventItemProcessor;
    private final JdbcBatchItemWriter<HitEvent> hitEventItemWriter;

    public HitStatsBatchJobConfig(
            JobRepository jobRepository,
            PlatformTransactionManager transactionManager,
            HitEventRedisItemReader hitEventRedisItemReader,
            HitEventItemProcessor hitEventItemProcessor,
            JdbcBatchItemWriter<HitEvent> hitEventItemWriter,
            JdbcCursorItemReader<HitEventCount> hitEventsCountItemReader,
            JdbcBatchItemWriter<HitEventCount> incrementHitStatsItemWriter
    ) {
        this.jobRepository = jobRepository;
        this.transactionManager = transactionManager;
        this.hitEventRedisItemReader = hitEventRedisItemReader;
        this.hitEventItemProcessor = hitEventItemProcessor;
        this.hitEventItemWriter = hitEventItemWriter;
        this.hitEventsCountItemReader = hitEventsCountItemReader;
        this.incrementHitStatsItemWriter = incrementHitStatsItemWriter;
    }

    @Bean
    public Step persistHitEventsStep() {
        return new StepBuilder("persist-hit-event-step", jobRepository)
                .<HitEventItem, HitEvent>chunk(100, transactionManager)
                .reader(hitEventRedisItemReader)
                .processor(hitEventItemProcessor)
                .writer(hitEventItemWriter)
                .build();
    }
}

• JobRepository:
  • Spring Batch의 핵심 컴포넌트 중 하나로, 배치 작업의 실행 상태와 메타데이터를 관리하는 역할을 합니다.
  • 배치 작업이 실행될 때 각 작업(Job)과 단계(Step)의 시작 및 완료 상태를 기록하며, 실패하거나 재시작해야 하는 경우 이를 추적합니다.
  • JobRepository는 배치 처리의 신뢰성과 안정성을 보장하는 중요한 역할을 합니다.
• PlatformTransactionManager:
  • Spring에서 트랜잭션 관리를 담당하는 인터페이스로, 데이터베이스 트랜잭션뿐만 아니라 JMS, JTA 등의 트랜잭션도 관리할 수 있습니다.
  • Spring Batch에서 PlatformTransactionManager는 배치 단계(Step)나 청크(chunk)를 처리할 때 트랜잭션을 시작하고 완료하는 데 사용됩니다. 이를 통해 데이터 일관성을 유지하고, 실패 시 트랜잭션을 롤백하는 등의 기능을 제공합니다.
• Step:
  • Spring Batch에서 배치 작업의 단위로, 배치 작업을 구성하는 여러 단계 중 하나를 의미합니다.
  • 각 Step은 특정 작업을 수행하는 작은 처리 흐름으로, 데이터를 읽고, 가공하고, 저장하는 과정을 포함합니다.
  • Spring Batch에서 Job은 여러 Step으로 구성되며, Step은 독립적으로 실행될 수 있습니다.
• StepBuilder:
  • Step을 구성하는 빌더 클래스입니다.
  • 이 빌더는 Step의 구성 요소(Reader, Processor, Writer 등)를 설정하고, 청크 기반 처리나 태스크 기반 처리와 같은 실행 방식을 정의하는 데 사용됩니다.
  • 이 빌더를 통해 Step을 유연하게 구성할 수 있습니다.
• <I, O>chunk(size):
  • Spring Batch에서 데이터를 청크(chunk) 단위로 처리하는 방식을 정의합니다.
  • 여기서 I는 입력 데이터 타입, O는 출력 데이터 타입을 의미합니다.
  • chunk(100)은 100개의 데이터를 한 번에 처리하고 나서 트랜잭션을 커밋하는 방식으로 동작합니다.
  • 청크 기반 처리는 대량의 데이터를 효율적으로 처리하기 위해 사용됩니다.
• reader(): 배치 작업에서 데이터를 읽어오는 역할을 하는 ItemReader를 설정합니다. 이 메서드를 통해 Step이 어떤 방식으로 데이터를 읽어올지를 정의할 수 있습니다. ItemReader는 데이터베이스, 파일, 메시지 큐, 또는 Redis와 같은 외부 소스에서 데이터를 읽어올 수 있습니다.
• processor(): 데이터를 가공하거나 변환하는 역할을 하는 ItemProcessor를 설정합니다. 이 메서드는 읽어온 데이터를 가공하거나 필터링한 후, 다음 단계로 전달할 형태로 변환하는 작업을 정의합니다. ItemProcessor는 입력 데이터와 출력 데이터의 타입을 다르게 설정할 수 있어, 데이터를 변환하거나 필터링하는 데 유용합니다.
• writer(): 배치 작업에서 데이터를 저장하는 역할을 하는 ItemWriter를 설정합니다. 이 메서드를 통해 가공된 데이터를 데이터베이스, 파일, 또는 메시지 큐 등에 기록할 수 있습니다. ItemWriter는 최종적으로 데이터를 저장하는 단계로, ItemReader와 ItemProcessor에서 처리된 데이터를 목적지에 저장합니다.

이러한 과정을 통해 Redis에 적재된 조회 이벤트 데이터를 MySQL로 옮겨와 집계할 수 있게 됩니다. Spring Batch의 유연한 구조 덕분에 복잡한 데이터 처리도 단계별로 쉽게 구현할 수 있습니다.

Aggregating Data in Database

지금까지 진행한 과정에서, 각 블로그 포스트에 대한 조회 이벤트를 데이터베이스에 저장하는 작업을 완료했습니다. 이제 각 포스트의 조회수를 계산할 수 있는데, 매번 모든 데이터를 조회하고 연산하게 되면 데이터베이스에 부하를 줄 수 있습니다. 이러한 상황에서는 일반적으로 반정규화를 통해 조회 수 합계를 별도로 저장해두고, 이를 이용해 성능을 최적화합니다. 이번 단계에서는 이전에 저장된 조회 이벤트 데이터를 기반으로 조회수의 증가분을 계산하고, 이를 조회수 테이블에 반영하는 작업을 구현하겠습니다.

이 작업에서는 MySQL에서 데이터를 조회하기 때문에, Spring Batch에서 제공하는 기본 ItemReader 구현체인 JdbcCursorItemReader를 활용합니다. 이 ItemReader는 MySQL에서 최근에 기록된 조회 이벤트 데이터를 가져와, 각 블로그 포스트의 조회수를 카운트한 결과를 반환하는 역할을 합니다:

@Slf4j
@RequiredArgsConstructor
@Configuration
public class HitEventItemReaderConfig {

    private final ClockHolder clockHolder;
    private final DataSource dataSource;

    @Bean
    @StepScope
    public JdbcCursorItemReader<HitEventCount> hitEventsCountItemReader() {
        JdbcCursorItemReader<HitEventCount> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);

        reader.setSql(
                """
                select
                    events.post_id as post_id,
                    count(events.post_id) as events_count
                from
                    HIT_EVENTS events
                where
                    events.created_at between ? and ?
                group by
                    events.post_id
                """);

        reader.setPreparedStatementSetter((ps) -> {
            ps.setTimestamp(1, clockHolder.lastMinuteTimestamp());
            ps.setTimestamp(2, clockHolder.currentTimestamp());
        });
        reader.setRowMapper(new HitEventCountRowMapper());
        return reader;
    }

    public static class HitEventCountRowMapper implements RowMapper<HitEventCount> {

        @Override
        public HitEventCount mapRow(ResultSet rs, int rowNum) throws SQLException {
            return new HitEventCount(
                    rs.getLong("post_id"),
                    rs.getLong("events_count")
            );
        }
    }
}

• JdbcCursorItemReader<T>: Spring Batch에서 제공하는 ItemReader 구현체로, 데이터베이스의 결과 집합을 커서(Cursor)를 사용해 순차적으로 읽어옵니다. 이를 통해 메모리 효율적으로 대량의 데이터를 처리할 수 있습니다.
  • setDataSource(): 데이터베이스와의 연결을 설정하는 메서드로, DataSource 객체를 주입받아 사용합니다.
  • setSql(): ItemReader가 실행할 SQL 쿼리를 설정합니다. 여기서는 특정 기간 내에 생성된 조회 이벤트를 조회해 각 포스트별 조회수를 집계합니다.
  • setRowMapper(): 데이터베이스에서 조회한 결과를 객체로 변환하는 RowMapper를 설정합니다. 여기서는 결과 집합에서 각 블로그 포스트의 ID와 조회 이벤트 카운트를 읽어 HitEventCount 객체로 변환합니다.

이렇게 읽어온 데이터를 MySQL의 조회수 테이블에 반영할 때는 중간 가공 작업 없이 바로 ItemWriter로 전달하여 업데이트 작업을 수행합니다. ItemWriter는 읽어온 데이터를 업데이트 쿼리로 처리하며, 아래와 같이 구현할 수 있습니다:

@Slf4j
@RequiredArgsConstructor
@Configuration
public class HitStatsItemWriterConfig {

    private final DataSource dataSource;

    @Bean
    public JdbcBatchItemWriter<HitEventCount> incrementHitStatsItemWriter() {
        return new JdbcBatchItemWriterBuilder<HitEventCount>()
                .itemSqlParameterSourceProvider(this::createSqlParameterSource)
                .sql("""
                     update HITS
                     set
                         total_hits = total_hits + :eventsCount,
                         updated_at = now()
                     where
                         post_id = :postId
                     """)
                .dataSource(dataSource)
                .build();
    }

    private MapSqlParameterSource createSqlParameterSource(HitEventCount hitEventCount) {
        return new MapSqlParameterSource()
                .addValue("postId", hitEventCount.postId())
                .addValue("eventsCount", hitEventCount.eventsCount());
    }
}

이제 ItemReader와 ItemWriter를 하나의 Step으로 결합하여, 조회수 증가분을 집계하고 데이터베이스에 반영하는 작업을 수행합니다. 이 Step은 Redis에서 읽어온 데이터를 바탕으로 조회수 테이블을 업데이트합니다.

@Slf4j
@Configuration
public class HitStatsBatchJobConfig {

    private final JdbcCursorItemReader<HitEventCount> hitEventsCountItemReader;
    private final JdbcBatchItemWriter<HitEventCount> incrementHitStatsItemWriter;
    ...
    
    @Bean
    public Step incrementHitStatsStep() {
        return new StepBuilder("increment-hit-stats-step", jobRepository)
                .<HitEventCount, HitEventCount>chunk(100, transactionManager)
                .reader(hitEventsCountItemReader)
                .writer(incrementHitStatsItemWriter)
                .build();
    }
}

이 Step은 데이터를 청크(chunk) 단위로 처리하며, MySQL에 업데이트된 조회수를 반영합니다. 이를 통해 조회 이벤트 데이터를 효율적으로 처리하고, 블로그 포스트의 총 조회수를 빠르게 계산할 수 있습니다.

Scheduling Batch Jobs

이제 Redis에 저장된 조회 이벤트 데이터를 MySQL로 옮기고, 추가된 조회 이벤트 증가분을 반영하는 두 개의 Step을 구현했습니다. 이 두 Step을 별도의 작업 단위로 처리할 수도 있지만, 하나의 Job 단위로 묶어 관리하는 것이 일반적인 방법입니다. 이 Job은 두 개의 Step을 순차적으로 실행하여 데이터를 처리하는 배치 작업을 구성합니다.

@Slf4j
@Configuration
public class HitStatsBatchJobConfig {

    ...
    
    @Bean
    public Job hitStatsJob() {
        return new JobBuilder("hit-stats-job", jobRepository)
                .start(persistHitEventsStep())
                .next(incrementHitStatsStep())
                .build();
    }

    @Bean
    public Step persistHitEventsStep() {...}

    @Bean
    public Step incrementHitStatsStep() {...}
}

• Job: Spring Batch에서 배치 작업을 실행하는 단위입니다. Job은 여러 Step을 순차적으로 실행하거나 조건에 따라 분기처리할 수 있습니다. 위 예제에서는 두 개의 Step을 차례로 실행하도록 설정한 hit-stats-job이라는 Job을 정의했습니다.

이 Job을 주기적으로 실행하기 위해 Spring의 스케줄러를 활용합니다. 스케줄링을 사용하면 배치 작업을 일정한 시간 간격마다 자동으로 실행할 수 있습니다. 블로그 포스트 하나를 읽는 데 걸리는 평균 시간을 고려하여, 3분마다 배치 작업이 실행되도록 설정했습니다:

@Slf4j
@RequiredArgsConstructor
@Profile({"local", "prod"})
@Component
public class HitStatsJobScheduler {

    private final JobLauncher jobLauncher;
    private final Job hitStatsJob;

    @Scheduled(cron = "0 */3 * * * *")
    public void runHitStatsJob() {
        try {
            JobParameters jobParameters = buildJobParameters("hit-stats-job", "hit-stats::");
            jobLauncher.run(hitStatsJob, jobParameters);
        } catch (JobExecutionException e) {
            log.error("runHitStatsJob: Failed hit stats job -", e);
        }
    }
}

• JobLauncher: Spring Batch에서 배치 작업을 실행하는 컴포넌트입니다. JobLauncher는 Job을 실행할 때 필요한 JobParameters와 함께 Job을 시작합니다. 여기서는 스케줄링된 작업이 실행될 때 hitStatsJob을 실행합니다.
• @Scheduled(cron): 스케줄링 애노테이션으로, 주기적으로 특정 메서드를 실행할 수 있도록 설정합니다. 여기서는 cron 표현식을 사용해 3분마다 runHitStatsJob() 메서드를 실행하도록 설정했습니다.

Just Hit It

이번 구현을 통해 Redis와 Spring Batch의 조합이 빈번한 조회 이벤트를 안정적으로 처리하면서도 서버 부하를 효과적으로 줄일 수 있다는 것을 확인했습니다. 이를 통해 시스템 성능을 유지하면서도, 추후 데이터를 분석하거나 다양한 방식으로 활용할 수 있는 기반이 마련되었습니다. 📊

현재 서비스에서는 조회수가 핵심 도메인이 아니기 때문에, 예외 상황이 발생해도 특별한 오류 처리를 하지 않고 데이터를 그대로 흘려보내는 방식으로 구현했습니다. 하지만 만약 조회수가 중요한 도메인이라면, 예외 상황을 무시하지 않고 재시도 로직이나 방어 코드를 추가하는 것이 필요합니다. 이를 통해 예기치 못한 오류로 인한 데이터 손실을 방지하고, 시스템의 신뢰성을 더욱 높일 수 있을 것입니다.

또한, Redis에 데이터를 기록하는 작업은 비동기로 처리하는 방법도 고려할 만합니다. 조회 이벤트가 과도하게 몰리면 Redis조차 버티기 어려운 상황이 발생할 수 있기 때문에, Rotation Lock이나 Kafka와 같은 메시징 큐 시스템을 도입해 확장성을 높이는 방법도 효과적일 것입니다. 이를 통해 시스템의 확장성을 더욱 강화하고, 안정적인 성능을 보장할 수 있습니다.

이제 프론트엔드와의 연동만 남았으니, 이 작업이 끝나면 조회 수 집계 기능이 완성됩니다.  마지막 단계까지 왔으니, 마무리만 잘하면 되겠네요… 😮‍💨