Versions Compared

Old Version 39

changes.mady.by.user PSMON

Saved on

compared with

New Version Current

changes.mady.by.user PSMON

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

인프라 활용계층과 도메인 코드 분리전략

...

Kafka를 포함 S3 작동 Flow가 짧은 코드로 가능한이유는 짧고 강력한 이유는~ 인프라 활용 계층을 계층의 코드와 도메인 로직과 분리가 되었기 때문이며때문입니다.

alphaka는 주변 스택을 잘 이해하고 일괄된 방식으로 인프라 계층의 사용전략을 설정화로 설정화로 다룹니다.

네트워크를 이용하는 스택에서 네트워크 순단은 발생할수 있기때문에 재시도 정책이 설정으로 분리되어 기능을 제공해줍니다.

...

Code Block
   retry-settings {
    max-retries = 3
    min-backoff = 200ms
    max-backoff = 10s
    random-factor = 0.0
  }


만약 위와같이 단절상황에 재시도 하는방법을 코드로 정의 한다고하면

특정 예외 상황을 이해하고 재시도하는 로직을 함께 작성할수 있으며 결과적으로 이러한 코드가 늘어나게되면

도메인 로직이 인프라 예외처리코드에 모두 숨겨져 도메인이 작동하는 방식 자체를 이해하기 어렵게 만들수 있습니다.



이 유닛테스트에 사용된 인프라계층의 설정값

Code Block
themeEmacs
titletest.conf
# Properties for akka.kafka.ProducerSettings can be
# defined in this section or a configuration section with
# the same layout.
akka.kafka.producer {
  # Config path of Akka Discovery method
  # "akka.discovery" to use the Akka Discovery method configured for the ActorSystem
  discovery-method = akka.discovery

  # Set a service name for use with Akka Discovery
  # https://doc.akka.io/docs/alpakka-kafka/current/discovery.html
  service-name = ""

  # Timeout for getting a reply from the discovery-method lookup
  resolve-timeout = 3 seconds

  # Tuning parameter of how many sends that can run in parallel.
  # In 2.0.0: changed the default from 100 to 10000
  parallelism = 10000

  # Duration to wait for `KafkaProducer.close` to finish.
  close-timeout = 60s

  # Call `KafkaProducer.close` when the stream is shutdown. This is important to override to false
  # when the producer instance is shared across multiple producer stages.
  close-on-producer-stop = true

  # Fully qualified config path which holds the dispatcher configuration
  # to be used by the producer stages. Some blocking may occur.
  # When this value is empty, the dispatcher configured for the stream
  # will be used.
  use-dispatcher = "akka.kafka.default-dispatcher"

  # The time interval to commit a transaction when using the `Transactional.sink` or `Transactional.flow`
  # for exactly-once-semantics processing.
  eos-commit-interval = 100ms

  # Properties defined by org.apache.kafka.clients.producer.ProducerConfig
  # can be defined in this configuration section.
  kafka-clients {
  }
}

akka.kafka.consumer {

  enable.auto.commit = true

  kafka-clients {
    bootstrap.servers = "localhost:9092"
  }
}

akka.kafka.committer {

  # Maximum number of messages in a single commit batch
  max-batch = 1000

  # Maximum interval between commits
  max-interval = 3s

  # Parallelsim for async committing
  parallelism = 100

  # API may change.
  # Delivery of commits to the internal actor
  # WaitForAck: Expect replies for commits, and backpressure the stream if replies do not arrive.
  # SendAndForget: Send off commits to the internal actor without expecting replies (experimental feature since 1.1)
  delivery = WaitForAck

  # API may change.
  # Controls when a `Committable` message is queued to be committed.
  # OffsetFirstObserved: When the offset of a message has been successfully produced.
  # NextOffsetObserved: When the next offset is observed.
  when = OffsetFirstObserved
}

# 이 설정은 유닛테스트를 위한 LocalStack 호환버전이며, AWS-S3이용시 호환은 아래링크 참고 호환을 시킵니다.
# link : https://github.com/akka/alpakka/blob/main/s3/src/main/resources/reference.conf
alpakka.s3 {
  buffer = "memory"
  disk-buffer-path = ""

  # default values for AWS configuration
  aws {
    credentials {
      provider = static
      access-key-id = "test"
      secret-access-key = "test"
    }

    region {
      provider = static
      default-region = "us-east-1"
    }
  }

  path-style-access = true
  access-style = virtual
  endpoint-url = "http://localhost:4567"
  list-bucket-api-version = 2
  validate-object-key = true

  retry-settings {
    max-retries = 3
    min-backoff = 200ms
    max-backoff = 10s
    random-factor = 0.0
  }

  multipart-upload {
    retry-settings = ${alpakka.s3.retry-settings}
  }

  sign-anonymous-requests = true
}

...


추가 도전과제

이 와같은 함수형 Stream방식의 개발방법은 다양한 언어에서도 활용할수 있는 컨셉으로 꼭 AkkaStream이 아니여도 됩니다.

오히려 자신이 가진 기본프레임워크가 강력한 기능을 이미 가지고 있다고 하면 기본 StreamAPI를 이용할것을 권장합니다.



  • 자바진영이라고 하면  기본으로

...

  • 제공 동시성 병렬처리의 개발방법이 만족스럽지 못할수 있으며 이때 Webplux/rx.Java와같이  ReactiveStream을 준수하는것을 채택해서 더 강력해질수 있습니다.
  • 닷넷진영의 경우 기본언어 스펙이 오랫동안 제공해온  Linq/TPL/Await 조합이 이미 강력하고 심플하며 대부분 이것만으로 해결되기 때문에 큰 관심사가 아닐수 있습니다.

...


Code Block
themeEmacs
Erik Meijer의 한마디~

"이보게, 브라이언 괴츠, C#,파이선,자바스크립트는 물론 심지어 PHP도 async, await를 지원하고 있다네. 그런 기능이 없는 언어는 자바일뿐이야.

람다를 이용해서 콜백함수를 사용하면 된다고? 천만에 콜백은 최악이야. 도움이 안된다고. 자바 9 버전에 담으려고 하는 걸 다 내려놓고 지금당장

asymc, await부터 넣으라고. 그래야 모두가 행복해질수 있어"

...