devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flutter: Websocket的使用与封装_flutter web_socket_channel

作者：菜鸟追梦旅行 | 2024-03-19 21:11:15

赞

踩

flutter web_socket_channel

1.引入相关插件库

  1.   # websocket
  2.   web_socket_channel: ^2.4.0
  3.   # 引入rxdart 解决Bad state: Stream has already been listened to.的报错问题
  4.   rxdart: ^0.27.7
  5.   # 状态管理*
  6.   provider: ^6.0.5

2.代码编写及封装

  1. import 'dart:async';
  2.  
  3. import 'package:rxdart/subjects.dart';
  4. import 'package:web_socket_channel/web_socket_channel.dart';
  5. import 'package:web_socket_channel/status.dart' as status;
  6.  
  7. typedef WebsocketMessageCallback = void Function(dynamic message);
  8.  
  9. /// 注册流控制器需要在哪些页面使用
  10. ///
  11. /// 目前分三种类型：
  12. ///
  13. /// 1.[customerLoginPage]游客模式下也就是在未登录时候（用户处于登录相关页面）
  14. /// 2.[customerMainPage]用户已登录，处于主页及其他登录后的页面下
  15. /// 3.[chatRoomPage]用户处在聊天室里（游客下的在线客服聊天室、用户已登录下的在线客服聊天室、买卖用户之间的聊天室）
  16. enum StreamControllerNameEnum {
  17.   customerLoginPage,
  18.   customerMainPage,
  19.   chatRoomPage;
  20. }
  21.  
  22. class WebsocketHelper {
  23.   WebsocketHelper._();
  24.  
  25.   static WebsocketHelper? _singleton;
  26.  
  27.   factory WebsocketHelper() => _singleton ??= WebsocketHelper._();
  28.  
  29. /// 用于连接websocket的链接uri
  30.   Uri? wsUri;
  31.  
  32. /// websocket连接后的对象
  33.   WebSocketChannel? _webSocketChannel;
  34.  
  35. /// 指定的stream流控制器存放map
  36.   Map<String, BehaviorSubject<String>>? streamControllerList;
  37.  
  38. /// 是否开启心跳
  39.   bool isOpenHeartBeat = true;
  40.  
  41. /// 用于控制心跳轮询
  42.   StreamSubscription<String>? _subscription;
  43.  
  44. /// 是否是用户主动触发关闭连接
  45.   bool isDisconnectByUser = false;
  46.  
  47. /// 另辟一个单独的消息回调函数
  48.   WebsocketMessageCallback? messageCallback;
  49.  
  50. /// 连接断开回调
  51.   Function()? onDone;
  52.  
  53. /// 连接出错回调
  54.   Function? onError;
  55.  
  56.   /// step one - ex: ws://localhost:1234
  57.   initSocket({required String wsPath, bool isOpenHeartBeat = true}) {
  58.     if (_webSocketChannel != null) {
  59.       print("socket实例已存在, 请勿重复创建");
  60.       return;
  61.     }
  62.  
  63.     // 自己项目中后端需要前端拼一个登录令牌用于控制后端逻辑处理，这里使用的是登录后的token
  64.     var authorization = "登录后的token";
  65.     wsUri = Uri.tryParse("$wsPath?Authorization=$authorization");
  66.     // wsUri = Uri.tryParse(wsPath);
  67.     if (wsUri == null) return;
  68.     this.isOpenHeartBeat = isOpenHeartBeat;
  69.     _connectWebsocket();
  70.   }
  71.  
  72.   /// [isRunForReConnect] 是否是由重连机制触发的此方法
  73.   void _connectWebsocket({bool isRunForReConnect = false}) {
  74.     _webSocketChannel = WebSocketChannel.connect(wsUri!);
  75.     if (!isRunForReConnect) {
  76.       isDisconnectByUser = false;
  77.     }
  78.   }
  79.  
  80.   /// step two - listen
  81.   void listen(
  82.       {WebsocketMessageCallback? messageCallback,
  83.       Function()? onDone,
  84.       Function? onError}) {
  85.     this.messageCallback = messageCallback;
  86.     this.onDone = onDone;
  87.     this.onError = onError;
  88.     streamControllerList ??= <String, BehaviorSubject<String>>{
  89.       // StreamControllerNameEnum.customerLoginPage.name: BehaviorSubject(),
  90.       // StreamControllerNameEnum.customerMainPage.name: BehaviorSubject(),
  91.       // StreamControllerNameEnum.chatRoomPage.name: BehaviorSubject()
  92.     };
  93.  
  94.     // 监听一系列连接情况（如收到消息、onDone：连接关闭、onError：接连异常）
  95.     _webSocketChannel?.stream.listen((message) {
  96.       print(
  97.           "websocket onData message = ${message.toString()}, type = ${message.runtimeType}");
  98.       if (message is String && message.isEmpty) {
  99.         // 消息为空（可能得情况：心跳 or another）
  100.         return;
  101.       }
  102.       // 通过流控制器把消息分发出去，在需要的页面监听此流的消息
  103.       streamControllerList?.forEach((key, value) {
  104.         // print("key = $key, value.isClosed = ${value.isClosed}");
  105.         if (!value.isClosed) {
  106.           value.sink.add(message);
  107.         }
  108.       });
  109.       this.messageCallback?.call(message);
  110.     }, onDone: () {
  111.       print("websocket onDone ...");
  112.       this.onDone?.call();
  113.       // 掉线重连
  114.       reConnect();
  115.     }, onError: (Object error, StackTrace stackTrace) {
  116.       print(
  117.           "websocket onError error = ${error.toString()}, stackTrace = ${stackTrace.toString()}");
  118.       showToast(msg: "连接服务器失败!");
  119.       this.onError?.call(error, stackTrace);
  120.     }, cancelOnError: false);
  121.     // 连接建立成功时的回调通知，可在此做心跳操作
  122.     _webSocketChannel?.ready.then((value) {
  123.       print("webSocket ready");
  124.       isDisconnectByUser = false;
  125.       if (isOpenHeartBeat) {
  126.         // 收到连接成功的回馈，开始执行心跳操作
  127.         _startHeartBeat();
  128.       }
  129.     });
  130.   }
  131.  
  132.   /// 掉线重连
  133.   void reConnect() {
  134.     if (isDisconnectByUser) return;
  135.     Future.delayed(
  136.       const Duration(seconds: 3),
  137.       () {
  138.         // disconnect();
  139.         _subscription?.cancel();
  140.         _subscription = null;
  141.         _webSocketChannel?.sink.close(status.abnormalClosure, "掉线重连");
  142.         _webSocketChannel = null;
  143.         _connectWebsocket();
  144.         listen(
  145.             messageCallback: messageCallback, onDone: onDone, onError: onError);
  146.       },
  147.     );
  148.   }
  149.  
  150. /// 发送消息
  151.   void sendMessage({required String message, bool needDisplayMsg = true}) {
  152.     print("websocket sendMessage message = $message");
  153.     if (needDisplayMsg) {
  154.       streamControllerList?.forEach((key, value) {
  155.         if (!value.isClosed) {
  156.           value.sink.add(message);
  157.         }
  158.       });
  159.     }
  160.  
  161.     _webSocketChannel?.sink.add(message);
  162.   }
  163.  
  164. /// 开启心跳
  165.   void _startHeartBeat() {
  166.     if (_subscription != null) {
  167.       print("websocket startHeartBeat _subscription != null");
  168.       return;
  169.     }
  170.     Future.delayed(
  171.       const Duration(seconds: 30),
  172.       () {
  173.         var pollingStream = StreamTool().timedPolling(
  174.             const Duration(seconds: 30), () => Future(() => ""), 100000000);
  175.         //进行流内容监听
  176.         _subscription = pollingStream.listen((result) {
  177.           sendMessage(message: "heart beat", needDisplayMsg: false);
  178.         });
  179.       },
  180.     );
  181.   }
  182.  
  183. /// 断开连接并销毁
  184.   void disconnect({bool isDisconnectByUser = false}) {
  185.     this.isDisconnectByUser = isDisconnectByUser;
  186.     _subscription?.cancel();
  187.     _subscription = null;
  188.     streamControllerList?.forEach((key, value) {
  189.       value.close();
  190.     });
  191.     streamControllerList?.clear();
  192.     _webSocketChannel?.sink.close(status.normalClosure, "用户退出聊天界面，聊天关闭");
  193.     _webSocketChannel = null;
  194.   }
  195.  
  196. /// 新建指定stream流控制器进行消息流回调
  197.   setNewStreamController(StreamControllerNameEnum streamControllerName) {
  198.     if (streamControllerList?.containsKey(streamControllerName.name) ?? false) {
  199.       streamControllerList?[streamControllerName.name]?.close();
  200.     }
  201.     streamControllerList?[streamControllerName.name] = BehaviorSubject();
  202.   }
  203. }

