flutter开发-基于dart:io的socket解决方案

前言

本篇文章主要是记录笔者在实际开发中，关于flutter 中基于socket的业务功能开发，主要包括:

socket的连接/断开
网络监听与重连机制
自定义拆包，封包以及黏包处理

业务开发中的套接字简单数据结构

|::::::Head:::::::| ：：：：：：：Body ：：：：：：|
Head：这里主要描述了Body 的长度，一个 4byte 的 int 类型；
Body：这个部分还有很多结构字段，这里只是简单表示一下；

dart:io

File, socket, HTTP, and other I/O support for non-web applications.
Important: Browser-based applications can't use this library. Only servers, command-line scripts, and Flutter mobile apps can import and use dart:io.
This library allows you to work with files, directories, sockets, processes, HTTP servers and clients, and more. Many operations related to input and output are asynchronous and are handled using Futures or Streams, both of which are defined in the dart:async library.

从官方文档中，我们可以看出dart:io在flutter中的重要地位。这里我们主要用dart:io下的socket模块。下面我们看下socket模块下都有些什么：

WebSocket

WebSocket类提供对Web套接字协议的支持。这允许客户端和服务器应用程序之间的全双工通信。Web套接字服务器使用普通的HTTP服务器来接受Web套接字连接。初始握手是HTTP请求，然后将其升级为Web套接字连接。服务器使用WebSocketTransformer升级请求，并侦听返回的Web套接字上的数据。例如，这是一个微型服务器，用于监听WebSocket上的“ ws”数据：

runZoned(() async {
  var server = await HttpServer.bind('127.0.0.1', 8090);
  server.listen((HttpRequest req) async {
    if (req.uri.path == '/ws') {
      var socket = await WebSocketTransformer.upgrade(req);
      socket.listen(handleMsg);
    }
  });
}, onError: (e) => print("An error occurred."));

客户端使用connect()方法和使用Web套接字协议的URI连接到WebSocket。客户端可以使用add()方法写入WebSocket。

var socket = await WebSocket.connect('ws://127.0.0.1:8090/ws');
socket.add('Hello, World!');

Socket and ServerSocket

客户端和服务器使用套接字通过TCP协议进行通信。在服务器端使用ServerSocket，在客户端使用Socket。服务器使用bind（）方法创建一个监听套接字，然后监听套接字上的传入连接。例如：

ServerSocket.bind('127.0.0.1', 8081)
  .then((serverSocket) {
    serverSocket.listen((socket) {
      socket.transform(utf8.decoder).listen(print);
    });
  });

客户端使用connect（）方法连接Socket，该方法返回Future。使用write（），writeln（）或writeAll（）是通过套接字发送数据的最简单方法。例如：

Socket.connect('127.0.0.1', 8081).then((socket) {
  socket.write('Hello, World!');
});

项目中使用

在项目中的代码量非常的小，如下：

void connectSocket(String host,int port) async{
    this.host = host;
    this.port = port;
    try {
      await Socket.connect(host, port, timeout: Duration(seconds: 5)).then((socket){
        // ignore: missing_return
        mSocket=socket;
        print('---------连接成功------------$mSocket');
        if(onSuccess != null){
          onSuccess();
        }
        socket.listen(_receivedMsgHandler,
            onError: _errorHandler,
            onDone: _doneHandler,
            cancelOnError: false);
      });
    } catch (e) {
      print("连接socket出现异常，e=${e.toString()}");
    }
  }

这里的socket.listen()方法

//onData：收到socket消息处理的回调
//onError：scoket发生错误时回调
//onDone：流关闭或者消息事件发送完成后回调
StreamSubscription<T> listen(void onData(T event)?,
      {Function? onError, void onDone()?, bool? cancelOnError});

_receivedMsgHandler,当接收到socket消息时，会触发该回调

黏包处理

...
//存放子数据
List<int> mutableData = List<int>();
...
void _receivedMsgHandler(Uint8List data){
    mutableData.addAll(data);
    while(true){
      packageLen = _getPackageLength(mutableData);
      if(mutableData.length - 4 >= packageLen){
        if(getMessageData != null){
          getMessageData(Uint8List.fromList(mutableData));
        }
        if(mutableData == null){break;}
        //清空处理过的数据流
        mutableData?.replaceRange(0, packageLen+4, []);
        packageLen = 0;
        if(mutableData.length < 4){
          break;
        }
      }else{
        break;
      }
    }
  }
