Flutter Dart中如何使用RethinkDB changefeeds获取数据库实时变更数据

你当前使用的递归全量轮询方案，不仅会重复返回相同数据，还会占用不必要的数据库和带宽资源。RethinkDB原生提供Changefeeds特性，专门用于数据变更实时订阅，仅在数据发生增删改时推送变更内容，完全匹配你的需求。

基础实现代码 #

配合rethink_db_ns包的实现示例如下：

Future<void> listenUserChanges(RethinkDb r, Connection connection) async {
  // 调用changes()方法开启变更订阅
  var changeFeed = await r.table('users').changes().run(connection);
  
  // 异步迭代变更事件，仅当数据发生变更时才会触发回调
  await for (var change in changeFeed) {
    // 变更对象包含两个核心字段：old_val（变更前值）、new_val（变更后值）
    print('变更前数据：${change['old_val']}');
    print('变更后数据：${change['new_val']}');
    
    // 此处可写入你自身的业务逻辑处理变更
  }
}

常用配置与使用场景 #

• 初始化先拉取全量当前数据、再监听后续变更：给changes方法传入includeInitial: true参数，首次迭代会返回当前表内所有数据的new_val，之后再返回后续变更事件
• 监听单条数据变更：如需只监听指定id的行变更，可将查询改为r.table('users').get('目标数据id').changes().run(connection)，减少不必要的事件推送
• 区分变更类型：可通过old_val和new_val的存在性判断操作类型：
  • 新增：old_val为null，new_val有值
  • 更新：old_val和new_val均有值
  • 删除：new_val为null，old_val有值

注意事项 #

1. 订阅过程中需要保持RethinkDB连接处于活跃状态，不要主动关闭连接，否则会导致订阅中断
2. 若出现网络波动、连接断开的情况，自行实现重连逻辑后重新执行订阅方法即可恢复监听

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