반응형

스프링을 사용하다보면 @Transactional을 굉장히 많이 사용한다.

 

에러가 발생하면 transaction을 rollback 시켜주기 때문에 사용한다라고는 알고 있지만, 어떠한 방식으로 동작하는지는 제대로 알아보지 않은 것 같다.

 

우선 블로그들과 spring framework의 github를 참고했다.

 

@Transactional을 사용하면 interceptor가 중간에 가져와서 invokeWithinTransaction가 실행되도록 한다.

 

안에서 작성된 코드는 이정도 되는 것 같다.

 

https://github.com/spring-projects/spring-framework/blob/main/spring-tx/src/main/java/org/springframework/transaction/annotation/Transactional.java

protected @Nullable Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
			final InvocationCallback invocation) throws Throwable {

		// If the transaction attribute is null, the method is non-transactional.
		TransactionAttributeSource tas = getTransactionAttributeSource();
		final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
		final TransactionManager tm = determineTransactionManager(txAttr, targetClass);

		if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager rtm) {
			boolean isSuspendingFunction = KotlinDetector.isSuspendingFunction(method);
			boolean hasSuspendingFlowReturnType = isSuspendingFunction &&
					COROUTINES_FLOW_CLASS_NAME.equals(new MethodParameter(method, -1).getParameterType().getName());

			ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {
				Class<?> reactiveType =
						(isSuspendingFunction ? (hasSuspendingFlowReturnType ? Flux.class : Mono.class) : method.getReturnType());
				ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(reactiveType);
				if (adapter == null) {
					throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type [" +
							method.getReturnType() + "] with specified transaction manager: " + tm);
				}
				return new ReactiveTransactionSupport(adapter);
			});

			return txSupport.invokeWithinTransaction(method, targetClass, invocation, txAttr, rtm);
		}

		PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
		final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

		if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager cpptm)) {
			// Standard transaction demarcation with getTransaction and commit/rollback calls.
			TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

			Object retVal;
			try {
				// This is an around advice: Invoke the next interceptor in the chain.
				// This will normally result in a target object being invoked.
				retVal = invocation.proceedWithInvocation();
			}
			catch (Throwable ex) {
				// target invocation exception
				completeTransactionAfterThrowing(txInfo, ex);
				throw ex;
			}
			finally {
				cleanupTransactionInfo(txInfo);
			}

			if (retVal != null && txAttr != null) {
				TransactionStatus status = txInfo.getTransactionStatus();
				if (status != null) {
					if (retVal instanceof Future<?> future && future.isDone()) {
						try {
							future.get();
						}
						catch (ExecutionException ex) {
							Throwable cause = ex.getCause();
							Assert.state(cause != null, "Cause must not be null");
							if (txAttr.rollbackOn(cause)) {
								status.setRollbackOnly();
							}
						}
						catch (InterruptedException ex) {
							Thread.currentThread().interrupt();
						}
					}
					else if (vavrPresent && VavrDelegate.isVavrTry(retVal)) {
						// Set rollback-only in case of Vavr failure matching our rollback rules...
						retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
					}
				}
			}

			commitTransactionAfterReturning(txInfo);
			return retVal;
		}

		else {
			Object result;
			final ThrowableHolder throwableHolder = new ThrowableHolder();

			// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
			try {
				result = cpptm.execute(txAttr, status -> {
					TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);
					try {
						Object retVal = invocation.proceedWithInvocation();
						if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
							// Set rollback-only in case of Vavr failure matching our rollback rules...
							retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
						}
						return retVal;
					}
					catch (Throwable ex) {
						if (txAttr.rollbackOn(ex)) {
							// A RuntimeException: will lead to a rollback.
							if (ex instanceof RuntimeException runtimeException) {
								throw runtimeException;
							}
							else {
								throw new ThrowableHolderException(ex);
							}
						}
						else {
							// A normal return value: will lead to a commit.
							throwableHolder.throwable = ex;
							return null;
						}
					}
					finally {
						cleanupTransactionInfo(txInfo);
					}
				});
			}
			catch (ThrowableHolderException ex) {
				throw ex.getCause();
			}
			catch (TransactionSystemException ex2) {
				if (throwableHolder.throwable != null) {
					logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
					ex2.initApplicationException(throwableHolder.throwable);
				}
				throw ex2;
			}
			catch (Throwable ex2) {
				if (throwableHolder.throwable != null) {
					logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
				}
				throw ex2;
			}

			// Check result state: It might indicate a Throwable to rethrow.
			if (throwableHolder.throwable != null) {
				throw throwableHolder.throwable;
			}
			return result;
		}
	}

 

우선 트랜잭션의 속성과 매니저를 가져온다.

TransactionAttributeSource tas = getTransactionAttributeSource();
TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
TransactionManager tm = determineTransactionManager(txAttr, targetClass);

 

특별한 설정이 없다면, 트랜잭션을 처리한다.

PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
...
if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager cpptm)) {

}

 

아래의 메서드로 트랜잭션이 진행된다.

 

  • 트랜잭션 시작 (createTransactionIfNecessary)
  • 비즈니스 로직 실행 (invocation.proceedWithInvocation())
  • 예외 발생 시 롤백 (completeTransactionAfterThrowing)
  • 정상 종료 시 커밋 (commitTransactionAfterReturning)

 

아래의 메서드로 에러 발생시 롤백이 예약된다.

status.setRollbackOnly();

 

 

 

만약, 해당 트랜잭션이 중첩이 되어 있다면

호출한 트랜잭션으로 전파된다.

if (throwableHolder.throwable != null) {
    throw throwableHolder.throwable;
}
return result;

 

간단하게 정리하자면

 

  • Spring이 트랜잭션 어노테이션이 붙은 메서드를 감지
  • 그 메서드를 프록시(Proxy) 객체로 감쌉니다
  • 실제 메서드를 호출할 때, 프록시가 트랜잭션 관련 로직을 앞뒤로 끼워 넣습니다

이런식으로 트랜잭션이 처리된다.

 

반응형

면접에서 나왔던 질문이지만, 내가 대답하지 못했던 부분에 대해서 알아보려고 한다.

 

우선 트랜잭션이 무엇인지, 왜 필요한지는 이미 안다고 생각한다.

 