3.提供一个轮询工具类StreamTool

  1. import 'dart:async';
  2.  
  3. typedef FutureGenerator<T> = Future<T> Function();
  4.  
  5. class StreamTool {
  6.   /// interval 轮询时间间隔
  7.   /// maxCount 最大轮询数
  8.   Stream<T> timedPolling<T>(Duration interval, FutureGenerator<T> future,
  9.       [int maxCount = 1]) {
  10.     StreamController<T>? controller;
  11.     int counter = 0;
  12.     bool polling = true;
  13.  
  14.     void stopTimer() {
  15.       polling = false;
  16.     }
  17.  
  18.     void tick() async {
  19.       counter++;
  20.       T result = await future();
  21.       if (controller != null && !controller.isClosed) {
  22.         controller.add(result);
  23.       }
  24.       if (counter == maxCount) {
  25.         stopTimer();
  26.         controller?.close();
  27.       } else if (polling) {
  28.         Future.delayed(interval, tick);
  29.       }
  30.     }
  31.  
  32.     void startTimer() {
  33.       polling = true;
  34.       tick();
  35.     }
  36.  
  37.     //StreamSubscription调用pause，cancel时，stream里面的轮询也能响应暂停或取消
  38.     controller = StreamController<T>(
  39.       onListen: startTimer,
  40.       onPause: stopTimer,
  41.       onResume: startTimer,
  42.       onCancel: stopTimer,
  43.     );
  44.  
  45.     return controller.stream;
  46.   }
  47. }

