主从同步原理

所有数据库同步原理几乎一样，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监听返回信息

1{
2   
3    
4        _id : {
5   
6     // 存储元信息
7            "_data" : <BinData|hex string> // resumeToken，用于断点恢复
8        },
9        "operationType" : "<operation>", // insert, delete, replace, update, drop, rename, dropDatabase, invalidate,部分仅支持4.0后的版本，详情见下
10        "fullDocument" : {
11   
12     <document> }, // 修改后的数据，出现在insert, replace, delete, update的事件中
13        "ns" : {
14   
15     // namespace
16            "db" : "<database>", // 操作库名
17            "coll" : "<collection" // 操作表名
18        },
19        "to" : {
20   
21     // 只在operationType为rename的时候有效，表示改名以后的namespace
22            "db" : "<database>",
23            "coll" : "<collection"
24        },
25        "documentKey" : {
26   
27     "_id" : <value> }, // 相当于o2字段。出现在insert, replace, delete, update事件中。正常只包含_id，对于sharded collection，还包括shard key
28        "updateDescription" : {
29   
30     // 只在operationType为update的时候出现，相当于是增量的修改，而replace是替换
31            "updatedFields" : {
32   
33     <document> }, // 更新的field的值
34            "removedFields" : [ "<field>", ... ] // 删除的field列表
35        },
36        "clusterTime" : <Timestamp>, // 相当于ts字段
37        "txnNumber" : <NumberLong>, // 相当于oplog里面的txnNumber，只在事务里面出现。事务号在一个事务里面单调递增
38        "lsid" : {
39   
40     // 相当于lsid字段，只在事务里面出现。logic session id，请求所在的session的id
41            "id" : <UUID>,
42            "uid" : <BinData>
43        }
44    }

go 代码

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

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

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

initMongo.go代码：

1package mongo
2
3import (
4	"context"
5	"go.mongodb.org/mongo-driver/mongo"
6	"go.mongodb.org/mongo-driver/mongo/options"
7	"log"
8	"time"
9)
10
11func initMasterDBClient() *mongo.Database {
12   
13    
14	var err error
15	clientOptions := options.Client().ApplyURI("mongodb://ip:端口/?connect=direct").SetConnectTimeout(5 * time.Second)
16
17	// 连接到MongoDB
18	client, err := mongo.Connect(context.TODO(), clientOptions)
19	if err != nil {
20   
21    
22		log.Fatal(err)
23	}
24
25	//选择数据库
26	return client.Database("数据库")
27}
28
29func initLiveDBClient() *mongo.Database {
30   
31    
32	var err error
33	clientOptions := options.Client().ApplyURI("mongodb://ip:端口/?connect=direct").SetConnectTimeout(5 * time.Second)
34
35	// 连接到MongoDB
36	client, err := mongo.Connect(context.TODO(), clientOptions)
37	if err != nil {
38   
39    
40		log.Fatal(err)
41	}
42
43	//选择数据库
44	return client.Database("数据库")
45}
46
47func initSlaveDBClient() *mongo.Database {
48   
49    
50	var err error
51	clientOptions := options.Client().ApplyURI("mongodb://ip:端口/?connect=direct").SetConnectTimeout(5 * time.Second)
52
53	// 连接到MongoDB
54	client, err := mongo.Connect(context.TODO(), clientOptions)
55	if err != nil {
56   
57    
58		log.Fatal(err)
59	}
60
61	//选择数据库
62	return client.Database("数据库")
63}
64

sync.go代码：

