diff --git a/build.gradle b/build.gradle
index ccf557b..8ee9733 100644
--- a/build.gradle
+++ b/build.gradle
@@ -38,6 +38,8 @@ dependencies {
testCompileOnly 'org.projectlombok:lombok'
testAnnotationProcessor 'org.projectlombok:lombok'
+
+ implementation 'org.springframework.boot:spring-boot-starter-amqp'
}
tasks.named('test') {
diff --git a/docker-compose.yml b/docker-compose.yml
index d25f78e..3aff84c 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -15,6 +15,10 @@ services:
SPRING_DATASOURCE_URL: jdbc:mysql://mysql:3306/notification
SPRING_DATASOURCE_USERNAME: notification
SPRING_DATASOURCE_PASSWORD: notification
+ SPRING_RABBITMQ_HOST: rabbitmq
+ SPRING_RABBITMQ_PORT: 5672
+ SPRING_RABBITMQ_USERNAME: notification
+ SPRING_RABBITMQ_PASSWORD: notification
healthcheck:
test:
@@ -27,6 +31,12 @@ services:
retries: 5
start_period: 20s
+ depends_on:
+ mysql:
+ condition: service_healthy
+ rabbitmq:
+ condition: service_healthy
+
prometheus:
image: prom/prometheus:v3.13.0
@@ -99,6 +109,21 @@ services:
timeout: 3s
retries: 10
+ rabbitmq:
+ image: rabbitmq:4-management
+ container_name: notification-rabbitmq
+ ports:
+ - "5672:5672"
+ - "15672:15672"
+ environment:
+ RABBITMQ_DEFAULT_USER: notification
+ RABBITMQ_DEFAULT_PASS: notification
+ healthcheck:
+ test: [ "CMD", "rabbitmq-diagnostics", "-q", "ping" ]
+ interval: 10s
+ timeout: 5s
+ retries: 5
+
volumes:
prometheus-data:
grafana-data:
diff --git a/docs/04-experiment.md b/docs/04-experiment.md
index 39c964e..13ca2ad 100644
--- a/docs/04-experiment.md
+++ b/docs/04-experiment.md
@@ -178,6 +178,58 @@ Transaction 2
Provider 호출 방식과 지연 시간은 변경하지 않고,
DB Transaction의 범위만 변경하여 Connection 점유가 성능에 미치는 영향을 비교했다.
+### Experiment 2. RabbitMQ 비동기 처리
+
+Transaction Boundary 분리 후 DB Connection Pool 병목은 완화되었지만,
+Notification API는 여전히 Mock Provider 호출이 완료될 때까지 응답을 반환하지 않았다.
+
+한 요청은 Push 2건, SMS 1건, Email 1건으로 총 4번의 Provider 호출을 수행하며,
+각 Provider에 100ms의 지연이 존재하므로 최소 약 400ms의 응답 시간이 발생했다.
+
+Provider 호출을 HTTP 요청 처리 경로에서 제거하기 위해 RabbitMQ를 도입했다.
+
+```text
+기존
+
+HTTP Request
+ ↓
+DB 저장
+ ↓
+Provider 호출
+ ↓
+DB 상태 변경
+ ↓
+HTTP Response
+```
+```text
+RabbitMQ 적용 후
+
+HTTP Request
+ ↓
+DB 저장
+ ↓
+RabbitMQ Publish
+ ↓
+202 Accepted
+
+---------------------
+
+RabbitMQ Queue
+ ↓
+Consumer
+ ↓
+Mock Provider
+ ↓
+Delivery 상태 변경
+```
+
+Push, SMS, Email은 서로 다른 외부 Provider의 장애와 처리량에 독립적으로 대응할 수 있도록 채널별 Queue로 분리하였다.
+* `notification.push.queue`
+* `notification.sms.queue`
+* `notification.email.queue`
+
+API는 실제 알림 전송 완료를 기다리지 않고 요청 접수 후 `202 Accepted`를 반환한다.
+
## 11. 개선 후 결과
### Experiment 1. Transaction Boundary 분리 결과
@@ -203,6 +255,47 @@ Connection Pool의 지속적인 포화가 사라졌다.
+### Experiment 2. RabbitMQ 비동기 처리 결과
+
+| VU | 지표 | Transaction Boundary | RabbitMQ Async | 변화 |
+|---:|---|---:|---:|---:|
+| 30 | RPS | 65.88 | 496.78 | 7.54배 |
+| 30 | Avg | 454.66ms | 60.05ms | 86.8% 감소 |
+| 30 | p95 | 560.98ms | 168.54ms | 70.0% 감소 |
+| 30 | p99 | 1.47s | 316.49ms | 78.5% 감소 |
+| 30 | Error Rate | 0% | 0% | 동일 |
+| 50 | RPS | 117.74 | 511.23 | 4.34배 |
+| 50 | Avg | 421.79ms | 97.41ms | 76.9% 감소 |
+| 50 | p95 | 456.86ms | 266.21ms | 41.7% 감소 |
+| 50 | p99 | 515.22ms | 514.63ms | 유사 |
+| 50 | Error Rate | 0% | 0% | 동일 |
+
+RabbitMQ 비동기화 후 Provider 호출이 HTTP 요청 경로에서 제거되면서
+API 처리량이 크게 증가하고 평균 및 p95 응답 시간이 감소했다.
+
+50 VU에서는 최초 Baseline의 23.55 RPS에서 511.23 RPS로
+약 21.7배의 API 처리량 증가를 확인했다.
+
+그러나 높은 Producer 처리량으로 인해 Consumer가 메시지 유입 속도를 따라가지 못하면서
+Queue에 대량의 메시지가 적체되었다.
+
+#### Queue 상태
+| VU | PUSH Ready | SMS Ready | EMAIL Ready | Total Ready | Unacked |
+|---:|---:|---:|---:|---:|---:|
+| 30 | 28,494 | 13,572 | 13,577 | 55,643 | 750 |
+| 50 | 29,416 | 14,048 | 14,050 | 57,514 | 750 |
+
+따라서 RabbitMQ 적용으로 API의 응답 성능은 개선되었지만,
+실제 알림 전달 처리량의 병목은 Consumer 영역으로 이동했다.
+
+30 VU
+
+
+
+50 VU
+
+
+
## 12. 결과 분석
### 12.1 DB Connection Pool 병목 확인
@@ -242,12 +335,12 @@ Provider 호출 중에는 DB Connection을 점유하지 않으므로,
50 VU / 약 0.422초
≈ 118 RPS
```
-실제 처리량인 117.74 RPS와 유사핟.
+실제 처리량인 117.74 RPS와 유사하다.
-이를 통해 Transcation Boundary 분리 이후에는 DB Connection Pool보다
+이를 통해 Transaction Boundary 분리 이후에는 DB Connection Pool보다
동기식 Provider 호출 시간이 처리량과 응답 시간에 더 직접적인 영향을 주는 것으로 판단했다.
-### 12.3 남아 있는 병목
+### 12.3 Transaction Boundary 분리 후 남아 있던 병목
DB Connection Pool 병목은 완화되었지만,
Notification API는 여전히 모든 Provider 호출이 끝날 때까지 HTTP 응답을 반환하지 않는다.
@@ -259,6 +352,54 @@ Notification API는 여전히 모든 Provider 호출이 끝날 때까지 HTTP
Provider 호출을 HTTP 요청 처리 경로에서 분리하고,
API가 알림 요청을 Queue에 등록한 뒤 즉시 응답하도록 비동기 구조로 변경한다.
+### 12.4 Rabbit 비동기화 효과
+
+RabbitMQ 적용 후 HTTP 요청은 Mock Provider의 실행 완료를 기다리지 않고,
+Notification 및 Delivery를 저장한 뒤 메시지를 Queue에 발행하고 즉시 응답한다.
+
+30 VU에서는 RPS는 65.88에서 496.78로 약 7.54배 증가했고,
+p95는 560.98ms에서 168.54ms로 약 70% 감소했다.
+
+50 VU에서도 RPS는 117.74에서 511.23으로 4.34배 증가했으며,
+p95는 456.86ms에서 266.21ms로 감소했다.
+
+최초 Baseline과 비교하면 50 VU 기준 RPS는
+23.55에서 511.23으로 약 21.7배 증가했다.
+
+이는 Provider의 블로킹 I/O를 HTTP 요청 경로에서 분리함으로써
+API가 알림의 실제 전송 시간과 독립적으로 요청을 처리할 수 있게 된 결과다.
+
+### 12.5 새로운 병목: Consumer 처리량
+
+비동기화 이후 API 처리량은 크게 증가했지만,
+실제 알림 처리 속도가 함께 증가한 것은 아니다.
+
+30 VU 테스트에서 14,921개의 Notification 요청이 발생했다.
+테스트 사용자는 요청당 Push 2건, SMS 1건, Email 1건을 생성하므로
+약 59,684개의 Delivery 메시지가 RabbitMQ에 발행된다.
+
+테스트 종료 후 약 55,643개의 메시지가 Ready 상태로 남아 있었으며,
+각 Queue에서는 250개의 메시지가 Unacked 상태로 Consumer에 전달되어 있었다.
+
+50 VU에서도 약 57,514개의 Ready 메시지가 남았다.
+
+따라서 Producer인 Notification API의 처리량이 Consumer의 처리량보다 높아지면서
+Queue Backlog가 지속적으로 증가하는 새로운 병목이 발생했다.
+
+이는 비동기 시스템에서 API RPS만으로 전체 시스템 처리량을 평가할 수 업으며,
+다음 지표를 함께 측정해야 힘을 보여준다.
+
+- API 요청 처리량
+- Queue Backlog
+- 메시지 유입률
+- Consumer 처리량
+- End-to-End 알림 전달 지연
+
+다음 단계에서는 Consumer concurrency와 RabbitMQ prefetch 설정을 분석하고,
+Consumer 확장을 통해 Queue 적체를 완화할 수 있는지 실험한다.
+
+
+
## 13. Platform Thread와 Virtual Thread 비교
실제 블로킹 I/O가 존재하는 경우 반드시 수행한다.
diff --git a/docs/images/rabbitmq-async-30vu.png b/docs/images/rabbitmq-async-30vu.png
new file mode 100644
index 0000000..cc99ecf
Binary files /dev/null and b/docs/images/rabbitmq-async-30vu.png differ
diff --git a/docs/images/rabbitmq-async-50vu.png b/docs/images/rabbitmq-async-50vu.png
new file mode 100644
index 0000000..74a34c0
Binary files /dev/null and b/docs/images/rabbitmq-async-50vu.png differ
diff --git a/docs/images/rabbitmq-queue-30vu.png b/docs/images/rabbitmq-queue-30vu.png
new file mode 100644
index 0000000..c672356
Binary files /dev/null and b/docs/images/rabbitmq-queue-30vu.png differ
diff --git a/docs/images/rabbitmq-queue-50vu.png b/docs/images/rabbitmq-queue-50vu.png
new file mode 100644
index 0000000..f12d9ac
Binary files /dev/null and b/docs/images/rabbitmq-queue-50vu.png differ
diff --git a/k6/notification-async.js b/k6/notification-async.js
new file mode 100644
index 0000000..82a20ee
--- /dev/null
+++ b/k6/notification-async.js
@@ -0,0 +1,47 @@
+import http from 'k6/http';
+import { check } from 'k6';
+
+export const options = {
+ vus: 50,
+ duration: '30s',
+
+ summaryTrendStats: [
+ 'avg',
+ 'min',
+ 'med',
+ 'max',
+ 'p(90)',
+ 'p(95)',
+ 'p(99)',
+ ],
+
+ thresholds: {
+ http_req_failed: ['rate<0.01'],
+ },
+};
+
+const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080';
+
+export default function () {
+ const eventId = `baseline-${__VU}-${__ITER}-${Date.now()}`;
+
+ const payload = JSON.stringify({
+ eventId: eventId,
+ userId: 1,
+ channels: ['PUSH', 'SMS', 'EMAIL'],
+ });
+
+ const response = http.post(
+ `${BASE_URL}/api/v1/notifications`,
+ payload,
+ {
+ headers: {
+ 'Content-Type': 'application/json',
+ },
+ }
+ );
+
+ check(response, {
+ 'status is 202': (r) => r.status === 202,
+ });
+}
\ No newline at end of file
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessage.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessage.java
new file mode 100644
index 0000000..dded4ea
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessage.java
@@ -0,0 +1,11 @@
+package com.backendsystemdesignlab.notification.messaging;
+
+import com.backendsystemdesignlab.notification.user.domain.NotificationChannel;
+
+public record DeliveryMessage(
+ Long notificationId,
+ Long deliveryId,
+ NotificationChannel channel,
+ String destination
+) {
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryPublisher.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryPublisher.java
new file mode 100644
index 0000000..9f8fec4
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryPublisher.java
@@ -0,0 +1,33 @@
+package com.backendsystemdesignlab.notification.messaging;
+
+import com.backendsystemdesignlab.notification.notification.dto.DeliveryCommand;
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class DeliveryPublisher {
+
+ private final RabbitTemplate rabbitTemplate;
+
+ public void publishAll(Long notificationId, List deliveries) {
+ for (DeliveryCommand delivery : deliveries) {
+ publish(notificationId, delivery);
+ }
+ }
+
+ private void publish(Long notificationId, DeliveryCommand delivery) {
+ DeliveryMessage message = new DeliveryMessage(notificationId, delivery.deliveryId(), delivery.channel(), delivery.destination());
+
+ String routingKey = switch (delivery.channel()) {
+ case PUSH -> RabbitMqConfig.PUSH_ROUTING_KEY;
+ case SMS -> RabbitMqConfig.SMS_ROUTING_KEY;
+ case EMAIL -> RabbitMqConfig.EMAIL_ROUTING_KEY;
+ };
+
+ rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE, routingKey, message);
+ }
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java
new file mode 100644
index 0000000..bb6f345
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java
@@ -0,0 +1,74 @@
+package com.backendsystemdesignlab.notification.messaging;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMqConfig {
+
+ public static final String EXCHANGE = "notification.exchange";
+ public static final String PUSH_QUEUE = "notification.push.queue";
+ public static final String SMS_QUEUE = "notification.sms.queue";
+ public static final String EMAIL_QUEUE = "notification.email.queue";
+ public static final String PUSH_ROUTING_KEY = "notification.push";
+ public static final String SMS_ROUTING_KEY = "notification.sms";
+ public static final String EMAIL_ROUTING_KEY = "notification.email";
+
+ // Exchange: 메시지를 받아서 적절한 Queue로 전달하는 라우터
+ @Bean
+ public DirectExchange notificationExchange() {
+ return new DirectExchange(EXCHANGE, true, false); // Routing Key가 정확하게 일치하는 Queue로 보내기
+ // durable: RabbitMQ 서버가 재시작되어도 Exchange 정의를 유지하겠다, autoDelete: Consumer 등이 없어지면 Exchange 자동 삭제
+ }
+
+ @Bean
+ public Queue pushQueue() {
+ return new Queue(PUSH_QUEUE, true);
+ }
+
+ @Bean
+ public Queue smsQueue() {
+ return new Queue(SMS_QUEUE, true);
+ }
+
+ @Bean
+ public Queue emailQueue() {
+ return new Queue(EMAIL_QUEUE, true);
+ }
+
+ @Bean
+ public Binding pushBinding(DirectExchange notificationExchange, Queue pushQueue) {
+ // pushQueue를 notificationExchange에 연결하고, notification.push라는 Routing Key로 연결해라
+ return BindingBuilder
+ .bind(pushQueue)
+ .to(notificationExchange)
+ .with(PUSH_ROUTING_KEY);
+ }
+
+ @Bean
+ public Binding smsBinding(DirectExchange notificationExchange, Queue smsQueue) {
+ return BindingBuilder
+ .bind(smsQueue)
+ .to(notificationExchange)
+ .with(SMS_ROUTING_KEY);
+ }
+
+ @Bean
+ public Binding emailBinding(DirectExchange notificationExchange, Queue emailQueue) {
+ return BindingBuilder
+ .bind(emailQueue)
+ .to(notificationExchange)
+ .with(EMAIL_ROUTING_KEY);
+ }
+
+ @Bean
+ public MessageConverter messageConverter() {
+ return new JacksonJsonMessageConverter();
+ }
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/EmailDeliveryConsumer.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/EmailDeliveryConsumer.java
new file mode 100644
index 0000000..69c24d1
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/EmailDeliveryConsumer.java
@@ -0,0 +1,38 @@
+package com.backendsystemdesignlab.notification.messaging.consumer;
+
+import com.backendsystemdesignlab.notification.messaging.DeliveryMessage;
+import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig;
+import com.backendsystemdesignlab.notification.notification.provider.EmailProvider;
+import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
+import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class EmailDeliveryConsumer {
+
+ private final EmailProvider emailProvider;
+ private final NotificationTransactionService transactionService;
+
+ @RabbitListener(queues = RabbitMqConfig.EMAIL_QUEUE)
+ public void consume(DeliveryMessage message) {
+
+ boolean success;
+
+ try {
+ ProviderResult result = emailProvider.send(message.destination());
+ success = result.success();
+ } catch (RuntimeException e) {
+ success = false;
+ }
+
+ transactionService.recordDeliveryResult(
+ message.notificationId(),
+ message.deliveryId(),
+ success
+ );
+ }
+
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/PushDeliveryConsumer.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/PushDeliveryConsumer.java
new file mode 100644
index 0000000..2d67c6e
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/PushDeliveryConsumer.java
@@ -0,0 +1,38 @@
+package com.backendsystemdesignlab.notification.messaging.consumer;
+
+import com.backendsystemdesignlab.notification.messaging.DeliveryMessage;
+import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig;
+import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
+import com.backendsystemdesignlab.notification.notification.provider.PushProvider;
+import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class PushDeliveryConsumer {
+
+ private final PushProvider pushProvider;
+ private final NotificationTransactionService transactionService;
+
+ @RabbitListener(queues = RabbitMqConfig.PUSH_QUEUE)
+ public void consume(DeliveryMessage message) {
+
+ boolean success;
+
+ try {
+ ProviderResult result = pushProvider.send(message.destination());
+ success = result.success();
+ } catch (RuntimeException e) {
+ success = false;
+ }
+
+ transactionService.recordDeliveryResult(
+ message.notificationId(),
+ message.deliveryId(),
+ success
+ );
+ }
+
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/SmsDeliveryConsumer.java b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/SmsDeliveryConsumer.java
new file mode 100644
index 0000000..943dcd0
--- /dev/null
+++ b/src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/SmsDeliveryConsumer.java
@@ -0,0 +1,38 @@
+package com.backendsystemdesignlab.notification.messaging.consumer;
+
+import com.backendsystemdesignlab.notification.messaging.DeliveryMessage;
+import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig;
+import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
+import com.backendsystemdesignlab.notification.notification.provider.SmsProvider;
+import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class SmsDeliveryConsumer {
+
+ private final SmsProvider smsProvider;
+ private final NotificationTransactionService transactionService;
+
+ @RabbitListener(queues = RabbitMqConfig.SMS_QUEUE)
+ public void consume(DeliveryMessage message) {
+
+ boolean success;
+
+ try {
+ ProviderResult result = smsProvider.send(message.destination());
+ success = result.success();
+ } catch (RuntimeException e) {
+ success = false;
+ }
+
+ transactionService.recordDeliveryResult(
+ message.notificationId(),
+ message.deliveryId(),
+ success
+ );
+ }
+
+}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/notification/controller/NotificationController.java b/src/main/java/com/backendsystemdesignlab/notification/notification/controller/NotificationController.java
index 5f14af9..64b6f13 100644
--- a/src/main/java/com/backendsystemdesignlab/notification/notification/controller/NotificationController.java
+++ b/src/main/java/com/backendsystemdesignlab/notification/notification/controller/NotificationController.java
@@ -21,7 +21,7 @@ public class NotificationController {
@PostMapping
public ResponseEntity send(@Valid @RequestBody SendNotificationRequest request) {
SendNotificationResponse response = notificationService.send(request);
- return ResponseEntity.ok(response);
+ return ResponseEntity.accepted().body(response);
}
}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationDeliveryRepository.java b/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationDeliveryRepository.java
index 2a7c821..3a092db 100644
--- a/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationDeliveryRepository.java
+++ b/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationDeliveryRepository.java
@@ -1,8 +1,11 @@
package com.backendsystemdesignlab.notification.notification.repository;
+import com.backendsystemdesignlab.notification.notification.domain.DeliveryStatus;
import com.backendsystemdesignlab.notification.notification.domain.NotificationDelivery;
import org.springframework.data.jpa.repository.JpaRepository;
public interface NotificationDeliveryRepository extends JpaRepository {
long countByNotificationId(Long notificationId);
+
+ long countByNotificationIdAndStatus(Long notificationId, DeliveryStatus status);
}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationRepository.java b/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationRepository.java
index 803438c..be07ca9 100644
--- a/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationRepository.java
+++ b/src/main/java/com/backendsystemdesignlab/notification/notification/repository/NotificationRepository.java
@@ -1,10 +1,22 @@
package com.backendsystemdesignlab.notification.notification.repository;
import com.backendsystemdesignlab.notification.notification.domain.Notification;
+import jakarta.persistence.LockModeType;
import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
import java.util.Optional;
public interface NotificationRepository extends JpaRepository {
Optional findByEventId(String eventId);
+
+ @Lock(LockModeType.PESSIMISTIC_WRITE)
+ @Query("""
+ select n
+ from Notification n
+ where n.id = :id
+ """)
+ Optional findByIdForUpdate(@Param("id") Long id);
}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationService.java b/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationService.java
index 52ed969..eaeea0f 100644
--- a/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationService.java
+++ b/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationService.java
@@ -1,28 +1,17 @@
package com.backendsystemdesignlab.notification.notification.service;
-import com.backendsystemdesignlab.notification.notification.domain.NotificationStatus;
+import com.backendsystemdesignlab.notification.messaging.DeliveryPublisher;
import com.backendsystemdesignlab.notification.notification.dto.*;
-import com.backendsystemdesignlab.notification.notification.provider.EmailProvider;
-import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
-import com.backendsystemdesignlab.notification.notification.provider.PushProvider;
-import com.backendsystemdesignlab.notification.notification.provider.SmsProvider;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
-import java.util.ArrayList;
-import java.util.List;
-
@Service
@RequiredArgsConstructor
public class NotificationService {
private final NotificationTransactionService transactionService;
-
- private final PushProvider pushProvider;
- private final SmsProvider smsProvider;
- private final EmailProvider emailProvider;
-
+ private final DeliveryPublisher deliveryPublisher;
public SendNotificationResponse send(SendNotificationRequest request) {
@@ -39,47 +28,13 @@ public SendNotificationResponse send(SendNotificationRequest request) {
);
}
- // Provider 호출 (DB 트랜잭션을 사용하지 않음)
- List results = sendDeliveries(prepared.deliveries());
-
- // DB 작업
- transactionService.complete(prepared.notificationId(), results);
-
- boolean allSucceeded = results.stream().allMatch(DeliveryResult::success);
+ deliveryPublisher.publishAll(prepared.notificationId(), prepared.deliveries());
return new SendNotificationResponse(
prepared.notificationId(),
- allSucceeded
- ? NotificationStatus.COMPLETED
- : NotificationStatus.FAILED,
+ prepared.status(), // PROCESSING 비동기 이기 때문에 아직 Provider 전송이 안끝남
prepared.deliveryCount()
);
}
-
- private List sendDeliveries(List deliveries) {
-
- List results = new ArrayList<>();
-
- for (DeliveryCommand delivery : deliveries) {
- boolean success;
-
- try {
- ProviderResult result =
- switch (delivery.channel()) {
- case PUSH -> pushProvider.send(delivery.destination());
- case SMS -> smsProvider.send(delivery.destination());
- case EMAIL -> emailProvider.send(delivery.destination());
- };
- success = result.success();
- } catch (RuntimeException e) {
- success = false;
- }
-
- results.add(new DeliveryResult(delivery.deliveryId(), success));
- }
-
- return results;
- }
-
}
diff --git a/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationTransactionService.java b/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationTransactionService.java
index 73cb3ce..ff1cb95 100644
--- a/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationTransactionService.java
+++ b/src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationTransactionService.java
@@ -1,5 +1,6 @@
package com.backendsystemdesignlab.notification.notification.service;
+import com.backendsystemdesignlab.notification.notification.domain.DeliveryStatus;
import com.backendsystemdesignlab.notification.notification.domain.Notification;
import com.backendsystemdesignlab.notification.notification.domain.NotificationDelivery;
import com.backendsystemdesignlab.notification.notification.dto.DeliveryCommand;
@@ -108,6 +109,44 @@ public PreparedNotification prepare(SendNotificationRequest request) {
);
}
+ @Transactional
+ public void recordDeliveryResult(Long notificationId, Long deliveryId, boolean success) {
+ Notification notification = notificationRepository.findByIdForUpdate(notificationId)
+ .orElseThrow(() -> new IllegalArgumentException("알림을 찾을 수 없습니다."));
+
+ NotificationDelivery delivery = deliveryRepository.findById(deliveryId)
+ .orElseThrow(() -> new IllegalArgumentException("전송 정보를 찾을 수 없습니다."));
+
+ if (delivery.getStatus() == DeliveryStatus.SENT || delivery.getStatus() == DeliveryStatus.FAILED) {
+ return;
+ }
+
+ delivery.recordAttempt();
+
+ if (success) {
+ delivery.markSent();
+ } else {
+ delivery.markFailed();
+ }
+
+ deliveryRepository.flush();
+
+ long total = deliveryRepository.countByNotificationId(notificationId);
+ long sent = deliveryRepository.countByNotificationIdAndStatus(notificationId, DeliveryStatus.SENT);
+ long failed = deliveryRepository.countByNotificationIdAndStatus(notificationId, DeliveryStatus.FAILED);
+
+ // 아직 처리 중인 Delivery 존재
+ if (sent + failed < total) {
+ return;
+ }
+
+ if (failed > 0) {
+ notification.fail();
+ } else {
+ notification.complete();
+ }
+ }
+
@Transactional
public void complete(Long notificationId, List results) {
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index d96c8a8..5dedcec 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -19,6 +19,12 @@ spring:
ddl-auto: update
open-in-view: false
+ rabbitmq:
+ host: ${SPRING_RABBITMQ_HOST:localhost}
+ port: ${SPRING_RABBITMQ_PORT:5672}
+ username: ${SPRING_RABBITMQ_USERNAME:notification}
+ password: ${SPRING_RABBITMQ_PASSWORD:notification}
+
management:
endpoints:
web: