Go使用MongoDB Pipeline按ID监听文档更新不生效问题

你的代码加了ID匹配后永久阻塞，核心原因有2个：

• 变更流匹配字段写错位置：MongoDB变更流的更新事件默认不会返回完整文档内容，文档主键_id默认存储在事件顶层的documentKey._id字段，你写的fullDocument._id匹配规则永远不会命中，自然收不到任何符合条件的事件。这也解释了为什么去掉匹配条件监听全集合可以正常收到事件。
• 附带的逻辑bug会加剧异常：
  • 创建变更流的错误处理逻辑完全写反：coll.Watch返回错误时scannerStream是空值，此时调用Close会直接触发空指针panic
  • 变更流循环退出后没有检查流错误，遇到网络异常、上下文超时等问题时不会主动通知主协程，会导致主协程永久阻塞在channel接收处
  • 你创建的无缓冲channel没有错误退出路径，一旦goroutine因为异常提前退出，主协程会永远卡在<-chn的位置

1. 把匹配条件里的fullDocument._id替换为documentKey._id，这是变更流事件里主键的默认存储位置
2. 如果你确实需要在事件中拿到更新后的完整文档内容，在调用Watch时添加FullDocument: updateLookup选项，此时事件中才会携带fullDocument字段
3. 修正错误处理逻辑，补充变更流的异常判断

修复后的核心代码如下：

func iterateChangeStream(routineCtx context.Context,stream *mongo.ChangeStream, chn chan string) {
    defer stream.Close(routineCtx)

    for stream.Next(routineCtx) {
        var data bson.M
        if err := stream.Decode(&data); err != nil {
            fmt.Println("decode event error: ", err)
            continue
        }
        chn <- "updated"
        return
    }
    // 检查循环退出是否因为流异常
    if err := stream.Err(); err != nil {
        fmt.Println("change stream error: ", err)
    }
}

func (s Storage) ListenForScannerUpdateById(id primitive.ObjectID) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    chn := make(chan string)

    coll := s.db.Collection("scanners")

    scan, err := s.GetScannerById(id)
    fmt.Println(scan)

    matchPipeline := bson.D{
        {
            "$match", bson.D{
                {"operationType", "update"},
                // 修正：变更流中_id默认存在documentKey下
                {"documentKey._id", id},
            },
        },
    }

    // 可选：需要返回更新后全量文档时开启该选项
    streamOpts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
    scannerStream, err := coll.Watch(ctx, mongo.Pipeline{matchPipeline}, streamOpts)
    // 修正错误处理逻辑
    if err != nil {
        fmt.Printf("create change stream failed: %v", err)
        return
    }

    routineCtx, cancelRoutine := context.WithCancel(context.Background())
    defer cancelRoutine()

    go iterateChangeStream(routineCtx, scannerStream, chn)
    msg, ok := <-chn
    if ok {
        fmt.Println(msg)
    }
    close(chn)
    return
}

补充说明：如果开启了FullDocument选项，更新事件返回的fullDocument是文档当前的最新值，而非修改前的值，仅在你需要读取更新后内容时开启即可，仅做ID匹配不需要开这个选项。

内容的提问来源于stack exchange，提问作者Daan van de Haar