그리고 당연히 트랜잭션의 레벨이 높아질수록, 동시성이 떨어지기 때문에 성능이 떨어진다.

하지만 그만큼 고립시켜서 동작하기 때문에 일관성은 높아진다.

 

우선 크게

Read Uncommitted, Read Committed, Repeatable Read, Serializable

이렇게 4가지로 나뉘며, 오른쪽으로 갈수록 레벨이 높아진다.

 

  • Read Uncommitted

트랜잭션이 처리중이거나, 아직 commit되지 않은 데이터를 다른 트랜잭션이 읽는 것을 허용한다.

 

 

이렇게 그냥 다른 트랜잭션에서 처리를 하고 있더라도, 데이터를 그냥 조회해서 값을 가져온다.

하지만 조회 후, 데이터가 rollback 되어 버리면 부정확한 데이터를 가져오게 된다.

 

이런 Dirty Read가 발생할 수 있기 때문에, 정확도가 너무 낮아 표준에서 인정하지 않는 격리 수준이라고 한다.

 

  • Read Committed

커밋된 데이터만 조회할 수 있도록 한다.

그렇기에 다른 트랜잭션이 접근 할 수 없어 대기하게 된다.

그렇기에 commit 되기 전의 데이터를 읽는 Dirty Read는 발생하지 않게 된다.

 

하지만, 하나의 트랜잭션에서 다른 트랜잭션의 commit에 따라 데이터의 조회 결과가 달라지는 Non-Repeatable Read 문제가 생기게 된다.

그렇지만 자주 발생하는 문제는 아니며, 대부분의 기본 설정이 이 Read Commited라고 한다.

 

  • Repeatable Read

보통의 RDBMS는 변경 전의 데이터를 Undo 공간에 백업해둔다.

그렇기에 트랜잭션의 번호를 확인해, 자신보다 더 늦은 트랜잭션의 번호가 존재한다면 이 Undo 공간에서 데이터를 조회하게 된다.

그렇기 때문에, 다른 트랜잭션에 의해 변경되더라도 동일한 데이터를 조회할 수 있게 된다.

 

하지만 조회 트랜잭션동안 새로운 데이터가 추가되는 것을 감지하지 못하는 Phantom Read가 발생할 수도 있다고 한다.

그래도 이 Phantom Read는 트랜잭션의 번호를 확인해 무시하는 방법으로 해결 할 수 있으며, 현재의 RDBMS에서는 특정 상황이 아니라면 대부분 발생하지 않는 현상이기에 크게 신경쓰지는 않아도 된다고 한다.

 

  • Serializable

마지막은 가장 강력한 Serializable 레벨이다.

그냥 모든 트랜잭션을 순차적으로 실행시키기에 위에 소개했던 모든 정합성 관련 문제들이 발생하지 않는다.

하지만 동시성이 없어지기에 성능이 매우 떨어진다.

극단적으로 안정성이 필요한 경우가 아니라면 잘 사용하지 않는다고 한다.

반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리

Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

 

이제 debezium으로 발행한 이벤트를 소비 해보도록 하자.

 

오늘도 공식문서를 참고하여 글을 작성한다.

https://debezium.io/documentation/reference/stable/connectors/mongodb.html

 

Debezium connector for MongoDB :: Debezium Documentation

A long integer value that specifies the maximum volume of the blocking queue in bytes. By default, volume limits are not specified for the blocking queue. To specify the number of bytes that the queue can consume, set this property to a positive long value

debezium.io

 

우선 해당 컬렉션에서 변경사항이 생기면

 

이렇게 kafka에 이벤트가 전송되어야 한다.

 

그리고 앞으로 서버에서 직접 kafka로 publish 하는 게 아니기 때문에 잠깐 publish를 주석처리해 두자.

 

이렇게 일단 서버에서 직접 발행이 되지 못하도록 설정해두었다.

 

그리고 일단 listener를 통해 

    @KafkaListener(id = "\${kafka.consumer.payment-consumer-group-id}",
        topics = ["\${kafka.topic.payment-request}"])
    fun receive(
        @Payload value: String,
        @Header(KafkaHeaders.RECEIVED_KEY) keys: String,
        @Header(KafkaHeaders.RECEIVED_PARTITION) partitions: Int,
        @Header(KafkaHeaders.OFFSET) offsets: Long
    ) {

        logger.info("Received $value from $partitions partitions offsets $offsets")
    }

 

 

이렇게하고 데이터베이스를 변경해보았더니, 다음과 같은 문자열이 출력되었다.

 

이제 이거를 json으로 만들어보자.

 

ObjectMapper를 사용해서 json으로 만들어주었다.

https://jsonformatter.org/#google_vignette

 

Best JSON Formatter and JSON Validator: Online JSON Formatter

Online JSON Formatter / Beautifier and JSON Validator will format JSON data, and helps to validate, convert JSON to XML, JSON to CSV. Save and Share JSON

jsonformatter.org

여기에 json을 붙여넣으면 알아서 이쁘게 만들어준다.

 

여기서 payload를 잘봐야 한다.

우선 op가 연산의 종류를 말해준다.

C가 insert, D가 Delete, U가 Update이다.

우리는 여기서 생성되었음을 감지하는 C 일 때만 함수를 실행해야 하기에, 해당 타입일 때만 동작하게 해준다.

 

여기서 우선 필요한 값들을 추출하고, 기존에 호출하던 함수로 연결을 해주었다.

 

