Dart 기본기 - Stream, await, async, yield

Stream

비동기 데이터의 연속적인 흐름을 처리하는 객체로, 여러 개의 데이터를 순차적으로 전달할 때 사용됩니다. Future가 한 번만 결과를 반환하는 비동기 작업이라면, Stream은 여러 개의 데이터를 지속적으로 받을 수 있는 비동기 방식입니다.

 

---

1. Stream의 주요 특징

• 비동기 데이터 스트리밍: 여러 개의 데이터를 순차적으로 전달하며, 데이터가 들어오는 대로 처리 가능.
• 리얼타임 데이터 처리: WebSocket, 센서 데이터, 네트워크 스트리밍 등 실시간 이벤트 처리에 적합.
• 구독(Subscription) 방식: .listen()을 사용하여 데이터가 들어올 때마다 반응할 수 있음.

---

2. Stream vs Future 차이점

기능	Future	Stream
반환 값 개수	단 한 번	여러 개
비동기 처리 방식	한 번 완료되면 끝남	지속적으로 데이터가 들어옴
사용 예시	API 요청, 파일 읽기	WebSocket, 실시간 센서 데이터

Future<int> getFutureData() async {
  await Future.delayed(Duration(seconds: 2));
  return 42; // 한 번만 반환됨
}

Stream<int> getStreamData() async* {
  for (int i = 1; i <= 5; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i; // 여러 번 반환됨
  }
}

 

---

3. Stream 생성 및 사용법

(1) 단순한 Stream 생성

Stream<int> numberStream() async* {
  for (int i = 1; i <= 5; i++) {
    await Future.delayed(Duration(seconds: 1)); // 1초 대기 후 데이터 전송
    yield i; // 스트림에 값 추가
  }
}

void main() {
  numberStream().listen((data) {
    print("Received: $data");
  });
}


// Received: 1
// Received: 2
// Received: 3
// Received: 4
// Received: 5

 

 

• async* 함수 안에서 yield를 사용하면 데이터를 순차적으로 전달할 수 있음.
• .listen()을 사용하면 스트림에 새로운 데이터가 들어올 때마다 실행됨.

 

 

(2) StreamController를 사용한 Stream 생성

Dart에서는 StreamController를 사용하여 수동으로 Stream을 생성하고 데이터를 추가할 수 있습니다.

import 'dart:async';

void main() {
  final StreamController<int> controller = StreamController<int>();

  // 스트림 리스너 (데이터가 들어올 때마다 실행)
  controller.stream.listen((data) {
    print("Received: $data");
  });

  // 데이터 추가 (스트림에 새로운 값 전달)
  controller.add(1);
  controller.add(2);
  controller.add(3);

  controller.close(); // 스트림 종료
}

// Received: 1
// Received: 2
// Received: 3

 

 

• StreamController를 사용하면 직접 .add()를 통해 데이터 전달 가능.
• .close()를 호출하지 않으면 스트림이 계속 열려 있음.

---

4. Stream 데이터 처리 (변환 및 필터링)

Dart의 Stream은 데이터를 변환하거나 필터링할 수 있는 여러 기능을 제공합니다.

 

(1) map()을 이용한 변환

Stream<int> numberStream() async* {
  for (int i = 1; i <= 5; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i;
  }
}

void main() {
  numberStream().map((number) => "Number: $number").listen((data) {
    print(data);
  });
}

// Number: 1
// Number: 2
// Number: 3
// Number: 4
// Number: 5

 

 

(2) where()를 이용한 필터링

void main() {
  Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5, 6]);

  stream.where((number) => number % 2 == 0).listen((data) {
    print("Even Number: $data");
  });
}

// Even Number: 2
// Even Number: 4
// Even Number: 6

 

---

5. Stream의 두 가지 유형

(1) Single Subscription Stream (단일 구독 스트림)

• 일반적으로 한 번만 구독 가능하며, .listen()을 여러 번 호출하면 오류 발생.
• 기본적으로 StreamController는 Single Subscription Stream을 생성.

Stream<int> numberStream() async* {
  for (int i = 1; i <= 3; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i;
  }
}

void main() {
  Stream<int> stream = numberStream();

  stream.listen((data) {
    print("Listener 1: $data");
  });

  // 같은 Stream을 다시 구독하면 오류 발생!
  // stream.listen((data) {
  //   print("Listener 2: $data");
  // });
}

 

(2) Broadcast Stream (다중 구독 스트림)

• 여러 개의 리스너가 동일한 스트림을 구독할 수 있음.
• StreamController.broadcast()를 사용하여 생성.

void main() {
  final StreamController<int> controller = StreamController<int>.broadcast();

  // 여러 개의 리스너 추가 가능
  controller.stream.listen((data) {
    print("Listener 1: $data");
  });

  controller.stream.listen((data) {
    print("Listener 2: $data");
  });

  controller.add(1);
  controller.add(2);
  controller.add(3);

  controller.close();
}

// Listener 1: 1
// Listener 2: 1
// Listener 1: 2
// Listener 2: 2
// Listener 1: 3
// Listener 2: 3

 

• broadcast를 사용하면 하나의 스트림을 여러 리스너가 동시에 구독할 수 있음.

---

6. Stream 에러 처리

스트림에서 오류가 발생할 수 있으므로 에러 처리를 위한 두 가지 방법을 제공합니다.

 

(1) .handleError() 사용

Stream<int> errorStream() async* {
  yield 1;
  throw Exception("Error 발생!");
}

void main() {
  errorStream().handleError((error) {
    print("Error handled: $error");
  }).listen((data) {
    print("Received: $data");
  });
}

 

(2) try-catch & onError 사용

void main() {
  Stream<int> stream = Stream<int>.error("데이터 로딩 실패");

  stream.listen(
    (data) {
      print("Received: $data");
    },
    onError: (error) {
      print("Caught error: $error");
    },
  );
}