如何使用 Go 访问 MongoDB Change Stream 数据答案

【问题标题】：How to access MongoDB Change Stream data using Go如何使用 Go 访问 MongoDB Change Stream 数据
【发布时间】：2022-11-21 21:40:15
【问题描述】：

我正在使用 golang 观看所有 insert 事件的 mongodb 集合。我使用了 mongodb 函数，改变流.

我的要求是访问返回事件中的数据，它似乎是 bson.m 类型

这是我的代码示例：

func watch_for_events() {

    fmt.Println("Watching Started....")

    matchPipeline := bson.D{{"$match", bson.D{{"operationType", "insert"}}}}

    // open a change stream with an empty pipeline parameter
    coll := mongoConnection.Database(mongo_db).Collection(added_collection)
    changeStream, err := coll.Watch(context.TODO(), mongo.Pipeline{matchPipeline})
    if err != nil {
        panic(err)
    }
    defer changeStream.Close(context.TODO())
    // iterate over the cursor to print the change stream events
    for changeStream.Next(context.TODO()) {
        fmt.Println(changeStream.Current)
        fmt.Printf("POD name is: %v", changeStream.Current.Lookup("pod_name"))
    }

    fmt.Println("Watching Ended....")
}

这是我得到的输出

map[_id:map[_data:82637B3725000000012B022C0100296E5A10046C554EC75E644A81AF98CC28BAF03C0246645F69640064637B37253BC63551C4856EB50004] clusterTime:{1669019429 1} documentKey:map[_id:ObjectID("637b37253bc63551c4856eb5")] fullDocument:map[_id:ObjectID("637b37253bc63551c4856eb5") containers_and_images:[[nginx] [nginx:stable]] pod_name:nginx-847c4cd46c-dn2sc total_container_count:1] ns:map[coll:app1-added-pods db:ng-db] operationType:insert]

但是 fmt.Printf("POD name is: %v", changeStream.Current.Lookup("pod_name")) 行表示未找到元素。我需要获取诸如pod_name、namespace、total_container_count等数据。

有人可以帮我吗？

【问题讨论】：

标签： mongodb go mongo-go changestream

【解决方案1】：

ChangeStream.Current 字段中可用的更改事件是一个包含插入的完整文档的 fullDocument 字段的文档。请注意，changeStream.Current 的类型为bson.Raw，而Raw.Lookup() 采用可变参数string 切片形式的路径：

func (r Raw) Lookup(key ...string) RawValue

所以为了得到插入文档的pod_name字段，这样做：

fmt.Printf("POD name is: %v",
    changeStream.Current.Lookup("fullDocument", "pod_name"))

另请注意，如果您需要来自更改事件或插入文档的其他信息，您可以使用 ChangeStream.Decode() 将更改事件解码为 Go 值。

使用简单的地图，您可以通过以下方式访问pod_name：

for changeStream.Next(context.TODO()) {
    fmt.Println(changeStream.Current)
    var event struct {
        Doc bson.M `bson:"fullDocument"`
    }
    if err := changeStream.Decode(&event); err != nil {
        fmt.Printf("Failed to decode event: %v", err)
        continue
    }
    fmt.Printf("POD name is: %v", event.Doc["pod_name"])
}

如果您已经有一个类型来为您的文档建模，那么您当然可以使用它。假设这是您对文档建模的类型：

type YourDocument struct {
    ID      primitive.ObjectID `bson:"_id"`
    PodName string             `bson:"pod_name"`
    // Other properties
}

使用那个：

for changeStream.Next(context.TODO()) {
    fmt.Println(changeStream.Current)
    var event struct {
        Doc YourDocument `bson:"fullDocument"`
    }
    if err := changeStream.Decode(&event); err != nil {
        fmt.Printf("Failed to decode event: %v", err)
        continue
    }
    fmt.Printf("POD name is: %v", event.Doc.PodName)
}

【讨论】：