[Dart::Stream] Stream을 비동기처리 한다는 게 무슨 뜻일까? with Bloc

Question 1.

근데 나 궁금한 거 있는데 스트림을 비동기로 처리한다는 게 제대로 이해가 안가. 비동기는 어떤 함수나 로직이 수행이 완료될 때 까지 기다린다는 건데 스트림은 어떤 흐름이 시작되면 그것을 구독하고 이벤트가 일어나면 구독한 콜백이 실행되는 거잖아. 완전 다른 개념 아니야?

Answer 1.

좋은 질문이다. 비동기 처리와 스트림의 개념이 다소 헷갈릴 수 있지만, 이 둘은 밀접하게 연관되어 있다. 비동기 처리와 스트림은 모두 "작업을 기다리거나 블로킹하지 않고 처리"하는 방식에서 출발한다. 하지만 그 작동 방식과 사용 목적에는 차이가 있다.

비동기

비동기 처리의 핵심은 작업이 완료될 때까지 다른 작업을 방해하지 않고 진행할 수 있다는 것이다. 즉, 비동기 작업은 결과를 즉시 반환하지 않고, 작업이 완료되면 그 결과를 나중에 받게 된다.

Stream

Stream은 시간에 따라 일어나는 여러 비동기 이벤트들을 처리하는 것을 의미한다. 즉, 스트림은 일련의 데이터 이벤트나 에러 이벤트 그리고 완료 이벤트를 연속적으로 발생시키고, 이를 구독하는 방식으로 처리한다.

스트림은 비동기적으로 작동한다. 이벤트가 발생할 때마다 그 이벤트를 처리할 준비가 되어있고, 이벤트가 언제 발생할지는 미리 알 수 없다. Stream은 한 번 실행 후 결과를 반환하는 Future와 달리, 여러 번에 걸쳐 발생하는 데이터를 비동기적으로 처리하는 것이 목적이다.

비동기 예시

비동기 함수(Future)는 단일 결과를 처리한다. 비동기적으로 작업을 수행하고, 그 결과가 준비되면 반환한다. 그 과정에서 다른 작업이 중단되지 않는다.

// Future 비동기 처리 예시
Future<void> fetchData() async {
  await Future.delayed(Duration(seconds: 2));
  print('Data fetched!');
}

void main() {
  print('Start fetching data');
  fetchData();
  print('Doing other things...');
}

여기서 fetchData는 2초 후에 데이터를 가져오지만, 그동안 Doing other things...는 즉시 실행된다. 이것이 비동기의 기본 개념이다.

"어? 2초뒤에 Data fetched! 를 출력하고 Doing other things... 를 출력해야 하는 거 아닌가? 외부에서 데이터를 가져올 때 데이터를 가져오지 않았는데 아래 코드가 실행되면 안되잖아!"
그런 상황은 아래의 코드와 같이 사용하면 된다.

// Future 비동기 처리 예시
Future<void> fetchData() async {
  await Future.delayed(Duration(seconds: 2));
  print('Data fetched!');
}

void main() async {
  print('Start fetching data');
  await fetchData();
  print('Doing other things...');
}

fetchData 메소드 자체를 await 해야한다.

Stream의 비동기 처리

Stream은 여러 개의 이벤트를 처리한다. 스트림은 데이터가 시간에 따라 계속해서 발생하고 이것을 처리해야 한다. 스트림이 비동기적인 이유는 언제 데이터가 발생할지 알 수 없기 때문에, 데이터를 기다리거나 미리 준비된 이벤트를 처리할 수 있도록 해야 하기 때문이다.

// Stream 비동기 처리 예시
Stream<int> numberStream() async* {
  for (int i = 0; i < 5; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i;  // 데이터를 방출
  }
}

void main() async {
  print('Start listening to the stream');
  await for (int number in numberStream()) {
    print('Received: $number');
  }
  print('Stream ended');
}

• 여기서 numberStream은 1초마다 값을 방출하며, 이벤트가 발생할 때마다 구독자는 그 데이터를 받는다.
• 스트림은 한 번에 하나의 데이터를 보내고, 구독자는 이벤트가 발생할 때만 그 데이터를 처리한다.
• 이 스트림은 언제든지 새로운 데이터를 방출할 수 있으며, 그 데이터가 발생할 때까지 프로그램은 다른 작업을 계속 수행할 수 있다.

왜 Stream도 비동기적인가?

스트림도 비동기적인 이유는 데이터가 언제 도착할지 모른다는 점이다. 스트림은 "데이터가 준비되면 처리하는 방식"으로 동작하며, 데이터가 발생하지 않을 때는 다른 작업을 방해하지 않는다.

네트워크 요청의 응답, 사용자 입력, 타이머 이벤트 등 언제 발생할지 모르는 일련의 이벤트를 처리할 때 스트림이 유용하다. 이 이벤트들은 비동기적으로 발생하며, 즉시 처리되거나 준비될 때까지 기다릴 필요가 없다. 콜백과 유사한 방식으로 스트림은 구독자가 이벤트를 기다리며 비동기적으로 처리하므로, 콜백 함수를 사용하여 이벤트가 발생할 때 이를 처리하는 방식과 유사하다.