//获取包长
int _getPackageLength(List<int> data){
 Uint8List uint8Data = Uint8List.fromList(mutableData);
 ByteData data = ByteData.sublistView(uint8Data);
 int length = data.getInt32(0);
 return length;
}

关于Uint8List其实就是一个二进制的数据，iOS可以理解为NSData，在Dart中Uint8List是个抽象类，实现了List<int>协议。

发送二进制数据

void sendMessage(Uint8List message){
  mSocket.add(message);
}
//Adds byte [data] to the target consumer, ignoring [encoding].
void add(List<int> data);
//Converts [obj] to a String by invoking [Object.toString] and 
//[add]s the encoding of the result to the target consumer.
void write(Object? obj);

自定义封包与拆包
这里主要是将自定义的Model数据转换为约定好的二进制格式(这里是Uint8List)，即CustomModel->Uint8List，Uint8List->CustomModel。
这里主要涉及到BytesBuffer,ByteData,Uint8List的使用，主要涉及到dart类：dart:convert，dart:typed_data。这里我用到了一个不错的辅助插件：buffer: ^1.0.7(在pub.dev中可以找到)，其本质就是对上面三个类的封装。

封包：

// 模型数据 --> 二进制数据
static Uint8List writeDataWithModel(JXHMessageModel model){

    ByteDataWriter contentWrite = ByteDataWriter();

    contentWrite.writeInt16(model.messageType);
    contentWrite.writeInt32(model.eventId);

    contentWrite.writeInt32(model.roomId.length);
    contentWrite.write(model.roomId.toString().codeUnits);

    contentWrite.writeInt32(model.fromUser.length);
    contentWrite.write(model.fromUser.toString().codeUnits);

    contentWrite.writeInt32(model.toUser.length);
    contentWrite.write(model.toUser.toString().codeUnits);

    contentWrite.writeInt32(model.identifier);
    contentWrite.writeInt32(model.body.length);

    List<int> allKeys = model.body.keys.toList();
    for(int i = 0;i<allKeys.length;i++){
      int key = allKeys[i];
      String value = model.body[key];
      contentWrite.writeInt32(key);
      contentWrite.writeInt32(value.length);
      contentWrite.write(value.toString().codeUnits);
    }
    ByteDataWriter sumWriteData = ByteDataWriter(bufferLength: contentWrite.toBytes().length+4);
    print('lenght:${contentWrite.toBytes().length}');
    sumWriteData.writeInt32(contentWrite.toBytes().length);
    sumWriteData.write(contentWrite.toBytes());
    print('sum-lenght:${sumWriteData.bufferLength}');
    return sumWriteData.toBytes();
  }

拆包:

// 二进制数据 --> 模型数据
static JXHMessageModel getSocketModelFromData(Uint8List bytes){

    ByteDataReader reader = ByteDataReader();
    //读入数据
    reader.add(bytes);
    //这个类好像是内部已经维护了offset,这里是自动偏移
    int messageLength = reader.readInt32();
    int messageType = reader.readInt16();
    int eventId = reader.readInt32();

    int roomIdLen = reader.readInt32();
    String roomId = String.fromCharCodes(reader.read(roomIdLen));

    int fromUserLen = reader.readInt32();
    String fromUser = String.fromCharCodes(reader.read(fromUserLen));

    int toUserLen = reader.readInt32();
    String toUser = String.fromCharCodes(reader.read(toUserLen));

    int identifier = reader.readInt32();

    int keyValueCount = reader.readInt32();
    Map<int,String> body = Map();
    for(int i = 0;i<keyValueCount;i++){
      int key = reader.readInt32();
      int valueLen = reader.readInt32();
      String value =  Utf8Decoder().convert(reader.read(valueLen));
      body[key] = value;
    }

    return JXHMessageModel(
        messageLength: messageLength,
        messageType: messageType,
        eventId: eventId,
        roomId: roomId,
        fromUser: fromUser,
        toUser: toUser,
        identifier: identifier,
        keyValueCount: keyValueCount,
        body: body
    );
  }

以上只是举例，实际需要根据业务来定.....

总结

未完待续...