@KafkaListener(id = "\${kafka.consumer.payment-consumer-group-id}",
        topics = ["\${kafka.topic.payment-request}"])
    fun receive(
        @Payload value: String,
        @Header(KafkaHeaders.RECEIVED_KEY) keys: String,
        @Header(KafkaHeaders.RECEIVED_PARTITION) partitions: Int,
        @Header(KafkaHeaders.OFFSET) offsets: Long
    ) {

        val cdcJson = objectMapper.readTree(value)

        if(cdcJson["payload"]["op"].asText() == "c"){

            val paymentRequestJson = objectMapper.readTree(cdcJson["payload"]["after"].asText())

            val paymentRequestDto = PaymentRequestDto(
                id = paymentRequestJson["_id"]["\$oid"].asText(),
                customerId = paymentRequestJson["payload"]["customerId"]["\$oid"].asText(),
                price = paymentRequestJson["payload"]["price"]["\$numberLong"].asLong(),
                createdAt= LocalDateTime.ofEpochSecond(paymentRequestJson["createdAt"]["\$date"].asLong() / 1000, 0, ZoneOffset.UTC),
                paymentOrderStatus = PaymentOrderStatus.valueOf(paymentRequestJson["payload"]["paymentOrderStatus"].asText()),
            )

            logger.info("paymentRequestDto $paymentRequestDto")

            if(paymentRequestDto.paymentOrderStatus == PaymentOrderStatus.PENDING){
                logger.info("주문 {}의 결제가 진행 중입니다", paymentRequestDto.id)
                paymentRequestMessageListener.completePayment(paymentRequestDto)
            }else{
                logger.info("주문 {}의 결제가 취소 중입니다", paymentRequestDto.id)
                paymentRequestMessageListener.cancelPayment(paymentRequestDto)
            }.subscribe()
        }
    }

 

json에서 값을 가져와서 그대로 함수를 호출해주었다.

 

이렇게 바꿔준 후 Order 서버에서 새로운 주문을 요청해보았더니

 

Order 서버의 Payment_outbox를 감지해서 Payment 서버에서 이러한 이벤트가 다시 Order 서버로 발행되었다.

 

이렇게 데이터베이스에서의 변경을 감지하고 이벤트를 전송하는 Debezium을 사용해보았다.

 

이렇게 만들기는 했지만, 사실 Json을 사용하는 것보다 Avro model을 사용하는 것이 더 성능이 좋다고 한다.(kakao에서 저장공간의 최적화를 위해 Avro를 사용한다고 한다.)

 

Avro로 시도를 해보았지만, 너무 어려워서 실패했다.

그리고 전체 서비스를 Debezium outbox로 교체한 것은 아니었다.

 

이렇게 MSA에서 필수적으로 사용하는 패턴들만 사용을 해보고, 후에 Toge-do 앱 리펙토링 할 때 반영해보려고 한다.

반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리

Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

저번 포스팅에서 debezium 컨테이너를 생성했었다.

 

이제 이 debezium을 통해 kafka topic을 만들어보자.

 

늘 그렇듯, 공식문서를 보면서 한다.

https://debezium.io/documentation/reference/stable/connectors/mongodb.html

 

Debezium connector for MongoDB :: Debezium Documentation

A long integer value that specifies the maximum volume of the blocking queue in bytes. By default, volume limits are not specified for the blocking queue. To specify the number of bytes that the queue can consume, set this property to a positive long value

debezium.io

 

 

해당 항목을 참고하여 Debezium으로 http 요청을 보낸다.

 

  1.  그냥 Debezium에서 사용할 이름이다. 토픽이랑 비슷하게 맞추어서 생성하면 된다.
  2. connector.class는 mongodb를 사용하기 때문에 저 내용 그대로 넣어주면 된다. 데이터베이스마다 다 다르니, 공식문서..를 찾아 들어가서 넣어주면 된다.
  3. Mongodb 연결 주소이다. 당연히 비밀번호와 계정이 있다면 넣어주어야 하고, replicaSet의 정보도 주어야 한다.
  4. 토픽 이름의 prefix이다. 저기에 지정한 prefix에 따라 토픽의 이름이 생성된다. prefix가 A이고, B 데이터베이스의 C 컬렉션이면 토픽의 이름은 A.B.C로 생성이 된다.
  5. 변화를 감지할 데이터베이스이다. B 데이터베이스의 C 컬렉션이면 이 곳에는 B.C로 넣어주면 된다.

 

일단, debezium을 통해 토픽을 작성하는데 필요한 필수정보는 끝났다.

만약 더 추가할 내용이 있다면 공식문서를 통해 추가하도록 하자.

 

이제 POST로 debezium에게 요청하면 된다.

예시를 보면

POST http://localhost:8083/connectors/
Content-Type: application/json

{
  "name": "order-payment-request-connector",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "tasks.max": "1",
    "mongodb.connection.string": "",
    "mongodb.authSource": "admin",
    "collection.include.list": "orders.payment_outboxes",
    "topic.prefix": "debezium",
    "tombstones.on.delete": "false"
  }
}

 

이렇게 요청하면 된다.

나는 Intellij Http를 통해 요청했다.

 

요청을 보내고 kafka-ui에 보면

connect-status 토픽에 

이런 식으로 무슨 내용이 와있다.

이러면 설정이 된것이다.

 

만약 모든 topic을 조회하고 싶으면

GET http://localhost:8083/connectors/

 

해당 uri로 요청하면 된다.

그러면 모든 debezium의 모든 토픽이 응답된다.

 

만약 토픽을 삭제하고 싶다면

DELETE http://localhost:8083/connectors/{이름}

 

여기로 DELETE 요청을 보내면 된다.

 

이렇게 내가 만든 이름으로 topic이 생성된 것을 볼 수 있다.

 

이번에는 여기까지만 하고, 다음에는 해당 토픽으로 publish 해보고 subscribe 해보도록 하자.

반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리

Webflux 환경에서 MSA의 Saga, Outbox, CQRS, CDC를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

 

우선 CDC는 change data capture라는 뜻으로, 기존의 outbox는 데이터베이스에서 start 상태를 조회해서 kafka로 publish 했다.

 

이 방법의 문제는 스케줄러에 의해, 해당 시간에만 전송이 이루어지기 때문에 오랜 시간이 지나야 동기화가 진행된다.

이 방법을 해결하기 위해 데이터베이스의 트랜잭션을 감지해, create update delete의 연산이 일어나면 감지해서 kafka로 이벤트를 전송하게 된다.

 

우선 참고한 기술블로그들을 소개하겠다.

https://techblog.woowahan.com/10000/

 

CDC 너두 할 수 있어(feat. B2B 알림 서비스에 Kafka CDC 적용하기) | 우아한형제들 기술블로그

"어 이거 CDC 적용하면 딱이겠는데요? 한번 CDC로 해보면 어때요?" B2B 알림서비스 기획 리뷰 도중 제안받은 의견입니다. 저는 이때까지만 해도 CDC가 무엇인지 잘 모르는 상태였지만, 저 의견 덕분

