Go操作MongoDB:文档数据库实战

第一次用 MongoDB 的时候，我还在想：这不就是一个 JSON 数据库吗，有什么难的？结果真正用到项目里才发现，schema 灵活是优点也是坑。有一次同事把一个字段从字符串改成了数字，老数据还是字符串，新数据变成了数字，查询的时候类型不匹配，死活查不出来，排查了半天才发现是历史数据类型不一致的问题。

MongoDB 适合那种结构不固定、可能频繁变化的数据。比如用户画像，今天加个字段，明天改个类型，用 MySQL 得 alter table，用 MongoDB 直接塞就行。下面聊聊 Go 里怎么操作 MongoDB，以及我踩过的一些坑。

连接与基础配置

官方驱动 mongo-driver 已经挺成熟了，连接代码很简单：

package main

import (
"context"
"fmt"
"time"

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

type MongoDB struct {
 client   *mongo.Client
 database *mongo.Database
}

funcNewMongoDB(uri, dbName string)(*MongoDB, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 clientOptions := options.Client().
  ApplyURI(uri).
  SetMaxPoolSize(100).
  SetMinPoolSize(10).
  SetMaxConnIdleTime(30 * time.Second)

 client, err := mongo.Connect(ctx, clientOptions)
if err != nil {
returnnil, err
 }

if err := client.Ping(ctx, readpref.Primary()); err != nil {
returnnil, err
 }

 fmt.Println("MongoDB 连接成功")
return &MongoDB{
  client:   client,
  database: client.Database(dbName),
 }, nil
}

func(m *MongoDB)Close()error {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
return m.client.Disconnect(ctx)
}

func(m *MongoDB)Collection(name string) *mongo.Collection {
return m.database.Collection(name)
}

连接池参数根据实际并发量调整。我们线上一个服务 QPS 几千，MaxPoolSize 设到 200 才够用。如果连接池太小，高并发下会出现 connection pool exhausted 错误。

数据模型设计

MongoDB 的文档模型设计是个学问。嵌套还是引用？取决于查询模式。如果每次查用户都要带 profile，那就嵌套；如果 profile 数据量很大且不是每次都需要，就分开存。

package model

import (
"time"

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

type User struct {
 ID        primitive.ObjectID `bson:"_id,omitempty" json:"id"`
 Username  string`bson:"username" json:"username"`
 Email     string`bson:"email" json:"email"`
 Password  string`bson:"password" json:"-"`
 Profile   UserProfile        `bson:"profile" json:"profile"`
 Tags      []string`bson:"tags" json:"tags"`
 Status    int`bson:"status" json:"status"`
 CreatedAt time.Time          `bson:"created_at" json:"created_at"`
 UpdatedAt time.Time          `bson:"updated_at" json:"updated_at"`
}

type UserProfile struct {
 Nickname string`bson:"nickname" json:"nickname"`
 Avatar   string`bson:"avatar" json:"avatar"`
 Gender   int`bson:"gender" json:"gender"`
 Age      int`bson:"age" json:"age"`
 City     string`bson:"city" json:"city"`
}

type Article struct {
 ID        primitive.ObjectID `bson:"_id,omitempty" json:"id"`
 Title     string`bson:"title" json:"title"`
 Content   string`bson:"content" json:"content"`
 AuthorID  primitive.ObjectID `bson:"author_id" json:"author_id"`
 Tags      []string`bson:"tags" json:"tags"`
 ViewCount int`bson:"view_count" json:"view_count"`
 LikeCount int`bson:"like_count" json:"like_count"`
 Comments  []Comment          `bson:"comments" json:"comments"`
 Status    int`bson:"status" json:"status"`
 CreatedAt time.Time          `bson:"created_at" json:"created_at"`
 UpdatedAt time.Time          `bson:"updated_at" json:"updated_at"`
}

type Comment struct {
 ID        primitive.ObjectID `bson:"_id,omitempty" json:"id"`
 UserID    primitive.ObjectID `bson:"user_id" json:"user_id"`
 Content   string`bson:"content" json:"content"`
 CreatedAt time.Time          `bson:"created_at" json:"created_at"`
}

primitive.ObjectID 是 MongoDB 的默认 ID 类型，占 12 字节，包含时间戳信息，按 _id 排序天然就是按创建时间排序。如果你有自己的 ID 生成策略（比如雪花算法），也可以不用 ObjectID。

CRUD 操作

封装一个 Repository 层，把常用操作包起来：

package repository

import (
"context"
"time"

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

type UserRepository struct {
 collection *mongo.Collection
}

funcNewUserRepository(db *MongoDB) *UserRepository {
return &UserRepository{collection: db.Collection("users")}
}

// 创建
func(r *UserRepository)Create(user *User)error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 user.CreatedAt = time.Now()
 user.UpdatedAt = time.Now()

 result, err := r.collection.InsertOne(ctx, user)
if err != nil {
return err
 }
 user.ID = result.InsertedID.(primitive.ObjectID)
returnnil
}

// 批量创建
func(r *UserRepository)CreateMany(users []User)error {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 docs := make([]interface{}, len(users))
for i := range users {
  users[i].CreatedAt = time.Now()
  users[i].UpdatedAt = time.Now()
  docs[i] = users[i]
 }
 _, err := r.collection.InsertMany(ctx, docs)
return err
}

// 根据 ID 查询
func(r *UserRepository)FindByID(id primitive.ObjectID)(*User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

var user User
 err := r.collection.FindOne(ctx, bson.M{"_id": id}).Decode(&user)
if err != nil {
returnnil, err
 }
return &user, nil
}

// 根据用户名查询
func(r *UserRepository)FindByUsername(username string)(*User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

var user User
 err := r.collection.FindOne(ctx, bson.M{"username": username}).Decode(&user)
if err != nil {
returnnil, err
 }
return &user, nil
}

// 分页查询
func(r *UserRepository)FindAll(page, pageSize int)([]User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 skip := int64((page - 1) * pageSize)
 limit := int64(pageSize)

 opts := options.Find().
  SetSkip(skip).
  SetLimit(limit).
  SetSort(bson.D{{"created_at", -1}})

 cursor, err := r.collection.Find(ctx, bson.M{}, opts)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var users []User
if err := cursor.All(ctx, &users); err != nil {
returnnil, err
 }
return users, nil
}

// 条件查询（带总数）
func(r *UserRepository)FindByCondition(filter bson.M, page, pageSize int)([]User, int64, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 total, err := r.collection.CountDocuments(ctx, filter)
if err != nil {
returnnil, 0, err
 }

 skip := int64((page - 1) * pageSize)
 limit := int64(pageSize)
 opts := options.Find().SetSkip(skip).SetLimit(limit).SetSort(bson.D{{"created_at", -1}})

 cursor, err := r.collection.Find(ctx, filter, opts)
if err != nil {
returnnil, 0, err
 }
defer cursor.Close(ctx)

var users []User
if err := cursor.All(ctx, &users); err != nil {
returnnil, 0, err
 }
return users, total, nil
}

更新操作注意 $set 和 $inc 的区别：

// 更新用户
func(r *UserRepository)Update(id primitive.ObjectID, update bson.M)error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 update["updated_at"] = time.Now()
 _, err := r.collection.UpdateOne(ctx, bson.M{"_id": id}, bson.M{"$set": update})
return err
}

// 自增字段
func(r *UserRepository)Increment(id primitive.ObjectID, field string, value int)error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 _, err := r.collection.UpdateOne(ctx, bson.M{"_id": id}, bson.M{
"$inc": bson.M{field: value},
"$set": bson.M{"updated_at": time.Now()},
 })
return err
}

