Flutter结合Future与MQTT实时更新ListView/GridView内容方案咨询

你需要的是一个可动态更新的本地数据源，初始通过REST接口填充，后续通过MQTT消息修改数据源并触发UI刷新。FutureBuilder只适合一次性数据加载，StreamBuilder需要持续数据流，而你的场景更适合手动维护状态：

基础实现：StatefulWidget + setState #

这是最直观的方案，适合新手快速上手：

1. 定义数据模型 #

先创建对应REST和MQTT消息的Item模型：

class Item {
  final String id;
  final String name;
  // 根据你的接口添加其他字段

  Item({required this.id, required this.name});

  // 从JSON解析Item
  factory Item.fromJson(Map<String, dynamic> json) => Item(
        id: json['id'],
        name: json['name'],
      );
}

2. 实现页面逻辑 #

在StatefulWidget中维护列表状态，初始化时加载REST数据，同时建立MQTT连接监听更新：

import 'package:flutter/material.dart';
import 'package:http/http.dart' as http;
import 'dart:convert';
import 'package:mqtt_client/mqtt_client.dart';

class ItemListScreen extends StatefulWidget {
  const ItemListScreen({super.key});

  @override
  State<ItemListScreen> createState() => _ItemListScreenState();
}

class _ItemListScreenState extends State<ItemListScreen> {
  List<Item> _items = [];
  bool _isLoading = true;
  late MqttClient _mqttClient;

  @override
  void initState() {
    super.initState();
    // 先加载初始数据
    _loadInitialItems();
    // 初始化MQTT连接
    _setupMqtt();
  }

  // 从REST接口获取初始列表
  Future<void> _loadInitialItems() async {
    try {
      final response = await http.get(Uri.parse('https://your-api-url/items'));
      if (response.statusCode == 200) {
        final List<dynamic> rawData = json.decode(response.body);
        setState(() {
          _items = rawData.map((json) => Item.fromJson(json)).toList();
          _isLoading = false;
        });
      } else {
        setState(() => _isLoading = false);
        // 可添加错误提示
      }
    } catch (e) {
      setState(() => _isLoading = false);
      // 处理网络异常
    }
  }

  // 配置MQTT连接与消息监听
  void _setupMqtt() async {
    _mqttClient = MqttClient('your-mqtt-server-address', 'unique-client-id');
    _mqttClient.logging(on: false);

    // 连接成功后的回调
    _mqttClient.onConnected = () {
      // 订阅更新主题
      _mqttClient.subscribe('item-update-topic', MqttQos.atLeastOnce);
      // 监听消息
      _mqttClient.updates!.listen((messageList) {
        final message = messageList[0].payload as MqttPublishMessage;
        final payloadStr = MqttPublishPayload.bytesToStringAsString(message.payload.message);
        final updatedItem = Item.fromJson(json.decode(payloadStr));
        
        // 找到对应项替换并刷新UI
        setState(() {
          final targetIndex = _items.indexWhere((item) => item.id == updatedItem.id);
          if (targetIndex != -1) {
            _items[targetIndex] = updatedItem;
          }
          // 如果需要处理新增项，取消下面注释
          // else {
          //   _items.add(updatedItem);
          // }
        });
      });
    };

    // 构建连接消息
    final connMsg = MqttConnectMessage()
        .withClientIdentifier('unique-client-id')
        .keepAliveFor(60);

    _mqttClient.connectionMessage = connMsg;

    // 尝试连接
    try {
      await _mqttClient.connect();
    } catch (e) {
      _mqttClient.disconnect();
      // 处理连接失败逻辑
    }
  }

  @override
  void dispose() {
    _mqttClient.disconnect();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    if (_isLoading) {
      return const Center(child: CircularProgressIndicator());
    }

    return ListView.builder(
      itemCount: _items.length,
      itemBuilder: (ctx, index) {
        final item = _items[index];
        return ListTile(
          title: Text(item.name),
          // 根据需求添加其他UI元素
        );
      },
    );
  }
}

进阶方案：状态管理库（以Provider为例） #

如果项目规模较大，用状态管理库可以避免StatefulWidget的状态臃肿，让数据逻辑与UI分离：

1. 创建ItemProvider #

import 'package:flutter/foundation.dart';
import 'package:http/http.dart' as http;
import 'dart:convert';
import 'package:mqtt_client/mqtt_client.dart';

class ItemProvider extends ChangeNotifier {
  List<Item> _items = [];
  bool _isLoading = true;
  late MqttClient _mqttClient;

  List<Item> get items => _items;
  bool get isLoading => _isLoading;

  ItemProvider() {
    _loadInitialItems();
    _setupMqtt();
  }

  Future<void> _loadInitialItems() async {
    try {
      final response = await http.get(Uri.parse('https://your-api-url/items'));
      if (response.statusCode == 200) {
        final List<dynamic> rawData = json.decode(response.body);
        _items = rawData.map((json) => Item.fromJson(json)).toList();
        _isLoading = false;
        notifyListeners();
      } else {
        _isLoading = false;
        notifyListeners();
      }
    } catch (e) {
      _isLoading = false;
      notifyListeners();
    }
  }

  void _setupMqtt() async {
    // MQTT配置逻辑同前，唯一区别是更新数据后调用notifyListeners()
    _mqttClient = MqttClient('your-mqtt-server-address', 'unique-client-id');
    _mqttClient.onConnected = () {
      _mqttClient.subscribe('item-update-topic', MqttQos.atLeastOnce);
      _mqttClient.updates!.listen((messageList) {
        final message = messageList[0].payload as MqttPublishMessage;
        final payloadStr = MqttPublishPayload.bytesToStringAsString(message.payload.message);
        final updatedItem = Item.fromJson(json.decode(payloadStr));
        
        final targetIndex = _items.indexWhere((item) => item.id == updatedItem.id);
        if (targetIndex != -1) {
          _items[targetIndex] = updatedItem;
          notifyListeners(); // 通知UI刷新
        }
      });
    };

    // 连接逻辑同前
    final connMsg = MqttConnectMessage()
        .withClientIdentifier('unique-client-id')
        .keepAliveFor(60);
    _mqttClient.connectionMessage = connMsg;

    try {
      await _mqttClient.connect();
    } catch (e) {
      _mqttClient.disconnect();
    }
  }
}

2. 在UI中使用Provider #

import 'package:flutter/material.dart';
import 'package:provider/provider.dart';

class ItemListScreen extends StatelessWidget {
  const ItemListScreen({super.key});

  @override
  Widget build(BuildContext context) {
    return Consumer<ItemProvider>(
      builder: (ctx, provider, child) {
        if (provider.isLoading) {
          return const Center(child: CircularProgressIndicator());
        }
        return ListView.builder(
          itemCount: provider.items.length,
          itemBuilder: (ctx, index) {
            final item = provider.items[index];
            return ListTile(title: Text(item.name));
          },
        );
      },
    );
  }
}

核心思路总结 #

• 放弃用FutureBuilder/StreamBuilder直接绑定一次性Future或无数据流的场景，改为手动维护可变列表状态
• 初始数据通过REST接口加载后更新状态
• MQTT收到更新消息时，找到列表中对应ID的项替换，再通过setState或状态管理库的通知方法触发UI刷新
• 记得在页面销毁时断开MQTT连接，避免资源泄漏

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