聊天应用WebSocket重复监听消息与ListView重复构建问题排查

核心问题 #

问题出在WebSocket的sendMessage方法：每次调用该方法都会新增一个WebSocket流监听器。具体表现为：

• 第1次发消息：注册1个监听器，收到消息后向messages列表添加1条数据，显示1条
• 第2次发消息：又注册1个监听器，累计2个监听器，收到消息后触发2次messageListener，向列表添加2条重复数据，显示新增2条
• 第3次发消息：累计3个监听器，收到消息后添加3条重复数据，以此类推

服务器端消息发送正常，ListView本身无问题，是重复监听导致消息被多次添加到列表，进而ListView构建重复组件。

修复方案 #

1. 修改WebSocket控制器代码

将监听器注册移到初始化方法中，只执行一次；同时提供统一的消息订阅方式，避免重复注册：

class NotificationController {
  static final NotificationController _singleton = NotificationController._internal();

  StreamController<Map> messageStreamController = StreamController.broadcast();
  IOWebSocketChannel? channel;

  factory NotificationController() {
    return _singleton;
  }

  NotificationController._internal() {
    initWebSocketConnection();
  }

  initWebSocketConnection() async {
    var storedUserInfo = storage.getUserInfoStorage();
    Map storedData = await storedUserInfo;
    String userID = storedData['user_id'];

    try {
      channel = IOWebSocketChannel.connect(
        Uri.parse('ws://192.168.1.109:8001/chat/$userID/'),
        pingInterval: const Duration(seconds: 10),
      );
      // 仅在初始化时注册一次监听器
      channel?.stream.listen((data) {
        Map _message = json.decode(data);
        messageStreamController.add(_message);
      }, onError: (e) {
        print('WebSocket stream error: $e');
      }, onDone: () {
        _onDisconnected();
      });
    } on Exception catch (e) {
      print('Connection failed: $e');
      await Future.delayed(const Duration(seconds: 3));
      await initWebSocketConnection();
    }

    print("socket connection initialized");
  }

  void sendMessage(messageObject) {
    try {
      channel?.sink.add(json.encode(messageObject));
    } on Exception catch (e) {
      print('Send message error: $e');
    }
  }

  void _onDisconnected() {
    print('WebSocket disconnected, reconnecting...');
    initWebSocketConnection();
  }

  // 提供消息订阅方法
  StreamSubscription<Map> subscribeToMessages(Function(Map) listener) {
    return messageStreamController.stream.listen(listener);
  }

  // 取消订阅，避免内存泄漏
  void cancelSubscription(StreamSubscription<Map> subscription) {
    subscription.cancel();
  }
}

2. 修改UI层代码

在组件初始化时订阅消息流，仅执行一次：

List<ChatMessage> messages = [];
StreamSubscription<Map>? _messageSubscription;

@override
void initState() {
  super.initState();
  // 初始化时订阅消息流
  _messageSubscription = NotificationController().subscribeToMessages((message) {
    setState(() {
      messages.add(
        ChatMessage(message: message['message'], toUser: message['to_user'])
      );
    });
  });
}

@override
void dispose() {
  // 组件销毁时取消订阅
  _messageSubscription?.cancel();
  super.dispose();
}

// 发送消息的独立方法
void sendChatMessage(Map messageObject) {
  NotificationController().sendMessage(messageObject);
}

// ListView.builder代码保持不变
ListView.builder(
  shrinkWrap: true,
  scrollDirection: Axis.vertical,
  itemCount: messages.length,
  itemBuilder: (BuildContext context, int index) {
    return Stack(
      children: <Widget>[
        Container(
          padding: const EdgeInsets.only(
            left: 14,
            right: 14,
            top: 10,
            bottom: 10,
          ),
          child: Container(
            decoration: BoxDecoration(
              borderRadius: BorderRadius.circular(20),
              color: messages[index].toUser == widget.userUid
                  ? Colors.greenAccent
                  : Colors.blue,
            ),
            alignment: messages[index].toUser == widget.userUid
                ? Alignment.bottomRight
                : Alignment.topLeft,
            padding: const EdgeInsets.all(16),
            child: Text(
              messages[index].message as String,
              style: const TextStyle(fontSize: 15),
            ),
          ),
        ),
      ],
    );
  }
)

关键修改点 #

• 将WebSocket流监听逻辑从sendMessage移到初始化方法，确保仅注册一次监听器
• 使用独立StreamController转发消息，让UI层通过订阅该流接收消息，避免直接操作WebSocket流
• 在UI组件的initState中订阅消息，dispose中取消订阅，避免内存泄漏和重复订阅

内容的提问来源于stack exchange，提问作者coder