[Architecture] Saga 패턴을 통한 분산 환경 비동기 트랜잭션 관리

Saga 패턴

소개

Saga 패턴 흐름도

Saga 패턴이란

분산 환경에서 데이터베이스 간 데이터 일관성을 유지하기 위해 사용하는 비동기 트랜잭션 관리 패턴이다.

MSA에선 2PC나 ACID 트랜잭션을 적용하기 어렵기 때문에 여러개의 로컬 트랜잭션의 연쇄로 나누어 처리한다. (2PC의 경우 강한 Lock으로 인한 정합성이 보장되지만 성능이 상당히 저하될 수 있다)

주요 특성

Saga 패턴의 가장 중요한 개념은 결과적 일관성과 보상 트랜잭션이다.

• 로컬 트랜잭션 분할 : 각 서비스는 자신의 DB만 수정하고 다음 단계로 이벤트를 발행
• 보상 트랜잭션 : 중간 단계에서 오류가 발생하면 이미 완료된 이전 트랜잭션들을 역순으로 취소
• 격리성 부재 : Lock을 걸지 않기 때문에 트랜잭션 중에 데이터가 다른 서비스에 노출될 수 있음

구현 방식

• Choreography 방식
  • 중앙 제어자 없이 이벤트 기반 흐름 제어
  • 서비스 간 이벤트 발행/구독
  • 결합도가 낮고 간단한 시스템에 적합
  • 흐름 추적이 어렵고 순환 참조 위험
• Orchestration 방식
  • 중앙 Orchestrator가 각 서비스에 Command 전송
  • 전체 트랜잭션 상태 파악 및 복잡한 흐름 관리가 용이
  • Orchestrator 복잡도 증가 및 단일 실패점 가능성

장단점

• 장점
  • Lock을 오래 유지하지 않으므로 높은 처리량 확보 가능
  • 각 서비스가 자체 DB와 로컬 트랜잭션을 사용하므로 결합도가 낮음

---

RabbitMQ Saga 패턴 구현

실습 프로젝트 흐름도

RabbitMQ로 Saga 패턴을 구현한다. 아래 예제 코드는 OrderService만 포함한다. 구현 방식은 위에서 소개한 Choreography 방식이다.

RabbitMQ에 대한 내용은 다음 포스팅 참조 https://nohumbit.tistory.com/57

[Spring boot] RabbitMQ

OrderApplication

• application.yml

spring:
    application:
        name: order
    rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

message:
    exchange: market
    err:
        exchange: market.err
    queue:
        product: market.product
        err:
            order: market.err.order

• OrderQueueConfig.java

@Configuration
public class OrderApplicationQueueConfig {

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Value("${message.exchange}")
    private String exchange;

    @Value("${message.queue.product}")
    private String queueProduct;

    @Value("${message.queue.payment}")
    private String queuePayment;

    @Value("${message.err.exchange}")
    private String exchangeErr;

    @Value("${message.queue.err.order}")
    private String queueErrOrder;

    @Value("${message.queue.err.product}")
    private String queueErrProduct;

    @Bean public TopicExchange exchange() { return new TopicExchange(exchange); }
    @Bean public TopicExchange exchangeErr() { return new TopicExchange(exchangeErr); }

    @Bean public Queue queueProduct() { return new Queue(queueProduct); }
    @Bean public Binding bindingProduct() { return BindingBuilder.bind(queueProduct()).to(exchange()).with(queueProduct); }
    @Bean public Queue queueErrProduct() { return new Queue(queueErrProduct); }
    @Bean public Binding bindingErrProduct() { return BindingBuilder.bind(queueErrProduct()).to(exchangeErr()).with(queueErrProduct); }

    @Bean public Queue queuePayment() { return new Queue(queuePayment); }
    @Bean public Binding bindingPayment() { return BindingBuilder.bind(queuePayment()).to(exchange()).with(queuePayment); }
    @Bean public Queue queueErrOrder() { return new Queue(queueErrOrder); }
    @Bean public Binding bindingErrOrder() { return BindingBuilder.bind(queueErrOrder()).to(exchangeErr()).with(queueErrOrder); }
}

• DeliveryMessage.java

@Builder
public record DeliveryMessage(
      UUID orderId,
    UUID paymentId,

    String userId,

    Integer productId,
    Integer productQuantity,

    Integer payAmount,

    String errorType
) {
}