1package mongo
2
3import (
4	"context"
5	"go.mongodb.org/mongo-driver/bson"
6	"go.mongodb.org/mongo-driver/bson/primitive"
7	"go.mongodb.org/mongo-driver/mongo"
8	"go.mongodb.org/mongo-driver/mongo/options"
9	"log"
10	"time"
11)
12
13type StreamObject struct {
14   
15    
16	Id                *WatchId `bson:"_id"`
17	OperationType     string
18	FullDocument      map[string]interface{
19   
20    }
21	Ns                NS
22	UpdateDescription map[string]interface{
23   
24    }
25	DocumentKey       map[string]interface{
26   
27    }
28}
29
30type NS struct {
31   
32    
33	Database   string `bson:"db"`
34	Collection string `bson:"coll"`
35}
36
37type WatchId struct {
38   
39    
40	Data string `bson:"_data"`
41}
42
43const (
44	OperationTypeInsert  = "insert"
45	OperationTypeDelete  = "delete"
46	OperationTypeUpdate  = "update"
47	OperationTypeReplace = "replace"
48)
49
50var resumeToken bson.Raw
51
52func Sync() {
53   
54    
55	go syncMaster()
56
57	for {
58   
59    
60		time.Sleep(2 * time.Second)
61	}
62}
63
64func syncMaster() {
65   
66    
67	for {
68   
69    
70		//获得主库数据连接
71		client := initMasterDBClient()
72		watch(client)
73	}
74}
75
76func watch(client *mongo.Database) {
77   
78    
79	defer func() {
80   
81    
82		err := recover()
83		if err != nil {
84   
85    
86			log.Printf("同步出现异常: %+v \n", err)
87		}
88	}()
89
90	//设置过滤条件
91	pipeline := mongo.Pipeline{
92   
93    
94		bson.D{
95   
96    {
97   
98    "$match",
99			bson.M{
100   
101    "operationType": bson.M{
102   
103    "$in": bson.A{
104   
105    "insert", "delete", "replace", "update"}}},
106		}},
107	}
108
109	//当前时间前一小时
110	now := time.Now()
111	m, _ := time.ParseDuration("-1h")
112	now = now.Add(m)
113	timestamp := &primitive.Timestamp{
114   
115    
116		T: uint32(now.Unix()),
117		I: 0,
118	}
119
120	//设置监听option
121	opt := options.ChangeStream().SetFullDocument(options.UpdateLookup).SetStartAtOperationTime(timestamp)
122	if resumeToken != nil {
123   
124    
125		opt.SetResumeAfter(resumeToken)
126		opt.SetStartAtOperationTime(nil)
127	}
128
129	//获得watch监听
130	watch, err := client.Watch(context.TODO(), pipeline, opt)
131	if err != nil {
132   
133    
134		log.Fatal("watch监听失败：", err)
135	}
136
137	//获得从库连接
138	slaveClient := initSlaveDBClient()
139
140	for watch.Next(context.TODO()) {
141   
142    
143		var stream StreamObject
144		err = watch.Decode(&stream)
145		if err != nil {
146   
147    
148			log.Println("watch数据失败：", err)
149		}
150
151		log.Println("=============", stream.FullDocument["_id"])
152
153		//保存现在resumeToken
154		resumeToken = watch.ResumeToken()
155
156		switch stream.OperationType {
157   
158    
159		case OperationTypeInsert:
160			syncInsert(slaveClient, stream)
161		case OperationTypeDelete:
162			filter := bson.M{
163   
164    "_id": stream.FullDocument["_id"]}
165			_, err := slaveClient.Collection(stream.Ns.Collection).DeleteOne(context.TODO(), filter)
166			if err != nil {
167   
168    
169				log.Println("删除失败：", err)
170			}
171		case OperationTypeUpdate:
172			filter := bson.M{
173   
174    "_id": stream.FullDocument["_id"]}
175			update := bson.M{
176   
177    "$set": stream.FullDocument}
178			_, err := slaveClient.Collection(stream.Ns.Collection).UpdateOne(context.TODO(), filter, update)
179			if err != nil {
180   
181    
182				log.Println("更新失败：", err)
183			}
184		case OperationTypeReplace:
185			filter := bson.M{
186   
187    "_id": stream.FullDocument["_id"]}
188			_, err := slaveClient.Collection(stream.Ns.Collection).ReplaceOne(context.TODO(), filter, stream.FullDocument)
189			if err != nil {
190   
191    
192				log.Println("替换失败：", err)
193			}
194		}
195	}
196}
197
198func syncInsert(slaveClient *mongo.Database, stream StreamObject) {
199   
200    
201	defer func() {
202   
203    
204		_ = recover()
205	}()
206
207	_, err := slaveClient.Collection(stream.Ns.Collection).InsertOne(context.TODO(), stream.FullDocument)
208	if err != nil {
209   
210    
211		log.Println("插入失败：", err)
212	}
213}

main.go代码：

1package main
2
3import (
4	"moneky-data-sync/mongo"
5)
6
7func main() {
8   
9    
10	mongo.Sync()
11}
12

git hub代码地址

MongoDB手动同步主库（Change Stream）

主从同步原理

Change Stream

Change Events解析

go 代码

评论区

相关推荐