techblog.woowahan.com

 

우선 굉장히 어렵다...

 

솔직히 블로그를 작성하는 지금까지 완벽하게 성공하지는 못한 것 같다.

그래도 차근차근 글을 작성하며 익혀보도록 하겠다.

 

우선 docker로 debezium 컨테이너를 만들어보자.

 

사용한 debezium의 docker-compose.yml이다.

 

services:
  debezium:
    container_name: debezium
    image: debezium/connect:2.7.3.Final
    ports:
      - "8083:8083"
    environment:
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: "connect-config"
      OFFSET_STORAGE_TOPIC: "connect-offsets"
      STATUS_STORAGE_TOPIC: "connect-status"
      BOOTSTRAP_SERVERS: kafka:29092
      LOGGING_LEVEL: "DEBUG"
      CONNECT_SCHEMA_NAME_ADJUSTMENT_MODE: avro
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
    volumes:
      - "./debezium:/kafka/connect/debezium-connector-schemaregistry-7.2.6"
    networks:
      - seungkyu

networks:
  seungkyu:
    driver: bridge
    name: seungkyu
    external: true

 

네트워크를 kafka와 같은 네트워크로 설정하고, kafka의 주소를 BOOTSTRAP_SERVERS로 지정하면 된다.

 

avro를 사용하기 때문에 schema를 avro로 지정해주고, converter로 avroconverter로 지정을 해준다.

 

그리고 여기서는 debezium-connector-schemaregistry의 버전에 따라 jar 파일들을 넣어주어야 한다.

 

https://debezium.io/documentation/reference/stable/configuration/avro.html

 

Avro Serialization :: Debezium Documentation

Version: |

debezium.io

 

해당 주소에 가져와야 할 파일들이 적혀있다.

 

여기에 해당하는 jar 파일들을 버전에 맞추어 mount한 디렉토리에 저장해주면 된다.

 

또한, 이 중에서도

 

avro와 guava에 해당하는 jar 파일도 maven repository에서 찾아 디렉토리에 넣어주어야 한다.

 

그렇게 하면 일단, 시작!은 할 수 있게 된 것이다.

반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지

Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

 

우선 CQRS 패턴을 적용하기 위해 

https://youtu.be/BnS6343GTkY?si=kRkeJSen4kr-3tO9

 

해당 영상을 참고했다.

 

사실 CQRS가 Command Query Responsibility Segregation로 그냥 단순히 서비스 로직과 쿼리를 분리해서 무슨 이점이 있을까? 라는 생각을 가지고 있었다.

 

하지만 우아한 형제들에서 이런 사용 사례들을 보고 나니 사용하는 이유를 좀 알 수 있을 것 같았다.

 

그리고 내가 이해한 것이 맞다면 아래와 같은 이유로 사용할 것이다.

 

이런 MSA 환경에서 주문을 수행하기 위해서는 고객의 정보를 조회해야 한다.

 

이런 상황에서 동기적으로 동작하기 위해서는 Customer 서버에 Http 요청을 보내야 하지만 이 과정에서 큰 오버헤드가 발생하고, Http 응답이 오기까지 Order 서버에서 block된다는 문제가 발생한다.

그렇다고 Order 서버에서 Customer Database에 접근하면 분리의 원칙을 위반하고 데이터베이스를 조회하는 과정에서 병목이 발생할 가능성이 있다.

 

그렇기에

그렇기에 Customer 서버에서 데이터의 변경요청이 오면 이벤트를 발행해 다른 서버들에게 알려주고, 다른 서버들은 그 중 필수적으로 필요한 부분만 조회가 빠른 데이터베이스에 저장해 조회하면서 사용하는 것이다.

 

그렇게되면 고객 정보의 Create, Update, Delete는 Customer 서버에서 일어나고 고객정보의 Read는 Order 서버에서 일어나게 된다.

이런 것을 CQRS라고 말한다고 생각한다.

 

구현은 Saga, Outbox에 비해 별로 어렵지 않았다.

기존의 데이터는 MongoDB에 저장하고 있었고 Order 서버에서도 고객의 정보를 추가적으로 저장해야 하는데, 내가 사용할 수 있는 조회가 가장 빠른 데이터베이스인 Redis를 사용했다.

 

    override fun createCustomer(createCustomerCommand: CreateCustomerCommand): Mono<CreateCustomerResponse> {
        val customer = Customer(
            id = CustomerId(ObjectId.get()),
            username = createCustomerCommand.username,
            email = createCustomerCommand.email
        )

        val customerCreatedEvent = customerDomainService.validateAndCreateCustomer(customer)

        return customerMessagePublisher.publish(customerCreatedEvent)
            .then(
                customerRepository.save(customer)
            ).thenReturn(CreateCustomerResponse(customer.id.toString(), customer.username, customer.email))
    }

 

이렇게 고객의 생성 명령이 실행되면 데이터베이스에 저장하며, 이벤트를 발행한다.

물론 이 과정 내에서도 Transaction 처리와 Outbox 패턴을 적용해야 하지만, 그 부분은 생략하도록 하겠다.

 

@KafkaListener(id = "\${kafka.consumer.customer-consumer-group-id}",
        topics = ["\${kafka.topic.customer-create}"])
    override fun receive(
        @Payload values: List<CustomerCreateAvroModel>,
        @Header(KafkaHeaders.RECEIVED_KEY) keys: List<String>,
        @Header(KafkaHeaders.RECEIVED_PARTITION) partitions: List<Int>,
        @Header(KafkaHeaders.OFFSET) offsets: List<Long>
    ) {
        values.forEach{
            customerCreateMessageListener.createCredit(
                Credit(
                    customerId = CustomerId(id = ObjectId(it.id)),
                    totalCreditAmount = Money.ZERO
                )
            ).subscribe()
        }
    }

 

이벤트를 수신하는 부분은 이렇게 고객의 id를 받아서 결제 정보를 만들게 된다.

 

여기서 이벤트는 고객의 아이디, 이름, 이메일 모두 발행이 되었지만, 수신하는 부분에서는 고객의 id만 사용하게 된다.

 

