[쿠폰] 쿠폰 발급 수정 (saveAll, @Transactional)

for (MapRecord<String, Object, Object> message : messages) {
    // 중략
    CouponIssues issued = CouponIssues.builder()
            .coupons(Coupons.builder().id(Long.parseLong(couponIdStr)).build())
            .userId(userId)
            .build();
    couponIssuesRepository.save(issued);
    // 중략
}

기존 코드는 이렇게 CouponIssues 객체 하나씩 인서트하고 있었는데,

이렇게하면 Redis 등록하면서 인서트하는거나 마찬가지였다.

 

그래서 CouponId 기준으로 그룹지어 saveAll 처리하고 그에 따라 Redis Stream ack 처리하기로 했다.

하는 김에 쿠폰 재고 파악이 쉽도록 Coupons 에 issuedQuantity 컬럼을 추가해서 count() 없이 조회만으로 해결할 수 있도록 할 것이다.

 

---

 

 

1. Coupons Entity

@ColumnDefault("0")
@Column(nullable = false)
private int issuedQuantity;

기존 Coupons 클래스에 컬럼을 추가했다.

not null로 하고 deafult value 를 설정해야지만 null로 넣었을 때 deafult 값으로 들어간다.

 

 

 

2. CouponsRepository

@Modifying
@Query("UPDATE Coupons c SET c.issuedQuantity = c.issuedQuantity + :count, c.updatedAt = CURRENT_TIMESTAMP WHERE c.id = :couponId")
void incrementIssuedQuantityByCount(@Param("couponId") Long couponId, @Param("count") int count);

* DML(insert, update, delete) 쿼리를 사용할 때는 @Modifying 어노테이션을 선언해줘야한다.

 

UPDATE Coupons c 
   SET c.issuedQuantity = c.issuedQuantity + :count, 
       c.updatedAt = CURRENT_TIMESTAMP 
 WHERE c.id = :couponId

saveAll로 처리할 것이기 때문에 IssuedQuantity 도 특정 카운트를 더해줬다.

updateAt 에 @UpdateTimestamp 을 선언했더라도 Hibernate가 엔티티를 직접 관리하지 않고 쿼리를 날리는 것이기 때문에 작동하지 않는다. 그러므로 따로 컬럼 업데이트가 필요하다.

 

 

 

 

3. CouponIssueTransactionalService

@Slf4j
@RequiredArgsConstructor
@Service
public class CouponIssueTransactionalService {
    private final CouponIssuesRepository couponIssuesRepository;
    private final CouponsRepository couponsRepository;

    @Transactional
    public void saveIssueAndIncrementCount(Long couponId, List<CouponIssues> issues) {
        couponIssuesRepository.saveAll(issues);
        couponsRepository.incrementIssuedQuantityByCount(couponId, issues.size());
        log.info("Save coupon issued couponId: {}, count: {}", couponId, issues.size());
    }
}

coupon_issues 를 bulkInsert 하고, coupons의 issuedQuantity 를 증가시킨다.

 

동시성 문제와 롤백을 위해 @Transactional 을 선언해준 메서드를 다른 클래스에 정의했다.

같은 서비스에서 트랜잭션 메서드를 선언하면 self-invocation 문제가 생기기 때문에 다른 서비스에서 정의했다.

 

 

- saveAll 임에도 하나씩 Insert 되는 문제

saveAll 로 변경 후 확인해보니 로우 한 줄씩 인서트되는 로그가 보였다.

 

for (T entity : entities) {
    save(entity);
}

알아보니 saveAll() 구현체는 내부적으로 위와 같이 동작한다고 한다.

즉, 실제로는 save(entity)를 N번 반복 호출하고 있는 것이다;;

 

수가 얼마 얼마 안되는 서비스에서는 큰 영향이 없으나, 대용량 서비스에서는 batch insert 설정을 해줘야할 것 같다.

이건 추후에 보완해봐야겠다.

 

 

 

 

 

4. CouponWorker

private void saveCouponIssues(List<MapRecord<String, Object, Object>> messages) {
    Map<Long, List<CouponIssues>> issuesGroupedByCouponId = new HashMap<>();
    Map<Long, List<RecordId>> recordIdGroupedByCouponId = new HashMap<>();

    for (MapRecord<String, Object, Object> message : messages) {
        Map<Object, Object> value = message.getValue();

        Long couponId = Long.valueOf(String.valueOf(value.get("couponId")));
        String userId = String.valueOf(value.get("userId"));
        Timestamp createdAt = Timestamp.valueOf(String.valueOf(value.get("createdAt")));

        issuesGroupedByCouponId
                .computeIfAbsent(couponId, k -> new ArrayList<>())
                .add(CouponIssues.builder()
                        .coupons(Coupons.builder().id(couponId).build())
                        .userId(userId)
                        .createdAt(createdAt)
                        .build());

        recordIdGroupedByCouponId
                .computeIfAbsent(couponId, k -> new ArrayList<>())
                .add(message.getId());
    }

    for (Map.Entry<Long, List<CouponIssues>> entry : issuesGroupedByCouponId.entrySet()) {
        try {
            couponIssueTransactionalService.saveIssueAndIncrementCount(entry.getKey(), entry.getValue());
            redisTemplate.opsForStream()
                    .acknowledge(CouponConstants.STREAM_KEY, CouponConstants.GROUP_NAME, recordIdGroupedByCouponId.get(entry.getKey())
                            .toArray(new RecordId[0]));
        } catch (Exception e) {
            log.error("ERROR!", e);
        }
    }
}

messages를 반복해서 CouponId를 key로 갖는 맵을 만들고

key 기준으로 반복하며 CouponIssueTransactionalService에서 만들었던 메서드를 호출한다.

 

acknowledge는 메세지 처리가 되었음을 알리고 대기열에서 제거하는 메서드인데,

한개씩 또는 배열 형태로 모든 messageId를 넣어야지만 제거되기 때문에 성공한 messageId를 모두 넣어줘야한다.

 

 

5. 결과

여러번 CouponIssue를 해도 5초에 한번씩 쿼리가 실행된다.

 

DB에서도 coupon_issues 테이블의 로우 개수 만큼 issued_quantity 의 값이 증가된걸 확인할 수 있었다.

 

 

 

---

 

뭔가 하나를 수정하면 보완할게 계속 나오는 느낌이다...