• Order.java

@Getter
@AllAgrsConstructor
@NoArgsConstructor
@ToString
@Builder
public class Order {
    private UUID orderId;
    private String userId;
    private String orderStatus;
    private String errorType;

    public void cancelOrder(String receiveErrorType) {
        orderStatus = "CANCEL";
        errorType = receiveErrorType;
    }
}

• OrderEndpoint.java

@Slf4j
@RestController
@RequiredArgsConstructor
public class OrderEndpoint {
    private final OrderService orderService;
    private final RabbitTemplate rabbitTemplate;

    @GetMapping("order/{orderId}")
    public ResponseEntity<Order> getOrder(@PathVariable UUID orderId) {
        Order order = orderService.getOrder(orderId);
        return ResponseEntity.ok(order);
    }

    @PostMapping("/order")
    public ResponseEntity<Order> order(@RequestBody OrderRequestDto orderRequestDto) {
        Order order = orderService.createOrder(orderRequestDto);
        return ResponseEntity.ok(order);
    }

    @RabbitListener(queues = "${message.queue.err.order}")
    public void errOrder(DeliveryMessage message) {
        log.info("ERROR RECEIVE !!!");
        orderService.rollbackOrder(message);
    }
}

• OrderRequestDto.java

public record OrderRequestDto( 
    String userId,
    Integer productId,
    Integer productQuantity,
    Integer payAmount
) {
    public Order toOrder(){
        return Order.builder()
                .orderId(UUID.randomUUID())
                .userId(userId)
                .orderStatus("RECEIPT")
            .build();
    }

    public DeliveryMessage toDeliveryMessage(UUID orderId){
        return DeliveryMessage.builder()
                .orderId(orderId)
                .productId(productId)
                .productQuantity(productQuantity)
                .payAmount(payAmount)
            .build();
    }
}

• OrderService.java

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderService {

    @Value("${message.queue.product}")
    private String productQueue;

    @Value("${message.exchange}")
        private String exchange;

    private final RabbitTemplate rabbitTemplate;

    private Map<UUID, Order> orderStore = new ConcurrentHashMap<>();

    public Order createOrder(OrderEndpoint.OrderRequestDto orderRequestDto) {
        Order order = orderRequestDto.toOrder();
        DeliveryMessage deliveryMessage = orderRequestDto.toDeliveryMessage(order.getOrderId());

        orderStore.put(order.getOrderId(), order);

        log.info("send Message : {}",deliveryMessage.toString());

        rabbitTemplate.convertAndSend(exchange, productQueue, deliveryMessage);

        return order;
    }

    public void rollbackOrder(DeliveryMessage message) {
        Order order = orderStore.get(message.getOrderId());
        order.cancelOrder(message.getErrorType());
        log.info(order.toString());

    }

    public Order getOrder(UUID orderId) {
        return orderStore.get(orderId);
    }
}

유의점

위의 코드 흐름은 1. 다음 서비스로 이벤트 전달, 2. 에러 리스너 동작 시 HashMap에서 꺼내 Rollback 하는 방식이다.

실무에서는 아래와 같은 점들을 유의해야 한다.

• Transaction
  • 지금은 메모리(HashMap)에 주문을 저장하지만, 실제로는 DB에 ORDER_RECEIVED나 PENDING 상태로 저장해야 함
  • 롤백 요청이 오면 DB에서 주문을 CANCELLED로 UPDATE하고 보상 로직을 수행해야 함
• 멱등성
  • 메시지 브로커는 네트워크 이상 등으로 인해 같은 에러 메시지를 2번 이상 보낼 수도 있음(At-Least-Once 전달)
  • 따라서 상태를 통해 중복에 대한 멱등성을 보장해야 함
• 메시지 유실 방지 (Transactional Outbox 패턴)
  • DB에 주문은 저장됐는데, RabbitMQ로 메시지를 쏘기 직전에 서버가 다운되면 메시지가 유실될 수 있음
  • 실무에선 DB에 메시지를 저장해 두었다가 브로커로 발행하는 방식으로 원자성 보장
• Event의 형태
  • 현재는 범용 DTO를 사용하지만, 구간 별로 Message 형태를 따로 정의하는 것이 나을 수 있음
• Listener 레이어
  • 이벤트 Listener는 비동기로 처리되므로 Controller가 아닌 Component로 관리되어야 함 (listener 또는 consumer 패키지)