결제 과정에서는 고객의 이름, 이메일을 사용하지 않기 때문에 굳이 저장할 필요가 없다.

최소 데이터 보관 원칙에 의해 꼭 필요한 column들만 저장을 하는 것이 좋다.

 

    override fun createCredit(credit: Credit): Mono<Void> {
        return creditRepository.save(credit).then()
    }
    override fun save(credit: Credit): Mono<Credit>{
        return reactiveRedisTemplate.opsForValue().set(
            "${redisPrefix}${credit.customerId.id}",
            objectMapper.writeValueAsString(creditToCreditEntity(credit))
        ).thenReturn(credit)
    }

 

그리고는 조회가 빠른 Redis에 저장을 해두고 주문이 발생하면, Customer 서버에 데이터를 요청하지 않고 빠르게 주문을 수행할 수 있도록 한다.

반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지

Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

 

기존의 Saga 패턴에서 더 강화된 내용이라고 생각하면 될 것이다.

 

Saga 패턴은 긴 트랜잭션을 짧은 트랜잭션으로 나누고 process와 rollback을 사용하여 하나씩 나아가는 구조였다.

 

여기서 이벤트를 전송하고, 변경된 내용을 데이터베이스에 저장하게 되는데 만약 이벤트를 전송하고 데이터베이스에서 에러가 발생해 이벤트만 전송되게 된다면 문제가 발생할 수 있다.

 

이런 문제를 해결하기 위해 사용하는 패턴이다.

방법은 먼저 데이터베이스에 변경사항들을 저장하고, 스케줄러를 사용해 한 번에 이벤트를 전송하는 것이다.

이 때 변경사항들은 기존의 데이터베이스가 아닌, 이벤트를 위한 별도의 저장공간을 만들게 된다.

이곳에 보낼 데이터를 미리 저장해두고 나중에 보내기 때문에 보낼 편지함(Outbox)패턴이라고 불리게 된다.

그리고 모든 사항이 완료된 Outbox 데이터들은 데이터베이스의 최적화를 위해 스케줄러를 사용해서 지속적으로 삭제해준다.

 

이러한 방법을 MSA에서 Outbox 패턴이라고 한다.

 

이렇게 별도의 데이터베이스에 저장해두고 한 번에 보내게되며, 실제 서비스에서는 1~2초의 간격으로 스케줄러를 실행한다고 한다.

 

우선 아래에는 직접 작성한 Outbox 패턴으로 설명해보겠다.

 

@Component
class PaymentOutboxScheduler(
    private val paymentOutboxHelper: PaymentOutboxHelper,
    private val paymentRequestMessagePublisher: PaymentRequestMessagePublisher
): OutboxScheduler {

    private val logger = LoggerFactory.getLogger(PaymentOutboxScheduler::class.java)

    @Transactional
    @Scheduled(fixedDelay = 10000, initialDelay = 10000)
    override fun processOutboxMessages() {
        logger.info("결제를 요청하는 스케줄러가 실행되었습니다.")
        paymentOutboxHelper.getPaymentOutboxMessageByOutboxStatusAndOrderStatus(
            OutboxStatus.STARTED,
            listOf(OrderStatus.PENDING, OrderStatus.CANCELLING)
        ).publishOn(Schedulers.boundedElastic()).map{
            paymentOutboxMessage: PaymentOutboxMessage ->
            if(paymentOutboxMessage.payload.orderStatus == OrderStatus.CANCELLING) {

                paymentOutboxMessage.payload.paymentOrderStatus = PaymentOrderStatus.CANCELLING
            }
            paymentRequestMessagePublisher.publish(
                paymentOutboxMessage = paymentOutboxMessage,
                callback = ::updateOutboxStatus
            ).subscribe()
        }.subscribe()
    }

    private fun updateOutboxStatus(paymentOutboxMessage: PaymentOutboxMessage, outboxStatus: OutboxStatus): Mono<Void> {
        paymentOutboxMessage.outboxStatus = outboxStatus
        return paymentOutboxHelper.save(paymentOutboxMessage).then()
    }
}

 

주문 서버에서 결제 서버로 결제를 요청하는 과정이다.

우선 주문이 발생하면 Outbox의 상태가 Start인 값들만 조회한다.

 

처음에 Outbox에 Start로 저장을 하기 때문에 한번도 전송된 적이 없는 데이터를 불러오는 것이다.

 

그렇게 조회된 모든 데이터를 모두 publisher로 전송을 하며, 전송을 하면 callback을 사용해 Outbox의 Status를 Complete로 변경해준다.

 

publisher의 내용이다.

override fun publish(
        paymentOutboxMessage: PaymentOutboxMessage,
        callback: (PaymentOutboxMessage, OutboxStatus) -> Mono<Void>
    ): Mono<Void> {
        return mono{

            val paymentEventPayload = paymentOutboxMessage.payload

            logger.info("{} 주문에 대한 이벤트 전송을 준비 중입니다.", paymentEventPayload.orderId.toString())

            val paymentRequestAvroModel = paymentEventPayloadToPaymentRequestAvroModel(paymentEventPayload)

            reactiveKafkaProducer.send(
                paymentRequestTopic,
                paymentEventPayload.orderId.toString(),
                paymentRequestAvroModel
            ).publishOn(Schedulers.boundedElastic()).map{
                callback(paymentOutboxMessage, OutboxStatus.COMPLETED).subscribe()
            }.doOnError{
                callback(paymentOutboxMessage, OutboxStatus.FAILED).subscribe()
            }.subscribe()

            logger.info("{}의 주문이 메시지 큐로 결제 요청을 위해 전송되었습니다", paymentEventPayload.orderId.toString())

        }.then()
    }

 

이렇게 해당 데이터를 model로 변환하여 kafka로 전송을 하고, 전송 상태의 여부에 따라 callback을 사용하여 outbox의 상태를 변환한다.

 

결제 서버의 내용은 작성하지 않겠지만, 결제 서버에서도 수신 받은 내용에 따라 내용을 처리하고 kafka로 전송할 데이터를 outbox에 저장해주면 된다.

 

이제 여기서 문제가 생기게 된다.

 

