MongoDB自动同步主库数据（Change Stream）

迈不过友情╰ 2022-12-29 15:26 252阅读 0赞

# 主从同步原理 #

所有数据库同步原理几乎一样，MongoDB解析oplog，Mysql解析bin.log，今天实现了MongoDB同步机制，请关注小编下次更新Mysql同步机制。

*  **intial sync：初始化所有数据。**
 *  **replication：根据oplog实现增量同步。**

初始化所有数据把库库复制到目标库中，以下代码根据oplog实时读取并同步数据。

--------------------

## Change Stream ##

MongoDB官网提供操作流，通过watch机制监听oplog变更并反向通知程序。

MongoDB官网给出oplog操作类型：

*  **insert：添加数据**
 *  **delete：删除数据**
 *  **replace：替换数据**
 *  **update：更新数据**
 *  **drop：删除表**
 *  **rename：修改表名**
 *  **dropDatabase：删除数据库**
 *  **invalidate：drop/rename/dropDatabase 将导致invalidate被触发，并关闭 change stream**

还可以提供了监听条件、开始监听时间、Resume Tokens断点恢复等功能。

`注：断点恢复也属于监听条件，只支持一个监听条件`

## Change Events解析 ##

watch监听返回信息

{ 
            _id : {  // 存储元信息
                "_data" : <BinData|hex string> // resumeToken，用于断点恢复
            },
            "operationType" : "<operation>", // insert, delete, replace, update, drop, rename, dropDatabase, invalidate,部分仅支持4.0后的版本，详情见下
            "fullDocument" : {  <document> }, // 修改后的数据，出现在insert, replace, delete, update的事件中
            "ns" : {  // namespace
                "db" : "<database>", // 操作库名
                "coll" : "<collection" // 操作表名
            },
            "to" : {  // 只在operationType为rename的时候有效，表示改名以后的namespace
                "db" : "<database>",
                "coll" : "<collection"
            },
            "documentKey" : {  "_id" : <value> }, // 相当于o2字段。出现在insert, replace, delete, update事件中。正常只包含_id，对于sharded collection，还包括shard key
            "updateDescription" : {  // 只在operationType为update的时候出现，相当于是增量的修改，而replace是替换
                "updatedFields" : {  <document> }, // 更新的field的值
                "removedFields" : [ "<field>", ... ] // 删除的field列表
            },
            "clusterTime" : <Timestamp>, // 相当于ts字段
            "txnNumber" : <NumberLong>, // 相当于oplog里面的txnNumber，只在事务里面出现。事务号在一个事务里面单调递增
            "lsid" : {  // 相当于lsid字段，只在事务里面出现。logic session id，请求所在的session的id
                "id" : <UUID>,
                "uid" : <BinData>
            }
        }

## go 代码 ##

新建go mod工程，目录如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70]

utils可以忽略，小编自己写的映射。

*  **获取mongo连接包：** go get go.mongodb.org/mongo-driver/mongo

initMongo.go代码：

package mongo
    
    import (
    	"context"
    	"go.mongodb.org/mongo-driver/mongo"
    	"go.mongodb.org/mongo-driver/mongo/options"
    	"log"
    	"time"
    )
    
    func initMasterDBClient() *mongo.Database { 
    	var err error
    	clientOptions := options.Client().ApplyURI("mongodb://主库ip:端口/?connect=direct").SetConnectTimeout(5 * time.Second)
    
    	// 连接到MongoDB
    	client, err := mongo.Connect(context.TODO(), clientOptions)
    	if err != nil { 
    		log.Fatal(err)
    	}
    
    	//选择数据库
    	return client.Database("数据库")
    }
    
    func initSlaveDBClient() *mongo.Database { 
    	var err error
    	clientOptions := options.Client().ApplyURI("mongodb://从库ip:端口/?connect=direct").SetConnectTimeout(5 * time.Second)
    
    	// 连接到MongoDB
    	client, err := mongo.Connect(context.TODO(), clientOptions)
    	if err != nil { 
    		log.Fatal(err)
    	}
    
    	//选择数据库
    	return client.Database("数据库")
    }

sync.go代码：