4.新建全局的ChangeNotifier -> GlobalWebsocketVM

  1. class GlobalWebsocketVM extends ChangeNotifier {
  2. void startWebSocket() {
  3.     WebsocketHelper()
  4.       ..initSocket(wsPath: Api.wsUrlPath, isOpenHeartBeat: false)
  5.       ..listen(
  6.         messageCallback: (message) {
  7.           // 延迟500毫秒，使listview进行滑动到底部
  8.          // gotoListBottom();
  9.         },
  10.         onDone: () {},
  11.       );
  12.   }
  13.  
  14.  /// 获取socket实时数据流
  15.   ///
  16.   /// 每次都需要新绑定一个StreamController，避免数据流出现错乱情况
  17.   Stream<String>? getMessageStream(
  18.           StreamControllerNameEnum streamControllerName) =>
  19.       (WebsocketHelper()..setNewStreamController(streamControllerName))
  20.           .streamControllerList?[streamControllerName.name]
  21.           ?.stream;
  22. }

5.在入口类main.dart中MaterialApp中使用全局GlobalWebsocketVM

  1. late GlobalWebsocketVM socketVM;
  2.  
  3. @override
  4. void initState() {
  5.   socketVM = GlobalWebsocketVM();
  6. }
  7.  
  8. MaterialApp.router(
  9.       debugShowCheckedModeBanner: false,
  10.       onGenerateTitle: (context) => S.current.appName,
  11.       theme: ThemeData(
  12.         useMaterial3: true,
  13.         colorScheme: ColorScheme.fromSeed(seedColor: Colors.white),
  14.         appBarTheme: const AppBarTheme(
  15.             color: Colors.white, surfaceTintColor: Colors.white),
  16.         bottomAppBarTheme: BottomAppBarTheme.of(context)
  17.             .copyWith(color: Colors.white, surfaceTintColor: Colors.white),
  18.         scaffoldBackgroundColor: Colors.grey[200],
  19.         cardTheme: const CardTheme(
  20.             color: Colors.white, surfaceTintColor: Colors.white),
  21.         progressIndicatorTheme:
  22.             const ProgressIndicatorThemeData(color: AppColor.appThemeColor),
  23.         // 统一修改输入框光标颜色、文本选中颜色
  24.         textSelectionTheme: const TextSelectionThemeData(
  25.           cursorColor: AppColor.appThemeColor,
  26.           selectionColor: AppColor.appThemeColor,
  27.           selectionHandleColor: AppColor.appThemeColor,
  28.         ),
  29.         // ios主题色设置
  30.         cupertinoOverrideTheme:
  31.             const CupertinoThemeData(primaryColor: AppColor.appThemeColor),
  32.         iconButtonTheme: IconButtonThemeData(
  33.           style: AppButtonStyle.stGlobalDefaultBtn,
  34.         ),
  35.         textButtonTheme: TextButtonThemeData(
  36.           style: AppButtonStyle.stGlobalDefaultBtn,
  37.         ),
  38.         // primarySwatch: themeVM.theme,
  39.       ),
  40.       // locale: localeVM.getLocale(),
  41.       builder: FlutterSmartDialog.init(
  42.         builder: (context, mChild) {
  43.           return MultiProvider(
  44.             providers: [
  45.               ChangeNotifierProvider<UserVM>(
  46.                 create: (_) => userVM,
  47.               ),
  48.               // 在这里为每个页面添加GlobalWebsocketVM绑定
  49.               ChangeNotifierProvider<GlobalWebsocketVM>(
  50.                 create: (_) => socketVM,
  51.               ),
  52.             ],
  53.             builder: (context, child) => mChild ?? const SizedBox.shrink(),
  54.           );
  55.           /*return ChangeNotifierProvider<UserVM>(
  56.             create: (_) => userVM,
  57.             builder: (context, child) =>
  58.             mChild ?? const SizedBox.shrink(),
  59.           );*/
  60.         },
  61.       ),
  62.       localizationsDelegates: const [
  63.         S.delegate,
  64.         GlobalMaterialLocalizations.delegate,
  65.         GlobalWidgetsLocalizations.delegate,
  66.         GlobalCupertinoLocalizations.delegate,
  67.       ],
  68.       supportedLocales: S.delegate.supportedLocales,
  69.       // localeResolutionCallback: (_locale, supportedLocales) {
  70.       //   if (localeVM.getLocale() != null) {
  71.       //     //如果已经选定语言，则不跟随系统
  72.       //     return localeVM.getLocale();
  73.       //   } else {
  74.       //     //跟随系统
  75.       //     Locale locale;
  76.       //     if (supportedLocales.contains(_locale)) {
  77.       //       locale = _locale!;
  78.       //     } else {
  79.       //       //如果系统语言不是中文简体或美国英语，则默认使用美国英语
  80.       //       locale = const Locale('en', 'US');
  81.       //     }
  82.       //     return locale;
  83.       //   }
  84.       // },
  85.       routerConfig: RouterHelper.router,
  86.     );

