반응형 프로그래밍 with RxJS

테이블링 백엔드 팀은 2023년부터 새로운 기술이나 도입해보고 싶은 기술, 개발 중 경험한 노하우나 방법을 바탕으로 발표하는 시간을 갖기로했습니다. 안녕하세요. 백엔드 팀신종민입니다. 개발하면서 “비동기 작업의 복잡도 비용을 어떻게 낮출 수 있을까”하는 고민에서 반응형 프로그래밍에 관심을 갖게 되었고, 그 과정에서 학습한 RxJS라는 라이브러리에 대해서 간략하게 소개해보려합니다. 반응형 프로그래밍 “반응형 프로그래밍은 데이터의 흐름과 변경 사항을 전파하는 데에 중점을 둔 프로그래밍 패러다임입니다.” 데이터 흐름과 상태 변화에 초점을 맞추어 비동기 작업을 간결하게 표현하고 관리할 수있습니다. 데이터의 변화를 감지하고 자동으로처리합니다. RxJS Promise는 비동기 코드를 처리하는 데 사용할 수 있지만, 반복되는 키 입력이나 마우스 움직임과 같은 많은 양의 데이터 스트림을 다루는 경우 제약이 많이존재합니다. RxJS는 이런 데이터 스트림을 Observable과 Operator를 조합하여 변형하고 조작할 수 있습니다. 이에 따라 코드의 가독성과 유지보수성이 향상되고, 복잡한 비동기 동작을 직관적으로 표현할 수있습니다. 또한, Operator의 다양한 조합 가능하여 데이터 필터링, 변환, 병합, 그룹화 등 다양한 작업을 편리하게 처리할 수 있습니다. 비동기 코드를 처리하는 데 매우 강력하고 유연한 도구이며, 다양한 시나리오에서 활용할 수있습니다. Observable 개발자들은 다양한 상황에서 다양한 고민을 합니다. 특히 입력값에 따라 동기 또는 비동기 처리, 함수 호출, 이벤트, 콜백, 프라미스 등 어떤 방식으로 처리해야 할지 결정해야 합니다. 이러한 다양한 처리 방식을 추상화하여 일관된 인터페이스로 상호작용할 수 있는 방법은없을까요? 바로 Observable이라는 데이터 스트림 객체를 사용하면가능합니다. 기존 방식에서는 입력값에 따라 각각 다른 처리 방식을 구현해야 했지만, 모두 Observable 객체로 추상화하여 처리함으로써 Observable을 다루는 부분에 집중할 수 있게됩니다. RxJS 공식 문서에서는 Observable을 “시간에 따라 변화하는 값들의 컬렉션”으로설명합니다 즉, Observable은 일련의 값을 방출하는 컬렉션으로 생각할 수 있습니다. 이러한 값들은 시간에 따라 동적으로 생성되거나 도착할 수 있으며, 구독자는 Observable을 구독하여 값을 소비하고 반응할 수있습니다. Observable은 언제든지 구독, 취소 또는 완료할 수있습니다. Observable의 Lazy execution은 더 선언적이고 효율적인 코드를 작성할 수있습니다. Observable에 대한 다양한 메서드가지원됩니다. 모든 유형의 비동기 작업 및 Side effect를 처리하기 위한 단일 인터페이스입니다. Observable을 통해 연속적인 데이터를 다루고, 비동기 작업을 더 세밀하게 제어할 수 있으며, 취소도 할 수 있습니다. 게다가 Lazy execution을 통해 불필요한 작업을 방지하고, 다양한 Operator를 통해 다양한 데이터 처리 작업을 간편하게 수행할 수있습니다. Reactive 비동기 코드의 복잡성과 시간에 대한 처리는 프로그래밍에서 항상 어려운 문제 중 하나입니다. 시간은 불확실하며, 비동기 작업은 예측하기 어렵기 때문에 일반적인 동기 코드와는 다른 접근 방식이필요합니다. RxJS는 이러한 문제를 해결하기 위해 시간과 작업에 대한 반응적인 프로그래밍을 강조합니다. 시간에 대한 처리를 직접적으로 다루거나 작업이 언제 완료될지 추측하지 않고, 작업에 반응하여 처리하는방식입니다. Observable을 사용하면 데이터 스트림이 시간에 따라 지속적으로 발생하는 상황을 다룰 수 있습니다. 이를 통해 비동기 작업의 진행 상황을 실시간으로 관찰하고, 필요한 시점에 적절한 조치를 취할 수있습니다. 예를 들어, Observable을 사용하여 사용자 입력, 네트워크 요청, 타이머 등의 이벤트를 처리하면 작업의 진행과 완료를 실시간으로 감지하고 원하는 작업을 수행할 수있습니다. 이러한 반응적인 프로그래밍 접근 방식은 코드의 예측 가능성과 유연성을 높여줍니다. 시간에 대한 처리를 추상화하고, 비동기 작업을 명확하고 일관된 방식으로 다룰 수 있기 때문에 코드의 가독성과 유지보수성이향상됩니다. 또한, RxJS의 다양한 Operator를 활용하여 시간과 작업을 조작하고 제어할 수 있으며, 코드의 재사용성을 높이고 복잡성을 줄일 수있습니다. 따라서 RxJS를 사용하면 시간에 대한 프로그래밍을 보다 효과적으로 다룰 수 있으며, 비동기 코드의 복잡성을 줄이고 예측 불가능한 상황에서도 유연하게 대응할 수있습니다. RxJS를 활용한 데이터 처리구성 - Observable: 스트림으로부터 비동기적인 데이터를 읽어오는객체입니다. - Observer: Observable에서 발생한 이벤트를 처리하는객체입니다. - Operator: Observable에서 발행한 데이터를 변환하거나, 다른 Observable로 전송하는연산자입니다. - Subscription: Observable과 Observer 간의 상호작용을 설정하는 구독객체입니다. import ( of ) from 'rxjs' import ( map ) from 'rxjs/operators' // 1. 데이터 소스를 Observable로 생성 const dataSource = of(1, 2, 3, 4, 5) // 2. 연산자를 사용하여 데이터 변경 const modifiedStream = dataSource.pipe( map((value: number) => value * 2), ) // 3. Observer 생성 const observer = ( next: (value: number) => console.log('Received:', value), error: (error: any) => console.error('Error:', error), complete: () => console.log('Complete'), ) // 4. Observable 구독 등록 const subscription = modifiedStream.subscribe(observer) // 5. 일정 시간 후에 구독 정지 및 자원 해지 setTimeout(() => ( subscription.unsubscribe() ), 5000) 데이터 소스를 Observable로 생성한다. Operator를 사용해서 데이터를 변경, 추출, 합침,분리한다. 원하는 데이터를 받아 처리하는 Observer를 만든다. Observable의 subscribe를 통해 Observer를 등록한다. Observable 구독을 정지하고 자원을해지한다. 코드 예제 import ( from, interval ) from 'rxjs' import ( map, filter, take, mergeMap ) from 'rxjs/operators' // 가정: 외부 API로부터 실시간 주식 가격을 스트림으로 받아온다고 가정합니다. function fetchStockPrice(symbol) ( return interval(1000).pipe( map(() => ( const price = Math.random() * 100 // 가상의 주식 가격 생성 return ( symbol, price ) )) ) ) // 주식 가격 스트림을 처리하는 예시 from(['AAPL', 'GOOGL', 'MSFT', 'TSLA']) // 주식 심볼 목록 .pipe( mergeMap(symbol => fetchStockPrice(symbol)), // 각 심볼에 대한 주식 가격 스트림 생성 filter(priceData => priceData.price > 50), // 가격이 50보다 큰 데이터 필터링 take(5) // 처음 5개의 데이터만 가져오기 ) .subscribe(priceData => ( console.log(priceData) )) 위 코드는 주식 가격을 실시간으로 스트림으로 받아와 처리하는예시입니다. fetchStockPrice 함수는 외부 API를 호출하여 주식 가격을 비동기적으로 가져오는 작업을 수행합니다. 여기에서는 간단한 예시로서 interval을 사용하여 가상의 가격을생성합니다. from(['AAPL', 'GOOGL', 'MSFT'])는 배열에 포함된 주식 심볼들을 나타내는 Observable을 생성합니다. 이것은 주식 심볼 목록을 스트림으로만들어줍니다. mergeMap 연산자는 각 주식 심볼에 대해 fetchStockPrice 함수를 호출하여 주식 가격 스트림을 생성합니다. 각 심볼에 대한 스트림들은 병합되어 하나의 스트림으로처리됩니다. filter 연산자는 주식 가격이 50보다 큰 데이터만 필터링하여통과시킵니다. take(5) 연산자는 처음 5개의 데이터만 가져오도록제한합니다. 마지막으로, subscribe 함수를 호출하여 Observable을 구독하고, 각 주식 가격 데이터를 처리하는 로직을 구현합니다. 이 예시에서는 간단히 가격 데이터를 콘솔에출력합니다. Operators RxJS는 적절한 오퍼레이터를 찾고 사용하는 방법을 익히는 데 시간이필요합니다. https://rxviz.com/ https://reactive.how/ 위 링크를 통해 오퍼레이터 체인의 흐름과 데이터 변환을 실시간으로 시각화하여 살펴볼 수 있습니다. 다양한 예제와 함께 동작을 시뮬레이션하고, 데이터 흐름을 보면서 RxJS의 작동 방식을 더 잘 이해하실 수 있을겁니다. 마치며 선언적인 프로그래밍이가능했습니다. 데이터 처리 로직이 단순해져서 재사용성이높아졌습니다. 비동기 작업을 일관적으로 처리할 수있었습니다. catchError, retry 등의 연산자를 사용하여 예외 상황을 처리하고, 에러 핸들링을 쉽게 구현할 수있었습니다. RxJS 라이브러리를 사용해보면서 느낀 것은 프로그래밍은 Side effect를 올바르게 다루는 것이 핵심이라는 점이었습니다. 그래서 프로그래머가 문제와 요구사항을 고려하여 적절한 패러다임을 선택하고 필요에 따라 여러 패러다임을 조합하는 것이 중요할것입니다. ReactiveX는 다양한 언어를 지원하며 여러 플랫폼에서 사용하실 수 있습니다. 관심이 있다면 실습해 보시기를 추천해드립니다. 감사합니다! 참고 https://en.wikipedia.org/wiki/Reactive_programming https://rxjs.dev/ https://www.infoq.com/articles/reactive-programming-rxjs/ 반응형 프로그래밍 with RxJS was originally published in 테이블링 기술블로그 on Medium, where people are continuing the conversation by highlighting and responding to this story.