스케줄러를 사용하기 때문에 특정 시간에만 동기화가 이루어지게 되며, 해당 스케줄러가 동작하는 시간에만 CPU의 사용량이 늘어나게 된다.

 

이러한 방법을 해결하기 위해 마지막으로 CDC 패턴을 사용한다고 한다.

CDC와 관련된 내용은 CQRS 다음에 작성해보도록 하겠다.

반응형

프로젝트에서 DeepL API를 사용해 PDF를 번역하는 기능이 추가되었다.

https://developers.deepl.com/docs

 

Introduction | DeepL API Documentation

Learn more about the DeepL API's capabilities and common use cases.

developers.deepl.com

 

우선 DeepL의 공식문서이다.

처음에는 늘 그렇듯 curl을 사용해 파일을 보내고, 응답을 받을 것이라고 생각했지만 자바쪽으로 지원해주는 라이브러리가 있었다.

 

 

자바 라이브러리의 주소는 다음과 같다.

https://github.com/DeepLcom/deepl-java

 

GitHub - DeepLcom/deepl-java: Official Java library for the DeepL language translation API.

Official Java library for the DeepL language translation API. - DeepLcom/deepl-java

github.com

 

요즘에는 자료가 많이 없는 기능들을 사용하다보니, 이렇게 깃허브에 직접 찾아들어가 사용방법을 찾아보는 일이 많아졌다.

 

implementation("com.deepl.api:deepl-java:1.9.0")

 

현재 기준으로 가장 최신 버전인 해당 라이브러리를 추가해준다.

 

처음에는 해당 문서처럼 Translator를 생성하려고 했지만, 해당 기능은 Deprecated 되었다고 한다.

(그러면 문서 좀 수정해주지...)

 

지금은 DeepLClient 클래스를 생성하고, 해당 생성자에 DeepL의 API 키를 넘겨주면 된다.

 

DeepLClient(deepLKey)

 

해당 클래스에서도 translateDocument라는 함수를 사용할 것이다.

 

해당 함수를 오버로딩하고 있는 함수들은 다음과 같다.

    public DocumentStatus translateDocument(File inputFile, File outputFile, @Nullable String sourceLang, String targetLang, @Nullable DocumentTranslationOptions options) throws DocumentTranslationException, IOException {
        try {
            if (outputFile.exists()) {
                throw new IOException("File already exists at output path");
            } else {
                InputStream inputStream = new FileInputStream(inputFile);

                DocumentStatus var8;
                try {
                    OutputStream outputStream = new FileOutputStream(outputFile);

                    try {
                        var8 = this.translateDocument(inputStream, inputFile.getName(), outputStream, sourceLang, targetLang, options);
                    } catch (Throwable var12) {
                        try {
                            outputStream.close();
                        } catch (Throwable var11) {
                            var12.addSuppressed(var11);
                        }

                        throw var12;
                    }

                    outputStream.close();
                } catch (Throwable var13) {
                    try {
                        inputStream.close();
                    } catch (Throwable var10) {
                        var13.addSuppressed(var10);
                    }

                    throw var13;
                }

                inputStream.close();
                return var8;
            }
        } catch (Exception exception) {
            outputFile.delete();
            throw exception;
        }
    }

    public DocumentStatus translateDocument(File inputFile, File outputFile, @Nullable String sourceLang, String targetLang) throws DocumentTranslationException, IOException {
        return this.translateDocument((File)inputFile, (File)outputFile, (String)sourceLang, targetLang, (DocumentTranslationOptions)null);
    }

    public DocumentStatus translateDocument(InputStream inputStream, String fileName, OutputStream outputStream, @Nullable String sourceLang, String targetLang, @Nullable DocumentTranslationOptions options) throws DocumentTranslationException {
        DocumentHandle handle = null;

        try {
            handle = this.translateDocumentUpload(inputStream, fileName, sourceLang, targetLang, options);
            DocumentStatus status = this.translateDocumentWaitUntilDone(handle);
            this.translateDocumentDownload(handle, outputStream);
            return status;
        } catch (Exception exception) {
            throw new DocumentTranslationException("Error occurred during document translation: " + exception.getMessage(), exception, handle);
        }
    }

    public DocumentStatus translateDocument(InputStream inputFile, String fileName, OutputStream outputFile, @Nullable String sourceLang, String targetLang) throws DocumentTranslationException {
        return this.translateDocument(inputFile, fileName, outputFile, sourceLang, targetLang, (DocumentTranslationOptions)null);
    }

 

File을 넘기는 함수가 아닌 inpuStream을 넘기는 함수를 사용할 것이며, 사용 가능한 언어의 종류는 DeepL의 공식문서에 나와있다.

 

여기서 sourceLang은 Null이 가능하다.

현재 문서에 대한 정보를 주지 않아도, 번역을 해보니 sourceLang을 지정해 줄 때와 똑같은 결과가 나왔었다.

 

해당 함수를 사용하면 byteOutputStream이 나오게 된다.

해당 byteOutputStream을 byteArray로 바꾸어서 controller에서 응답해주면 된다.

@Service
class DeepLManager(
    @Value("\${deepL.key}")
    private val deepLKey: String
) {

    fun translateDocument(inputStream: InputStream, fileName: String, targetLang: LanguageEnum): ByteArray {
        ByteArrayOutputStream().use{
            byteArrayOutputStream ->
            DeepLClient(deepLKey).translateDocument(
                inputStream,
                fileName,
                byteArrayOutputStream,
                null,
                targetLang.targetLang,
            )
            return byteArrayOutputStream.toByteArray()
        }
    }

}

 

해당 서비스는 DeepL로 부터 번역된 ByteArray를 가져오는 서비스이고

그냥 응답받은 byteArrayOutputStream에서 toByteArray만 호출하면 ByteArray로 변환된다.

 

이렇게 응답받은 ByteArray을

return translateConnector.createTranslate(
            docsId, createTranslateReq
        ).map{
            ByteArrayOutputStream().use{

            }
            ResponseEntity.ok()
                .contentType(
                    MediaType.APPLICATION_OCTET_STREAM
                )
                .headers{
                    header ->
                    header.contentDisposition = ContentDisposition.builder("pdf")
                        .filename("${docsId}-${createTranslateReq.targetLang.name}.pdf")
                        .build()
                }
                .body(it)
        }

 

