当前位置：首页 > news >正文

鸿蒙Flutter StreamProvider流式数据处理：实时更新与StreamController

news 2026/7/20 17:55:51

一、StreamProvider简介

在Flutter开发中，我们经常需要处理实时数据更新，如WebSocket连接、实时聊天、实时监控等。StreamProvider是Provider库提供的一种特殊Provider，专门用于处理流式数据。

StreamProvider的核心特点：

持续监听Stream：StreamProvider会持续监听Stream的事件
实时更新UI：当Stream发出新数据时，自动更新UI
管理状态：支持loading、error、data三种状态
支持初始值：可以设置initialData作为初始值
自动取消订阅：当Widget销毁时，自动取消订阅

二、基本用法

2.1 创建StreamProvider

StreamProvider<int>(create:(context)=>Stream.periodic(constDuration(seconds:1),(count)=>count+1,),initialData:0,child:Consumer<int>(builder:(context,count,child){returnText('计数器:$count');},),)

2.2 创建流式数据源

Stream<int>createCounterStream(){returnStream.periodic(constDuration(seconds:1),(tick)=>tick,).take(10);}

三、处理实时数据更新

3.1 实时消息展示

classChatMessage{finalStringcontent;finalStringsender;ChatMessage({requiredthis.content,requiredthis.sender});}StreamProvider<List<ChatMessage>>(create:(context)=>chatService.messages,initialData:const[],child:Consumer<List<ChatMessage>>(builder:(context,messages,child){returnListView.builder(itemCount:messages.length,itemBuilder:(context,index){returnListTile(title:Text(messages[index].sender),subtitle:Text(messages[index].content),);},);},),)

四、使用StreamController

4.1 创建自定义流

classNotificationService{finalStreamController<String>_controller=StreamController.broadcast();Stream<String>getnotifications=>_controller.stream;voidsendNotification(Stringmessage){_controller.add(message);}voiddispose(){_controller.close();}}

4.2 使用StreamController注意事项

记得关闭StreamController：当不再使用时，必须调用close()方法
使用broadcast()创建广播流：如果需要多个订阅者，使用StreamController.broadcast()
处理错误：可以通过_controller.addError()添加错误

五、StreamProvider与ChangeNotifier配合

在实际项目中，StreamProvider经常与ChangeNotifierProvider配合使用：

MultiProvider(providers:[Provider(create:(_)=>NotificationService()),StreamProvider<String>(create:(context)=>context.read<NotificationService>().notifications,initialData:'等待通知...',),],child:constNotificationPage(),)

六、监听Stream状态

Consumer<AsyncValue<String>>(builder:(context,value,child){returnvalue.when(loading:()=>constText('连接中...'),error:(error,_)=>Text('错误:$error'),data:(message)=>Text('消息:$message'),);},)

七、完整示例

7.1 创建通知服务

classNotificationService{finalStreamController<String>_controller=StreamController.broadcast();Stream<String>getnotifications=>_controller.stream;voidsendNotification(Stringmessage){_controller.add(message);}voiddispose(){_controller.close();}}

7.2 创建计数器流服务

classCounterService{finalStreamController<int>_controller=StreamController<int>();int _count=0;Stream<int>getcounterStream=>_controller.stream;voidincrement(){_count++;_controller.add(_count);}voidreset(){_count=0;_controller.add(_count);}voiddispose(){_controller.close();}}

7.3 创建StreamProvider页面

classStreamProviderPageextendsStatelessWidget{constStreamProviderPage({super.key});@overrideWidgetbuild(BuildContextcontext){returnMultiProvider(providers:[Provider(create:(_)=>NotificationService()),Provider(create:(_)=>CounterService()),StreamProvider<String>(create:(context)=>context.read<NotificationService>().notifications,initialData:'等待通知...',),StreamProvider<int>(create:(context)=>context.read<CounterService>().counterStream,initialData:0,),],child:Scaffold(appBar:AppBar(title:constText('StreamProvider演示')),body:ListView(padding:constEdgeInsets.all(16),children:[_buildSection('基本用法',_basicUsage()),_buildSection('StreamController自定义流',_streamControllerUsage()),_buildSection('实时消息展示',_realTimeUsage()),_buildSection('状态处理',_stateUsage()),],),),);}Widget_buildSection(Stringtitle,Widgetcontent){returnCard(margin:constEdgeInsets.only(bottom:16),child:Padding(padding:constEdgeInsets.all(16),child:Column(crossAxisAlignment:CrossAxisAlignment.start,children:[Text(title,style:constTextStyle(fontSize:18,fontWeight:FontWeight.bold)),constSizedBox(height:12),content,],),),);}Widget_basicUsage(){returnConsumer<int>(builder:(context,count,child){returnColumn(children:[Text('计数器:$count',style:constTextStyle(fontSize:24)),constSizedBox(height:16),Row(mainAxisAlignment:MainAxisAlignment.center,children:[ElevatedButton(onPressed:()=>context.read<CounterService>().increment(),child:constText('增加'),),constSizedBox(width:8),ElevatedButton(onPressed:()=>context.read<CounterService>().reset(),child:constText('重置'),),],),],);},);}Widget_realTimeUsage(){returnConsumer<String>(builder:(context,message,child){returnColumn(children:[Text('通知消息:$message',style:constTextStyle(fontSize:18)),constSizedBox(height:16),TextField(decoration:constInputDecoration(labelText:'输入通知消息'),onSubmitted:(value){context.read<NotificationService>().sendNotification(value);},),],);},);}}