비동기와 Stream의 결론적인 차이

비동기 작업(Future)와 스트림(Stream)은 비슷하게 "작업을 기다리지 않고 처리한다"는 비동기적 특성을 공유하지만, 스트림은 다중 이벤트를 처리하는 것이 주 목적이고, 비동기 함수(Future)는 한 번의 결과를 처리하는 데 집중한다.

비동기라는 큰 틀에서 보면 둘 다 비동기적인 방식으로 데이터를 처리하지만, 스트림은 비동기적인 이벤트의 흐름을 구독하고 처리하는 방식이라서 비동기의 일종이라고 볼 수 있다.

Question 2.

// Stream 비동기 처리 예시
Stream<int> numberStream() async* {
  for (int i = 0; i < 5; i++) {
    await Future.delayed(Duration(seconds: 10));
    yield i;  // 데이터를 방출
  }
}

void main() async {
  print('Start listening to the stream');
  await for (int number in numberStream()) {
    print('Receiveda: $number');
  }
  print('Stream ended');
}

여기서는 Duration.delayed를 이벤트라고 할 수 있는 건가?

Answer 2.

맞다. Duration.delayed를 사용한 부분도 이벤트의 발생을 시뮬레이션한다고 볼 수 있다. 다만, 여기서 Duration.delayed는 실제 외부 이벤트가 아닌, 코드에서 인위적으로 시간 지연을 추가하여 이벤트가 발생하는 것처럼 처리하는 방식으로 예를 든 것이다.

• 실제 이벤트: 네트워크 요청이 완료되거나, 사용자가 버튼을 클릭하는 등의 외부에서 발생하는 비동기 이벤트.
• 시뮬레이션된 이벤트: 이 예제에서처럼 Future.delayed를 사용해 특정 시간이 지나면 발생하는 이벤트처럼 처리하는 경우.

Question 3. (깨달음)

아 그러면 bloc 상태 관리에는 어떤 위젯이 빌드될 때 그 위젯이 언제 종료될 지 모르니 해당 위젯이 켜질 때 비동기 Stream을 하고 특정 이벤트가 일어날 때 방출하는 형식이라서 필요한 모든 상태를 함께 스트림 등록을 하는 거구나!

Answer 3.

정확히 이해했다.

BLoC(Business Logic Component) 패턴에서 상태 관리는 비동기 Stream을 활용하여 이루어진다. BLoC 패턴의 핵심 아이디어는 UI와 비즈니스 로직을 완전히 분리하고, 상태 관리를 Stream을 통해 비동기적으로 처리하는 것이다. 이를 통해 애플리케이션의 상태가 언제든지 변화할 수 있는 상황에서 유연하게 대처할 수 있다.

BLoC에서는 상태(state)와 이벤트(event)를 관리하기 위해 Stream을 사용한다. Stream은 비동기적으로 상태나 이벤트를 처리하고, 구독(subscribe)하는 Widget은 그 변화에 반응한다.

어떤 위젯이 빌드될 때, 그 위젯이 언제 종료될지 미리 알 수 없다. 따라서 위젯이 활성화될 때(initState나 build 메소드에서) BLoC의 Stream에 구독을 시작한다.

특정 이벤트가 발생하면, BLoC은 이를 처리하고 상태를 업데이트한 후, 새로운 상태를 Stream을 통해 방출한다. 이후, 구독 중인 위젯은 이 새로운 상태를 받아 UI를 업데이트한다.

BLoC은 여러 상태를 관리해야 한다. 예를 들어, 로딩 중 상태, 데이터가 로드된 상태, 에러 상태 등이 있을 수 있다. 이 모든 상태를 하나의 Stream에서 관리하고, 특정 상태 변화가 발생할 때마다 이를 yield하여 스트림을 통해 방출한다. 구독하는 Widget은 이 상태 변화를 감지하고, 해당 상태에 맞게 UI를 업데이트한다.

// 의사 코드
 class MyBloc {
   final _stateController = StreamController<MyState>();

   Stream<MyState> get state => _stateController.stream;

   void fetchData() async {
     _stateController.add(LoadingState());
     try {
       final data = await fetchFromNetwork();  // 비동기 작업
       _stateController.add(DataLoadedState(data));
     } catch (e) {
       _stateController.add(ErrorState(e.toString()));
     }
   }

   void dispose() {
     _stateController.close();
   }
 }

 class MyWidget extends StatelessWidget {
   final MyBloc bloc = MyBloc();

   @override
   Widget build(BuildContext context) {
     return StreamBuilder<MyState>(
       stream: bloc.state,
       builder: (context, snapshot) {
         if (snapshot.data is LoadingState) {
           return CircularProgressIndicator();
         } else if (snapshot.data is DataLoadedState) {
           final data = (snapshot.data as DataLoadedState).data;
           return Text('Data: $data');
         } else if (snapshot.data is ErrorState) {
           return Text('Error: ${(snapshot.data as ErrorState).message}');
         }
         return Container();
       },
     );
   }
 }