이런 식으로 controller에서 응답해주면 된다.

이렇게 하면 지정해준 파일의 이름으로 번역된 파일을 다운받을 수 있다.

 

+ 근데 이게 생각보다 돈이 많이 나오는 거 같다.

이렇게 만들고 몇번 요청을 한 후, PM한테 금액을 확인해달라고 부탁하니 벌써 5만원이 나왔다고 한다...

다들 조심해서 사용하는 게 좋을 것 같다.......

'틔움랩' 카테고리의 다른 글

MockBean deprecated와 대체  (0) 2025.01.28
Github action을 통한 Spring CI/CD 설정  (0) 2025.01.25
반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지

Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

 

MSA에서 제일 중요한 부분 중 하나라고 생각한다.

현재 개발 중인 서버의 구조이다.

 

  1. 클라이언트가 주문을 하면, 주문 서버의 API를 사용해 주문을 하게 된다.
  2. 주문 서버만을 사용해 주문을 완료할 수 없으니, 우선 고객의 잔액을 확인하기 위해 payment 서버로 메시지를 보내게 된다.
  3. 결제 서버는 금액과 잔액을 확인한 후에, 결제가 가능하다면 금액을 뺀 잔액을 저장하고 주문 서버로 응답 메시지를 보내게 된다.
  4. 주문 서버는 결제가 가능하다면, 식당 서버로 메시지를 보내 해당 식당의 상태와 메뉴를 확인한다.
  5. 식당 서버는 해당 메시지를 확인 한 후 그에 맞는 메시지를 주문 서버에게 응답한다.
  6. 해당 메시지를 확인 한 후, 결과를 데이터베이스에 저장한다.

 

이런 과정을 통해 결제가 진행된다.

늘 그렇듯 이런 과정에 transaction 처리를 해야한다.

하지만 다른 서버간에 메시지를 보내는 과정에서 어떻게 transaction 처리를 할 수 있을까?

이럴 때 사용하는 것이 saga 패턴이다.

 

Saga 패턴마이크로서비스 아키텍처에서 분산 트랜잭션을 관리하는 방법 중 하나다.

각 서비스가 개별적으로 트랜잭션을 수행하고, 트랜잭션 간 일관성을 유지하기 위해 보상 작업(rollback)이나 이벤트 체인을 활용하는 방식을 말한다.

 

즉 해당 서버로부터 응답을 받고, 그 결과에 따라 작업을 process 할지 rollback 할지 결정하는 것이다.

 

우선 현재 서비스를 분석해보면, 다음과 같은 실패가 존재한다.

  1. 결제 과정에서 잔액이 부족해 결제를 실패하는 경우
  2. 결제는 성공했지만, 식당에서 문제가 있어 주문이 실패하는 경우

1번은 잔액을 보존하고, 주문 서버로 실패 이벤트만 전송하면 되지만 2번은 아니다.

2번은 식당 서버에서 실패 이벤트를 받으면, 결제 서버로도 실패 이벤트를 전송해 잔액을 복구해야 한다.

 

우선 서비스에 적용을 해보도록 하자.

Saga의 적용을 위해 아래와 같은 인터페이스를 생성한다.

interface SagaStep<T, SuccessEvent: DomainEvent<*>, FailEvent: DomainEvent<*>> {

    fun process(data: T): Mono<SuccessEvent>
    fun rollback(data: T): Mono<FailEvent>
}

여기서 DomainEvent는 common 모듈에서 각각의 이벤트를 위해 상속받아 구현한다.

 

결제와 관련된 Saga는 다음과 같이 구현한다.

@Component
class OrderPaymentSaga(
    private val orderDomainService: OrderDomainService,
    private val orderRepository: OrderRepository
): SagaStep<PaymentResponse, OrderPaidEvent, EmptyEvent> {

    private val logger = LoggerFactory.getLogger(OrderPaymentSaga::class.java)

    override fun process(data: PaymentResponse): Mono<OrderPaidEvent> {
        logger.info("주문 {}의 상태를 결제완료로 변경합니다", data.id)
        return orderRepository.findById(ObjectId(data.id))
            .flatMap{
                val orderPaidEvent = orderDomainService.payOrder(order = it)
                orderRepository.save(it)
                    .thenReturn(orderPaidEvent)
            }.doOnNext{
                logger.info("주문 {}의 상태가 결제완료로 변경되어 저장되었습니다.", it.order.orderId.id)
            }
    }

    override fun rollback(data: PaymentResponse): Mono<EmptyEvent> {
        logger.info("주문 {}의 상태를 취소로 변경합니다.", data.id)
        return orderRepository.findById(ObjectId(data.id))
            .flatMap {
                orderDomainService.cancelOrder(order = it)
                orderRepository.save(it)
                    .thenReturn(EmptyEvent())
            }.doOnNext{
                logger.info("주문 {}의 상태가 취소로 변경되어 저장되었습니다.", data.id)
            }
    }
}

 

주문이 실패한다면, 해당 주문만 실패로 데이터베이스에 저장해준다.

주문이 성공한다면, 해당 주문을 결제성공으로 데이터베이스에 저장하고 식당 서버로 승인 요청 이벤트를 전송한다.

 

승인과 관련된 Saga이다.

@Component
class RestaurantApprovalSaga(
    private val orderDomainService: OrderDomainService,
    private val orderRepository: OrderRepository,
    private val orderCancelledPaymentRequestMessagePublisher: OrderCancelledPaymentRequestMessagePublisher
): SagaStep<RestaurantApprovalResponse, EmptyEvent, OrderCancelledEvent> {

    private val logger = LoggerFactory.getLogger(RestaurantApprovalSaga::class.java)

    @Transactional
    override fun process(data: RestaurantApprovalResponse): Mono<EmptyEvent> {
        logger.info("{} 주문이 승인 완료되었습니다", data.id)

        return orderRepository.findById(ObjectId(data.id))
            .flatMap {
                orderDomainService.approveOrder(order = it)
                logger.info("바뀐거: $it")
                orderRepository.save(it)
                    .thenReturn(EmptyEvent())
            }.doOnNext{
                logger.info("{} 주문이 승인 완료되어 저장되었습니다", data.id)
            }
    }

    @Transactional
    override fun rollback(data: RestaurantApprovalResponse): Mono<OrderCancelledEvent> {
        logger.info("{} 주문이 미승인 되었습니다", data.id)
        return orderRepository.findById(ObjectId(data.id))
            .flatMap {
                val orderCancelledEvent = orderDomainService.cancelOrderPayment(it)
                orderRepository.save(it)
                    .thenReturn(orderCancelledEvent)
            }.doOnNext{
                logger.info("{} 주문이 취소 중 상태로 저장되었습니다", it.order.orderId.id.toString())
            }.doOnNext{
                orderCancelledPaymentRequestMessagePublisher.publish(it)
            }.doOnNext {
                logger.info("{} 주문의 결제 취소 이벤트를 전송했습니다.", it.order.orderId.id.toString())
            }
    }
}

 