// 数组操作
func(r *UserRepository)AddToArray(id primitive.ObjectID, field string, value interface{})error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 _, err := r.collection.UpdateOne(ctx, bson.M{"_id": id}, bson.M{
"$push": bson.M{field: value},
"$set":  bson.M{"updated_at": time.Now()},
 })
return err
}

func(r *UserRepository)RemoveFromArray(id primitive.ObjectID, field string, value interface{})error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 _, err := r.collection.UpdateOne(ctx, bson.M{"_id": id}, bson.M{
"$pull": bson.M{field: value},
"$set":  bson.M{"updated_at": time.Now()},
 })
return err
}

// 删除
func(r *UserRepository)Delete(id primitive.ObjectID)error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
 _, err := r.collection.DeleteOne(ctx, bson.M{"_id": id})
return err
}

// 软删除
func(r *UserRepository)SoftDelete(id primitive.ObjectID)error {
 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

 _, err := r.collection.UpdateOne(ctx, bson.M{"_id": id}, bson.M{"$set": bson.M{
"status":     -1,
"deleted_at": time.Now(),
"updated_at": time.Now(),
 }})
return err
}

复杂查询

MongoDB 的查询语法比 SQL 灵活，但也更容易写错：

// 多条件查询
func(r *UserRepository)SearchUsers(keyword string, status int, tags []string)([]User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 filter := bson.M{}
if keyword != "" {
  filter["$or"] = []bson.M{
   {"username": bson.M{"$regex": keyword, "$options": "i"}},
   {"email": bson.M{"$regex": keyword, "$options": "i"}},
  }
 }
if status >= 0 {
  filter["status"] = status
 }
iflen(tags) > 0 {
  filter["tags"] = bson.M{"$in": tags}
 }

 cursor, err := r.collection.Find(ctx, filter)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var users []User
if err := cursor.All(ctx, &users); err != nil {
returnnil, err
 }
return users, nil
}

