Repository navigation
[#37] 경매 입찰 큐 도입 - #38
Conversation
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars.
 |
| @Getter | ||
| @NoArgsConstructor(access = AccessLevel.PROTECTED) | ||
| @FieldDefaults(level = AccessLevel.PRIVATE) | ||
| public class BidRequestResult extends BaseEntity { |
There was a problem hiding this comment.
BidRequestResult 테이블과 Bids 테이블이 비슷한 역할을 한다고 생각이 들어서 합칠까했는디 고민했는데 별도 테이블로 분리하는것을 선택했습니다.
- bid 테이블은 성공한 입찰만 남는 경매 도메인 테이블입니다. 최고가 조회, 포인트 잠금, 낙찰 판정의 기준입니다.
- BidRequestResult 테이블은 비동기 요청의 처리 결과(성공/거절 + 사유)를 남기는 테이블입니다. 결과 조회와 중복 처리 방지에 사용할려고 추가했습니다
- 만약에 이 둘을 합치면 bid 테이블을 읽는 모든 쿼리에
status조건이 필요하고, 경합 시 대부분을 차지하는 거절 기록이 핵심 테이블에 쌓일 수 있다고 생각했습니다.
| import com.chalkak.bid.controller.request.BidRequest; | ||
| import java.util.UUID; | ||
|
|
||
| public record BidMessage( |
There was a problem hiding this comment.
FIFO 큐의 ContentBasedDeduplication=true는 메시지 본문 해시로 중복 발행을 걸러낸다고 해서,
본문에 requestId를 넣어서 같은 요청의 재전송만 걸러지게 했습니다.
만약 같은 메시지가 컨슈머에 다시 전달되는 경우는 컨슈머에서 requestId로 처리 여부를 확인하도록 하여 중복 처리를 방지했습니다.
There was a problem hiding this comment.
[Suggestion] 본문 해시로 걸러지는 건 send() 한 번 안에서 SDK가 재시도하는 경우까지예요. requestId는 request()가 호출될 때마다 새로 만들어져서, 클라이언트가 HTTP 요청을 다시 보내면 본문이 달라지고 중복으로 걸러지지 않아요.
예를 들어 SQS 전송은 성공했는데 202 응답 전에 클라이언트가 타임아웃 나서 재시도하면, 두 번째 요청은 새 requestId로 큐에 들어가 BID_AMOUNT_TOO_LOW로 거절돼요. 클라이언트는 두 번째 requestId로 조회해서 REJECTED를 보지만, 실제로는 첫 입찰이 성공한 상태예요. k6 스크립트에도 "타임아웃 건은 큐에 접수됐을 수 있다"고 적어 두셨는데, 실제 클라이언트도 같은 상황을 겪어요.
클라이언트가 Idempotency-Key 헤더로 요청 ID를 보내고 서버가 그 값을 requestId로 쓰면 어떨까요? 재시도도 같은 requestId로 들어오니 SQS 중복 제거(5분)와 컨슈머의 exists()가 같은 키로 동작해요.
그리고 지금은 중복 제거가 큐의 ContentBasedDeduplication 속성에 기대고 있는데, 이 속성은 localstack/init/create-queues.sh에만 있어요. SqsTemplate(3.4.0)은 이 속성이 꺼진 큐에는 임의 UUID를 중복 제거 ID로 넣어서, 운영 큐에서 속성을 빠뜨리면 에러 없이 중복이 그대로 통과해요. 코드에서 명시하면 의도도 더 잘 드러날 것 같아요.
sqsTemplate.send(to -> to
.queue(bidQueue)
.payload(message)
.messageGroupId(message.auctionId().toString())
.messageDeduplicationId(message.requestId().toString()));There was a problem hiding this comment.
넵 ! 멱등키 추가와 코드에서 명시적으로 messageDeduplicationId 추가하도록 하겠습니다.
피드백 감사합니다!
pjy1368
left a comment
pjy1368
left a comment
There was a problem hiding this comment.
큐 도입 범위를 소제목으로 나눠 주셔서 리뷰하기 편했어요. bid_requests를 bids와 분리한 이유를 남겨 주신 점, 다른 사람의 요청을 PENDING으로 숨긴 점, SQS 전송 동안 DB 커넥션을 잡지 않도록 request()를 트랜잭션에서 뺀 점이 좋았어요.
질문 주신 두 가지는 인라인 코멘트로 답했어요.
- 멱등성:
exists()와 결과 저장 사이에 틈이 있고save()가 merge로 동작해서, 재전달 때 ACCEPTED가 REJECTED로 덮어써지는 걸 로컬에서 재현했어요. 머지 전에 이 부분은 먼저 고쳐 주세요. - 처리량: 모듈 분리로 여러 경매 전체의 처리량은 늘지만, 그 전에 리스너 동시 처리 설정과 DB 커넥션 경합을 먼저 측정해 보면 좋겠어요.
그리고 운영 DB는 ddl-auto: validate라서 bid_requests 테이블이 없으면 서버가 기동하지 못하고, idx_bids_auction_id_bid_amount 인덱스도 자동으로 생기지 않아요. 운영 DB에는 DDL을 어떻게 반영하실 계획인가요?
| @SqsListener("${sqs.queue.bid}") | ||
| public void consume(Message<BidMessage> message) { | ||
| BidMessage bid = message.getPayload(); | ||
| if (bidRequestService.exists(bid.requestId())) { |
There was a problem hiding this comment.
[Blocking] 멱등성 질문에 대한 답변이에요.
exists() 확인과 accept() 저장이 서로 다른 트랜잭션이라, 같은 메시지가 두 번 전달되면 둘 다 exists() == false를 보고 처리로 들어갈 수 있어요. FIFO여도 처리 시간이 visibility timeout(큐 기본 30초, 리스너에는 연장 설정 없음)을 넘기면 같은 메시지가 다시 전달돼요.
이때 두 번째 처리는 경매 락을 기다렸다가 같은 금액이라 BID_AMOUNT_TOO_LOW로 거절되고 reject()가 결과를 저장해요. 그런데 requestId를 직접 넣는 엔티티라 save()가 INSERT가 아니라 merge(SELECT 후 UPDATE)로 동작해서, 이미 있는 ACCEPTED 행을 REJECTED로 덮어써요. 로컬에서 이 경로를 그대로 실행해 보니 ACCEPTED(bidId=1)가 REJECTED(AUCTION-004, bidId=null)로 바뀌었고 bids에는 입찰 1건이 남았어요. 입찰은 성공했는데 결과 조회에서는 거절로 보이게 돼요.
exists()는 빠른 확인용으로 두고, 최종 중복 방지는 bid_requests PK로 하면 어떨까요? 결과 저장이 실제 INSERT가 되면 중복 처리는 PK 위반으로 실패하고, 예외가 컨슈머 밖으로 나가 메시지가 다시 전달되면 그때는 exists()에서 걸러져요. BidRequestResult에 아래처럼 Persistable을 구현하면 돼요.
public class BidRequestResult extends BaseEntity implements Persistable<UUID> {
@Transient
boolean isNew = true;
@Override
public UUID getId() {
return requestId;
}
@Override
public boolean isNew() {
return isNew;
}
@PostLoad
@PostPersist
void markNotNew() {
this.isNew = false;
}
}이렇게 바꾸고 같은 상황을 돌려 보니 두 번째 저장은 duplicate key로 실패하고 행은 ACCEPTED로 남았어요. 저장 전 SELECT도 빠져서 메시지당 쿼리가 1개 줄어요. 같은 메시지를 두 번 처리해도 입찰 1건, 결과 ACCEPTED가 유지되는지 테스트로 남겨 두면 좋겠어요.
There was a problem hiding this comment.
넵! 요건 메세지 재전달 케이스를 고려를 못했습니다. 피드백 주신것처럼 DB 쪽에서도 멱등처리할 수 있도록 수정하도록 하겠습니다!
| private final BidRequestService bidRequestService; | ||
| private final SseEmitterStore sseEmitterStore; | ||
|
|
||
| @SqsListener("${sqs.queue.bid}") |
There was a problem hiding this comment.
[Suggestion] 처리량 질문에 대한 답변이에요.
경매(MessageGroupId)가 다르면 병렬로 처리되니까, 컨슈머를 별도 모듈로 빼서 늘리면 여러 경매 전체의 처리량은 늘어요. 다만 한 경매 안의 메시지는 어디서 돌든 하나씩 처리돼서, 인기 경매 하나의 처리량은 모듈을 나눠도 그대로예요. 모듈을 나누기 전에 지금 구조에서 아래 두 가지를 먼저 측정해 보면 어떨까요?
- 동시 처리 수: Spring Cloud AWS 기본값은 큐당 동시에 처리 중인 메시지 10개, poll 한 번에 10개예요. SQS FIFO는 메시지를 넘겨줄 때 같은 그룹 메시지로 최대한 채워서 줘요. 그래서 인기 경매가 있으면 10칸 대부분이 그 경매 메시지로 차고, 다른 경매 메시지는 기다려요.
spring.cloud.aws.sqs.listener.max-inflight-messages-per-queue와max-messages-per-poll만 올려도 같은 서버 안에서 경매끼리 병렬로 처리되는 수가 늘어요. - DB 커넥션 경합: 지금은 API와 컨슈머가 같은 Hikari 풀을 써요. 풀 설정이 없어서 기본값 10개예요. 컨슈머는 메시지 1건마다 트랜잭션을 2번(
exists(),accept()) 열어요. mixed 시나리오처럼 목록 조회가 몰리면 컨슈머가 커넥션을 기다리느라 처리량이 떨어질 수 있어요. 그때hikaricp_connections_pending값이 올라간다면, 모듈 분리의 효과는 병렬화보다 커넥션 풀을 따로 쓰게 되는 쪽에서 더 크게 나올 거예요.
한 경매의 처리량 상한은 대략 1초 ÷ 메시지 1건 처리 시간이라서, 1건에 20ms가 걸리면 초당 50건이에요. 이미 남기고 계신 consume_time, lock_acquire_time, consume_end_time과 ack 로그로 구간별 시간을 나눠 보면, 모듈 분리와 설정 조정 중 어느 쪽 효과가 큰지 숫자로 판단할 수 있어요. 모듈을 나누게 되면 SSE 알림 경로도 같이 바꿔야 해요(아래 32번째 줄 코멘트).
|
|
||
| try { | ||
| AuctionStatusResponse status = bidRequestService.accept(bid); | ||
| sseEmitterStore.broadcast(bid.auctionId(), AUCTION_STATUS_EVENT, status); |
There was a problem hiding this comment.
[Suggestion] SSE 전송이 컨슈머 스레드에서 동기로 실행돼요. SseEmitter.send()는 응답 스트림에 바로 쓰는 블로킹 호출이라, 데이터를 제때 읽지 못하는 구독자 때문에 소켓 버퍼가 차면 이 호출이 막히고 그 경매의 다음 입찰 처리도 같이 늦어져요. 구독자가 많은 인기 경매일수록 영향이 커요.
또 SseEmitterStore가 서버 메모리에 있어서, API 서버를 2대 이상 띄우거나 컨슈머를 별도 모듈로 빼면 컨슈머는 자기 서버에 연결된 구독자에게만 보낼 수 있어요.
입찰 처리와 알림 전달은 관심사가 다르니, 컨슈머는 결과 저장까지만 하고 알림은 이벤트로 넘기면 어떨까요? 처음에는 ApplicationEventPublisher와 비동기 리스너로 알림을 컨슈머 스레드에서 떼어 내고, 서버가 여러 대가 되면 그 리스너만 Redis Pub/Sub 같은 브로드캐스트로 바꾸면 돼요. 알림을 놓쳐도 GET /bid-requests/{requestId}로 결과를 확인할 수 있게 해 두셔서 이렇게 분리하기도 쉬울 것 같아요.
There was a problem hiding this comment.
SSE 블로킹 호출을 생각을 하지 못했는데 피드백 감사합니다! 수정하도록 하겠습니다.
| UUID requestId, | ||
| Long auctionId, | ||
| Long bidderId, | ||
| BidRequest bidRequest |
There was a problem hiding this comment.
[Suggestion] 큐 메시지가 컨트롤러 요청 DTO인 BidRequest를 그대로 담고 있어서, API 요청 형식을 바꾸면 큐 메시지 형식도 같이 바뀌어요. 큐에는 배포 전에 들어간 메시지가 남아 있을 수 있는데, 새 코드가 그 메시지를 역직렬화하지 못하면 처리가 계속 실패하고, FIFO라서 같은 경매의 뒤 메시지까지 전부 막혀요.
메시지에는 필요한 값만 직접 두면 어떨까요? (java.math.BigDecimal import도 같이 바꿔야 해요.)
| BidRequest bidRequest | |
| BigDecimal bidAmount |
TODO에 있는 실패 처리를 할 때 DLQ(maxReceiveCount)도 같이 두면, 메시지 하나 때문에 경매 전체가 멈추는 일을 막을 수 있어요.
There was a problem hiding this comment.
넵 ! 수정하도록 하겠습니다.
큐에 남아있는 메시지랑 형식이 안맞는 문제를 생각하지 못했습니다 피드백 주셔서 감사합니다!
| sqsTemplate.send(to -> to | ||
| .queue(bidQueue) | ||
| .payload(message) | ||
| .messageGroupId(message.auctionId().toString())); |
There was a problem hiding this comment.
[Nit] MessageGroupId로 경매별 순서가 보장되는데도 BidService.submit()의 비관적 락은 그대로 두셨네요. 마감 스케줄러가 같은 경매 행을 바꾸고, visibility timeout이 지나 재전달되면 같은 그룹 메시지가 동시에 처리될 수 있어서 유지하신 걸까요? 그렇다면 이 배경을 주석으로 남겨 주세요. 나중에 "큐가 순서를 보장하니 락은 빼도 되겠다"고 판단하는 일을 막을 수 있을 것 같아요.
There was a problem hiding this comment.
넵 이건 주석으로 남기도록 하겠습니다!
스케줄러가 접근할 수 있다고 판단하여 락을 그대로 두었습니다!
근데 말씀해주신것처럼 visibility timeout에 대해서는 고려못했는데 찾아보니 발생할 수 있는 문제점이었습니다.
피드백 주셔서 감사합니다
|
|
||
| private final SseEmitterStore sseEmitterStore; | ||
|
|
||
| @Scheduled(fixedRate = 30_000) |
There was a problem hiding this comment.
[Nit] heartbeat 주기(30초)와 SSE 타임아웃(30분)은 앞단 로드밸런서나 프록시의 idle timeout에 맞춰 바꿀 일이 생기는 값이에요. AuctionCloseScheduler처럼 설정으로 빼면 어떨까요?
| @Scheduled(fixedRate = 30_000) | |
| @Scheduled(fixedRateString = "${sse.heartbeat-interval}") |
SseEmitterStore의 TIMEOUT_MILLIS도 같은 방식으로 sse.timeout에서 읽으면 돼요.
There was a problem hiding this comment.
넵 ! 요건 외부에서 설정할 수 있도록 빼겠습니당
감사합니다.
| import com.chalkak.bid.controller.request.BidRequest; | ||
| import java.util.UUID; | ||
|
|
||
| public record BidMessage( |
There was a problem hiding this comment.
[Suggestion] 본문 해시로 걸러지는 건 send() 한 번 안에서 SDK가 재시도하는 경우까지예요. requestId는 request()가 호출될 때마다 새로 만들어져서, 클라이언트가 HTTP 요청을 다시 보내면 본문이 달라지고 중복으로 걸러지지 않아요.
예를 들어 SQS 전송은 성공했는데 202 응답 전에 클라이언트가 타임아웃 나서 재시도하면, 두 번째 요청은 새 requestId로 큐에 들어가 BID_AMOUNT_TOO_LOW로 거절돼요. 클라이언트는 두 번째 requestId로 조회해서 REJECTED를 보지만, 실제로는 첫 입찰이 성공한 상태예요. k6 스크립트에도 "타임아웃 건은 큐에 접수됐을 수 있다"고 적어 두셨는데, 실제 클라이언트도 같은 상황을 겪어요.
클라이언트가 Idempotency-Key 헤더로 요청 ID를 보내고 서버가 그 값을 requestId로 쓰면 어떨까요? 재시도도 같은 requestId로 들어오니 SQS 중복 제거(5분)와 컨슈머의 exists()가 같은 키로 동작해요.
그리고 지금은 중복 제거가 큐의 ContentBasedDeduplication 속성에 기대고 있는데, 이 속성은 localstack/init/create-queues.sh에만 있어요. SqsTemplate(3.4.0)은 이 속성이 꺼진 큐에는 임의 UUID를 중복 제거 ID로 넣어서, 운영 큐에서 속성을 빠뜨리면 에러 없이 중복이 그대로 통과해요. 코드에서 명시하면 의도도 더 잘 드러날 것 같아요.
sqsTemplate.send(to -> to
.queue(bidQueue)
.payload(message)
.messageGroupId(message.auctionId().toString())
.messageDeduplicationId(message.requestId().toString()));
|
개요
이번 PR에서는 크고작은 기능 변경이 많아서 변경 범위에 맞춰서 소제목을 나눴습니다.
입찰 API 비동기화
입찰 API(
POST /api/v1/auctions/{auctionId}/bids)를 동기 처리에서 SQS FIFO 큐 기반 비동기 처리로 변경202를 응답BidMessageConsumer가 메시지를 받아 기존BidService.submit(경매 행 비관적 락)으로 입찰을 처리하고, 결과를bid_requests테이블(BidRequestResult)에 저장MessageGroupId를 경매 ID로 지정requestId를 메시지 본문에 추가하여 중복 제거는 같은 요청이 재전송된 경우만 걸러내고, 서로 다른 요청은 모두 처리bid_requests테이블에 결과가 이미 있으면 처리를 건너뛰고 SQS가 같은 메시지를 다시 전달해도 결과가 한 번만 반영되도록 작성request()는 DB를 쓰지 않으므로@Transactional트랜잭션에서 뺐습니다. SQS 전송이 끝날 때까지 DB 커넥션을 붙잡지 않도록 했습니다.loadtest프로필에서만 동작하도록 했습니다.처리 결과 조회 API
GET /api/v1/bid-requests/{requestId}추가ACCEPTED/REJECTED(+errorCode) /PENDING중 하나입니다.PENDING으로 응답하여 다른 사람의 요청이 존재하는지는 드러나지 않도록 설정SSE 경매 이벤트 구독
GET /api/v1//auctions/{auctionId}/eventsSSE 구독 API 추가auction-status: 최고가 변경으로 인한 경매 상태를 구독자 전체에 전송bid-result: 입찰 처리 결과를 입찰자 본인에게 전송로컬 환경 구성
docker-compose.yml). init 스크립트가bid-queue.fifo를 만듭니다..env에LOCALSTACK_AUTH_TOKEN이 필요합니다.spring-boot-devtools를 제거했습니다.DB 인덱스
bids에(auction_id, bid_amount DESC)인덱스를 추가했습니다. (경매별 최고 입찰가 조회용)메인 리뷰어 지정 및 Due Date
리뷰 시 참고 사항
consume()이exists()로 결과 존재 여부를 확인한 뒤 처리하도록 되어있는데 멱등성 처리에 대해서 조금 더 고려해야하는 부분이 있는지 피드백 주시면 감사하겠습니다.consume()처리량 저하 문제는 consume을 별도 모듈로 추출하는것으로 해결해야하는지 궁금합니다.TODO
References
체크리스트