package mongo
    
    import (
    	"context"
    	"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"
    	"log"
    	"time"
    )
    
    type StreamObject struct { 
    	Id                *WatchId `bson:"_id"`
    	OperationType     string
    	FullDocument      map[string]interface{ }
    	Ns                NS
    	UpdateDescription map[string]interface{ }
    	DocumentKey       map[string]interface{ }
    }
    
    type NS struct { 
    	Database   string `bson:"db"`
    	Collection string `bson:"coll"`
    }
    
    type WatchId struct { 
    	Data string `bson:"_data"`
    }
    
    const (
    	OperationTypeInsert  = "insert"
    	OperationTypeDelete  = "delete"
    	OperationTypeUpdate  = "update"
    	OperationTypeReplace = "replace"
    )
    
    var resumeToken bson.Raw
    
    func Sync() { 
    	go syncMaster()
    
    	for { 
    		time.Sleep(2 * time.Second)
    	}
    }
    
    func syncMaster() { 
    	for { 
    		//获得主库数据连接
    		client := initMasterDBClient()
    		watch(client)
    	}
    }
    
    func watch(client *mongo.Database) { 
    	defer func() { 
    		err := recover()
    		if err != nil { 
    			log.Printf("同步出现异常: %+v \n", err)
    		}
    	}()
    
    	//设置过滤条件
    	pipeline := mongo.Pipeline{ 
    		bson.D{ { "$match",
    			bson.M{ "operationType": bson.M{ "$in": bson.A{ "insert", "delete", "replace", "update"}}},
    		}},
    	}
    
    	//当前时间前一小时
    	now := time.Now()
    	m, _ := time.ParseDuration("-1h")
    	now = now.Add(m)
    	timestamp := &primitive.Timestamp{ 
    		T: uint32(now.Unix()),
    		I: 0,
    	}
    
    	//设置监听option
    	opt := options.ChangeStream().SetFullDocument(options.UpdateLookup).SetStartAtOperationTime(timestamp)
    	if resumeToken != nil { 
    		opt.SetResumeAfter(resumeToken)
    		opt.SetStartAtOperationTime(nil)
    	}
    
    	//获得watch监听
    	watch, err := client.Watch(context.TODO(), pipeline, opt)
    	if err != nil { 
    		log.Fatal("watch监听失败：", err)
    	}
    
    	//获得从库连接
    	slaveClient := initSlaveDBClient()
    
    	for watch.Next(context.TODO()) { 
    		var stream StreamObject
    		err = watch.Decode(&stream)
    		if err != nil { 
    			log.Println("watch数据失败：", err)
    		}
    
    		log.Println("=============", stream.FullDocument["_id"])
    
    		//保存现在resumeToken
    		resumeToken = watch.ResumeToken()
    
    		switch stream.OperationType { 
    		case OperationTypeInsert:
    			syncInsert(slaveClient, stream)
    		case OperationTypeDelete:
    			filter := bson.M{ "_id": stream.FullDocument["_id"]}
    			_, err := slaveClient.Collection(stream.Ns.Collection).DeleteOne(context.TODO(), filter)
    			if err != nil { 
    				log.Println("删除失败：", err)
    			}
    		case OperationTypeUpdate:
    			filter := bson.M{ "_id": stream.FullDocument["_id"]}
    			update := bson.M{ "$set": stream.FullDocument}
    			_, err := slaveClient.Collection(stream.Ns.Collection).UpdateOne(context.TODO(), filter, update)
    			if err != nil { 
    				log.Println("更新失败：", err)
    			}
    		case OperationTypeReplace:
    			filter := bson.M{ "_id": stream.FullDocument["_id"]}
    			_, err := slaveClient.Collection(stream.Ns.Collection).ReplaceOne(context.TODO(), filter, stream.FullDocument)
    			if err != nil { 
    				log.Println("替换失败：", err)
    			}
    		}
    	}
    }
    
    func syncInsert(slaveClient *mongo.Database, stream StreamObject) { 
    	defer func() { 
    		_ = recover()
    	}()
    
    	_, err := slaveClient.Collection(stream.Ns.Collection).InsertOne(context.TODO(), stream.FullDocument)
    	if err != nil { 
    		log.Println("插入失败：", err)
    	}
    }

main.go代码：

package main
    
    import (
    	"moneky-data-sync/mongo"
    )
    
    func main() { 
    	mongo.Sync()
    }

[git hub代码地址][git hub]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70]: /images/20221120/f5ea748cc4cf4f01b98f158737f56b1f.png
[git hub]: https://github.com/Lailaimonkey/mongo-sync.git