Mongo Go 驱动变更流监听

示例说明

轮询 diff 迟钝且易漏，变更流把 MongoDB 的写入日志直接推给订阅者。管道里的 $match 在服务端执行——只有插入与更新事件会进网络，删除与心跳噪声被就地过滤。事件结构体用 bson 标签映射字段，Decode 后即可类型安全访问；ResumeToken 持久化后重启传入即续读。

技术要点

• $match 服务端过滤减少网络传输
• bson 标签映射事件结构体
• 恢复令牌实现跨重启续传

代码解读

watchOrders 以 nil 或已保存的令牌开启监听；SetFullDocument 让更新事件携带完整文档省去二次查询。循环里 Decode 到结构体后提取状态变化，错误解码跳过不中断流。外层 context 三十秒超时演示空闲窗口的退出路径。

使用建议

变更流需要副本集环境。令牌要与业务状态同级别的可靠存储。处理逻辑幂等容忍重连窗口的重放。

代码示例

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

func watchOrders(ctx context.Context, coll *mongo.Collection, resumeToken bson.Raw) {
	match := bson.M{"operationType": bson.M{"$in": bson.A{"insert", "update"}}}

	stream, err := coll.Watch(ctx, mongo.Pipeline{
		{{Key: "$match", Value: match}},
	}, options.ChangeStream().
		SetFullDocument(options.UpdateLookup).
		SetResumeAfter(resumeToken))
	if err != nil {
		log.Printf("watch failed: %v", err)
		return
	}
	defer stream.Close(ctx)

	for stream.Next(ctx) {
		var event struct {
			OperationType string             `bson:"operationType"`
			FullDocument  bson.M             `bson:"fullDocument"`
			ResumeToken   bson.Raw           `bson:"_id"`
		}
		if err := stream.Decode(&event); err != nil {
			log.Printf("decode failed: %v", err)
			continue
		}
		fmt.Printf("%s -> %v\n", event.OperationType, event.FullDocument["status"])
		// 持久化 resumeToken 后即可在进程重启时断点续传
	}

	if err := stream.Err(); err != nil {
		log.Printf("stream error: %v", err)
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := mongo.Connect(ctx, options.Client().ApplyURI("mongodb://localhost:27017"))
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	coll := client.Database("shop").Collection("orders")
	watchOrders(ctx, coll, nil)
}