// 范围查询
func(r *UserRepository)FindByAgeRange(minAge, maxAge int)([]User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 filter := bson.M{
"profile.age": bson.M{
"$gte": minAge,
"$lte": maxAge,
  },
 }

 cursor, err := r.collection.Find(ctx, filter)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var users []User
if err := cursor.All(ctx, &users); err != nil {
returnnil, err
 }
return users, nil
}

// 嵌套文档查询
func(r *UserRepository)FindByCity(city string)([]User, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 cursor, err := r.collection.Find(ctx, bson.M{"profile.city": city})
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var users []User
if err := cursor.All(ctx, &users); err != nil {
returnnil, err
 }
return users, nil
}

// 投影查询（只返回指定字段）
func(r *UserRepository)FindUsernamesOnly()([]bson.M, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 opts := options.Find().SetProjection(bson.M{
"username": 1,
"email":    1,
"_id":      0,
 })

 cursor, err := r.collection.Find(ctx, bson.M{}, opts)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var results []bson.M
if err := cursor.All(ctx, &results); err != nil {
returnnil, err
 }
return results, nil
}

简单聚合

聚合管道是 MongoDB 的强项，基础的统计很简单：

// 按状态统计用户数量
func(r *UserRepository)CountByStatus()([]bson.M, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 pipeline := mongo.Pipeline{
  {{"$group", bson.D{
   {"_id", "$status"},
   {"count", bson.D{{"$sum", 1}}},
  }}},
  {{"$sort", bson.D{{"count", -1}}}},
 }

 cursor, err := r.collection.Aggregate(ctx, pipeline)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var results []bson.M
if err := cursor.All(ctx, &results); err != nil {
returnnil, err
 }
return results, nil
}

// 按城市统计
func(r *UserRepository)CountByCity()([]bson.M, error) {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

 pipeline := mongo.Pipeline{
  {{"$group", bson.D{
   {"_id", "$profile.city"},
   {"count", bson.D{{"$sum", 1}}},
   {"avg_age", bson.D{{"$avg", "$profile.age"}}},
  }}},
  {{"$sort", bson.D{{"count", -1}}}},
  {{"$limit", 10}},
 }

 cursor, err := r.collection.Aggregate(ctx, pipeline)
if err != nil {
returnnil, err
 }
defer cursor.Close(ctx)

var results []bson.M
if err := cursor.All(ctx, &results); err != nil {
returnnil, err
 }
return results, nil
}

几个容易踩的坑

类型不一致。MongoDB 是弱类型的，同一个字段在不同文档里可以是不同类型。查询时 bson.M{"age": 18} 匹配不到 age 是 "18"（字符串）的文档。建议写数据时做类型校验，或者查询时用 $type 过滤。

ObjectID 的时间精度。ObjectID 包含秒级时间戳，不要用它来做需要毫秒精度的排序。如果有高并发写入，同一秒内生成的 ObjectID 排序顺序不保证和写入顺序完全一致。

数组更新原子性。$push 和 $pull 是原子操作，但如果数组很大（比如几万条评论），更新性能会很差。建议数组控制在几百条以内，超了就拆分到单独的 collection。

默认连接超时。mongo.Connect 默认没有超时，网络不通时会一直卡着。一定要加 context.WithTimeout。

忽略游标关闭。Find 返回的 cursor 一定要 defer cursor.Close()，否则连接泄漏。我们线上就出现过因为没关 cursor 导致连接池耗尽的情况。

CountDocuments 慢。大数据量下 CountDocuments 会很慢，因为它要扫描匹配的所有文档。如果只需要估算值，用 EstimatedDocumentCount。或者自己维护一个计数器 collection。

MongoDB 用好了很爽，schema 灵活、查询丰富、水平扩展方便。但如果把它当 MySQL 用，什么关系都嵌套、什么字段都建索引，后面维护起来会很痛苦。设计文档模型时多想想查询模式，按查询模式来反推存储结构。