当使用Golang监听MongoDB的Change Stream时，可以通过记录每次获取的最后一个事件的resumeToken来实现断点恢复。下面是一个简单的示例代码：
package main
import (
"context"
"fmt"
"log"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
func main() {
// 创建MongoDB客户端
client, err := mongo.Connect(context.Background(), options.Client().ApplyURI("mongodb://localhost:27017"))
if err != nil {
log.Fatal(err)
}
// 获取数据库和集合
db := client.Database("mydb")
coll := db.Collection("mycollection")
// 配置change stream选项，设置resumeToken字段为上一次获取到的resumeToken值
changeStreamOpts := options.ChangeStream().SetResumeAfter(getResumeToken())
// 打开Change Stream流
changeStream, err := coll.Watch(context.Background(), bson.D{}, changeStreamOpts)
if err != nil {
log.Fatal(err)
}
defer changeStream.Close(context.Background())
for changeStream.Next(context.Background()) {
var event bson.M
if err := changeStream.Decode(&event); err != nil {
log.Println(err)
continue
}
// 处理收到的事件数据
fmt.Printf("Received Event: %v\n", event)
// 保存当前事件的resumeToken用于断点恢复
saveResumeToken(event["_id"])
}
if changeStream.Err() != nil {
log.Fatal(changeStream.Err())
}
}
func getResumeToken() *bson.Document {
// TODO: 从持久化存储中获取上次保存的resumeToken
// 如果是第一次运行，则返回nil或初始值
return nil
}
func saveResumeToken(token interface{}) {
// TODO: 将当前事件的resumeToken持久化保存，以便下次恢复使用
}
在示例代码中，我们首先创建了一个MongoDB客户端，并获取到要监听的数据库和集合。然后通过options.ChangeStream().SetResumeAfter()方法设置change stream选项的resumeAfter字段为上次记录的resumeToken值。
接下来，我们打开Change Stream流，并使用changeStream.Next()方法循环处理收到的事件数据。在处理每个事件之前，我们会将当前事件的resumeToken保存下来。你可以根据自己需求选择将其持久化存储在文件、数据库或其他方式中。
最后，在getResumeToken()函数中，你需要实现从持久化存储中获取上次保存的resumeToken的逻辑，并返回相应的bson.Document对象。同样，在saveResumeToken()函数中，你需要实现将当前事件的resumeToken持久化保存的逻辑。
请注意，以上示例只是一个简单演示，实际应用中可能还需要考虑错误处理、连接管理、程序退出等其他方面的问题。
内容由零声教学AI助手提供，问题来源于学员提问