주문이 승인되었다면, 해당 주문만 승인 완료로 데이터베이스에 저장해주면 된다. (이미 결제 서버에서 결제는 완료되었기 때문에)

주문이 실패했다면, 결제 서버의 잔액을 복구시켜야 하기 때문에 결제서버로 실패 이벤트를 전송해준다.

 

이런 식으로 메시지를 받을 때마다, process 할건지 rollback 할건지 결정해서 다음 프로세스를 진행하면 된다.

그리고 당연히 메시지를 받고 보내는 과정까지의 해당 서버는 transaction 하게 동작해야 한다.

 

enum class OrderStatus {
    PENDING, PAID, APPROVED, CANCELLING, CANCELLED
}

 

주문의 상태는 다음과 같다.

PENDING: 주문이 막 생성된 상태

PAID: 결제가 완료된 상태 (Order -> Payment -> Order로 성공 이벤트를 응답받음)

APPROVED: 주문이 승인된 상태 (Order -> Payment -> Order -> Restaurant -> Order로 성공 이벤트를 응답받음)

CANCELLING: 승인이 취소된 상태 (Order -> Payment -> Order -> Restaurant -> Order로 실패 이벤트를 응답받음)

CANCELLED: 모든 취소가 완료된 상태

 

해당 상태들을 데이터베이스에 저장해가며, 장기 transaction을 메시지를 응답받고 보내는 과정으로 분해하여 transaction을 적용하며 saga패턴을 적용하면 된다.

 

해당 서버 내에서 transaction을 적용하는 것은 어렵지 않겠지만, 이런 프로세스를 이해 할 수 있도록 saga 패턴을 적용할 서버를 제대로 분석하는 것이 필요하다고 생각한다.

'MSA' 카테고리의 다른 글

MSA에 CQRS 패턴 적용하기  (0) 2025.03.01
MSA에 Outbox 패턴 적용하기  (0) 2025.02.28
DDD에서 Hexagonal Architecture로 변경하기  (0) 2025.02.20
Spring + Kafka에서 avro 사용하기  (1) 2025.02.15
DDD의 핵심 요소  (0) 2025.02.14
반응형

https://github.com/Seungkyu-Han/micro_service_webflux

 

GitHub - Seungkyu-Han/micro_service_webflux: Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지

Webflux 환경에서 MSA의 Saga, Outbox, CQRS를 연습해보기 위한 리포지토리입니다. - Seungkyu-Han/micro_service_webflux

github.com

기존에 사용하던 Architecture는 보통 DDD였다.

물론 DDD도 굉장히 좋은 설계이지만, 다음과 같은 문제가 있었다.

 

 

보통 이렇게 개발이 되는데, 여기서 가장 핵심인 부분은 Domain Layer이다.

다른 Layer에 영향을 최대한 받지 않고, 독립적으로 존재할 수 있어야 가장 좋을 것이다.

하지만 지금은 Data Layer에 종속이 되어 있으며, Data Layer가 변경이 될 때마다 Domain Layer도 수정이 되어야 한다는 문제가 생긴다.

 

그렇기 때문에 이런 구조를 다음과 같이 변경해보려고 한다.

Domain Layer에서 사용할 Data Layer의 인터페이스만 만들어두고, 해당 인터페이스는 Data Layer에서 하는 것이다.

Domain Layer에서 사용하는 인터페이스를 포트라고 하고, 해당 인터페이스를 구현하는 객체를 어뎁터라고 한다.

 

이렇게 Domain Layer에서 포트를 만들고, Data Layer에서 어뎁터로 연결하는 방식을 사용하여 Domain Layer의 독립성을 높이는 방법이 Hexagonal Architecture이다.

 

 

이런 방법으로 다른 Layer들이 Domain Layer에 의존하도록 한다.

 

예를 들어보자면, 결제와 관련된 서비스를 개발하고 있다.

 

여기서 가장 독립적인 모듈은 domain인 payment-domain이다.

이런 식으로 포트를 만든다.

여기서 포트는 입력으로 들어오는 포트와 나는 출력 포트가 있다.

 

입력으로 들어오는 포트는 domain layer에서 기존처럼 개발 할 수 있다.

어차피 상위 layer에서 해당 포트를 사용하는 구조이기 때문이다.

 

출력으로 나가는 포트는 여기서 구현하는 것이 아닌, 인터페이스만 만들어두고 해당 인터페이스를 사용해서 domain layer를 개발한다.

이런 식으로 인터페이스만 만들어 사용한다.

 

해당 인터페이스의 구현은 외부 persistence 모듈이다.

 

이런 식으로 어뎁터 클래스를 만들고

dependency injection을 통해 외부에서 구현한 repository를 주입해준다.

message와 관련한 kafka도 데이터베이스와 같다.

 

이런 식으로 설계한다면, domain layer는 어떤 모듈도 의존하지 않는 가장 독립적인 상태로 개발이 가능하다.

'MSA' 카테고리의 다른 글

MSA에 CQRS 패턴 적용하기  (0) 2025.03.01
MSA에 Outbox 패턴 적용하기  (0) 2025.02.28
MSA에 SAGA 패턴 적용하기  (0) 2025.02.24
Spring + Kafka에서 avro 사용하기  (1) 2025.02.15
DDD의 핵심 요소  (0) 2025.02.14

+ Recent posts