7.4 代码解析

在这个演示页面中：

基本用法：使用StreamProvider创建计数器流，实时更新UI
StreamController自定义流：使用StreamController创建自定义流
实时消息展示：展示实时通知消息的更新
状态处理：使用AsyncValue.when处理loading、error、data三种状态

八、StreamProvider与FutureProvider的区别

特性	FutureProvider	StreamProvider
数据类型	一次性数据	持续流式数据
适用场景	网络请求、数据库查询	WebSocket、实时聊天
状态更新	一次更新	多次更新
订阅管理	自动完成	自动订阅/取消订阅

九、关键要点

9.1 StreamProvider用于处理持续的流式数据

StreamProvider适合处理持续的流式数据，如WebSocket连接、实时聊天等。

9.2 initialData设置初始值

initialData参数用于设置初始值，在Stream发出第一个事件之前显示。

9.3 使用StreamController创建自定义流

使用StreamController可以创建自定义的流，控制数据的发送。

9.4 记得在适当的时候关闭StreamController

当不再使用StreamController时，必须调用close()方法，否则会导致内存泄漏。

9.5 AsyncValue.when处理loading、error、data三种状态

和FutureProvider一样，StreamProvider也可以使用AsyncValue.when处理三种状态。

十、常见错误

10.1 错误：忘记关闭StreamController

classNotificationService{finalStreamController<String>_controller=StreamController();voiddispose(){// 错误：忘记关闭StreamController}}

正确：关闭StreamController

classNotificationService{finalStreamController<String>_controller=StreamController();voiddispose(){_controller.close();}}

10.2 错误：使用普通StreamController多订阅

// 错误：普通StreamController只能有一个订阅者finalStreamController<String>_controller=StreamController();// 第二个订阅会抛出错误_controller.stream.listen((data)=>print(data));_controller.stream.listen((data)=>print(data));

正确：使用broadcast()创建广播流

// 正确：广播流可以有多个订阅者finalStreamController<String>_controller=StreamController.broadcast();

10.3 错误：在事件处理中使用context.watch

onPressed:(){// 错误：watch只能在build方法中使用finaldata=context.watch<AsyncValue<String>>();}

正确：使用context.read

onPressed:(){context.read<NotificationService>().sendNotification('消息');}

十一、最佳实践

11.1 创建专门的服务类管理Stream

将Stream的创建和管理封装在服务类中：

classChatService{finalStreamController<List<Message>>_controller=StreamController.broadcast();Stream<List<Message>>getmessages=>_controller.stream;voidaddMessage(Messagemessage){// 更新消息列表_controller.add(updatedMessages);}voiddispose(){_controller.close();}}

11.2 使用Provider管理服务类生命周期

使用Provider管理服务类的生命周期：

MultiProvider(providers:[Provider(create:(_)=>ChatService()),StreamProvider(create:(context)=>context.read<ChatService>().messages,initialData:const[],),],child:constChatPage(),)

11.3 在StatefulWidget中管理StreamController

如果StreamController的生命周期与Widget绑定，使用StatefulWidget管理：

classStreamPageextendsStatefulWidget{constStreamPage({super.key});@overrideState<StreamPage>createState()=>_StreamPageState();}class_StreamPageStateextendsState<StreamPage>{lateStreamController<int>_controller;@overridevoidinitState(){super.initState();_controller=StreamController();}@overridevoiddispose(){_controller.close();super.dispose();}@overrideWidgetbuild(BuildContextcontext){returnStreamProvider(create:(_)=>_controller.stream,initialData:0,child:constStreamDisplay(),);}}

11.4 使用AsyncValue.when处理状态

使用AsyncValue.when清晰地处理三种状态：

Consumer<AsyncValue<String>>(builder:(context,value,child){returnvalue.when(loading:()=>constCircularProgressIndicator(),error:(error,stack)=>Text('错误:$error'),data:(message)=>Text('消息:$message'),);},)