6.页面中调用，在initState方法中建立连接，在build中使用StreamBuilder进行消息监听

  1. @override
  2.   void initState() {
  3.     super.initState();
  4.     // 建立连接ws global init
  5.     context.read<GlobalWebsocketVM>().startWebSocket();
  6.     
  7.   }
  8.  
  9. @override
  10.   Widget build(BuildContext context) {
  11.     return Scaffold(
  12.             resizeToAvoidBottomInset: false,
  13.             appBar: TitleBar.build(title: "正与${model.titleContent}沟通"),
  14.             // 监听聊天消息并刷新聊天列表
  15.             body: StreamBuilder<String>(
  16.           stream: context
  17.         .read<GlobalWebsocketVM>()
  18.         .getMessageStream(StreamControllerNameEnum.chatRoomPage),
  19.           builder: (context, snapshot) {
  20.             if (snapshot.connectionState == ConnectionState.active) {
  21.               if (snapshot.data?.isEmpty ?? true) {
  22.                 return const SizedBox.shrink();
  23.               }
  24.               addMessageAndRefreshUI("orderNo", snapshot.data!);
  25.               return const SizedBox.shrink();
  26.             }
  27.             return const SizedBox.shrink();
  28.           },
  29.           // catchError: (context, error) => error.toString(),
  30.         ),,
  31.           );
  32. }
  33.  
  34. void addMessageAndRefreshUI(String tag, String message) {
  35.     print("收到聊天消息：" + message);
  36. }

7.发送消息

  1. /// 在合适的地方（比如发送按钮点击发送聊天消息）
  2. void sendChatMessage() {
  3. WebsocketHelper().sendMessage(
  4.       message: "我发送一条消息",
  5.       needDisplayMsg: false,
  6.     );
  7. }

8.退出app断开websocket清理内存（可以在任何想断开websocket的地方调用销毁）

  1. /// 通常在dispose中调用销毁，可以在任何想断开websocket的地方调用销毁
  2. @override
  3.   void dispose() {
  4.     ScanHelper().dispose();
  5.     WebsocketHelper().disconnect(isDisconnectByUser: true);
  6.     super.dispose();
  7.   }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/268650
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号