前言
最近有一個需求是要實時獲取到新插入到MongoDB的數據，而插入程序本身已經有一套處理邏輯，所以不方便直接在插入程序里寫相關程序，傳統的數據庫大多自帶這種觸發器機制，但是Mongo沒有相關的函數可以用（也可能我了解的太少了，求糾正），當然還有一點是需要python實現，于是收集整理了一個相應的實現方法。
一、引子
首先可以想到，這種需求其實很像數據庫的主從備份機制，從數據庫之所以能夠同步主庫是因為存在某些指標來做控制，我們知道MongoDB雖然沒有現成觸發器，但是它能夠實現主從備份，所以我們就從它的主從備份機制入手。
二、OPLOG
首先，需要以master模式來打開mongod守護，命令行使用–master,或者配置文件增加master鍵為true。
此時，我們可以在Mongo的系統庫local里見到新增的collection——oplog，此時oplog.$main里就會存儲進oplog信息，如果此時還有充當從數據庫的Mongo存在，就會還有一些slaves的信息，由于我們這里并不是主從同步，所以不存在這些集合。
再來看看oplog結構：
|
1
2
3
4
5
6
7
|
"ts" : Timestamp(6417682881216249, 1), 時間戳"h" : NumberLong(0), 長度"v" : 2, "op" : "n", 操作類型"ns" : "", 操作的庫和集合"o2" : "_id" update條件"o" : {} 操作值，即document |
這里需要知道op的幾種屬性：
|
1
2
3
4
5
|
insert，'i'update, 'u'remove(delete), 'd'cmd, 'c'noop, 'n' 空操作 |
從上面的信息可以看出，我們只要不斷讀取到ts來做對比，然后根據op即可判斷當前出現的是什么操作，相當于使用程序實現了一個從數據庫的接收端。
三、CODE
在Github上找到了別人的實現方式，不過它的函數庫太老舊，所以在他的基礎上進行修改。
Github地址：https://github.com/RedBeard0531/mongo-oplog-watcher
mongo_oplog_watcher.py如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
#!/usr/bin/pythonimport pymongoimport reimport timefrom pprint import pprint # pretty printerfrom pymongo.errors import AutoReconnectclass OplogWatcher(object): def __init__(self, db=None, collection=None, poll_time=1.0, connection=None, start_now=True): if collection is not None: if db is None: raise ValueError('must specify db if you specify a collection') self._ns_filter = db + '.' + collection elif db is not None: self._ns_filter = re.compile(r'^%s\.' % db) else: self._ns_filter = None self.poll_time = poll_time self.connection = connection or pymongo.Connection() if start_now: self.start() @staticmethod def __get_id(op): id = None o2 = op.get('o2') if o2 is not None: id = o2.get('_id') if id is None: id = op['o'].get('_id') return id def start(self): oplog = self.connection.local['oplog.$main'] ts = oplog.find().sort('$natural', -1)[0]['ts'] while True: if self._ns_filter is None: filter = {} else: filter = {'ns': self._ns_filter} filter['ts'] = {'$gt': ts} try: cursor = oplog.find(filter, tailable=True) while True: for op in cursor: ts = op['ts'] id = self.__get_id(op) self.all_with_noop(ns=op['ns'], ts=ts, op=op['op'], id=id, raw=op) time.sleep(self.poll_time) if not cursor.alive: break except AutoReconnect: time.sleep(self.poll_time) def all_with_noop(self, ns, ts, op, id, raw): if op == 'n': self.noop(ts=ts) else: self.all(ns=ns, ts=ts, op=op, id=id, raw=raw) def all(self, ns, ts, op, id, raw): if op == 'i': self.insert(ns=ns, ts=ts, id=id, obj=raw['o'], raw=raw) elif op == 'u': self.update(ns=ns, ts=ts, id=id, mod=raw['o'], raw=raw) elif op == 'd': self.delete(ns=ns, ts=ts, id=id, raw=raw) elif op == 'c': self.command(ns=ns, ts=ts, cmd=raw['o'], raw=raw) elif op == 'db': self.db_declare(ns=ns, ts=ts, raw=raw) def noop(self, ts): pass def insert(self, ns, ts, id, obj, raw, **kw): pass def update(self, ns, ts, id, mod, raw, **kw): pass def delete(self, ns, ts, id, raw, **kw): pass def command(self, ns, ts, cmd, raw, **kw): pass def db_declare(self, ns, ts, **kw): passclass OplogPrinter(OplogWatcher): def all(self, **kw): pprint (kw) print #newlineif __name__ == '__main__': OplogPrinter() |
首先是實現一個數據庫的初始化，設定一個延遲時間（準實時）：
|
1
2
|
self.poll_time = poll_timeself.connection = connection or pymongo.MongoClient() |
主要的函數是start() ,實現一個時間的比對并進行相應字段的處理：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
def start(self): oplog = self.connection.local['oplog.$main'] #讀取之前提到的庫 ts = oplog.find().sort('$natural', -1)[0]['ts'] #獲取一個時間邊際 while True: if self._ns_filter is None: filter = {} else: filter = {'ns': self._ns_filter} filter['ts'] = {'$gt': ts} try: cursor = oplog.find(filter) #對此時間之后的進行處理 while True: for op in cursor: ts = op['ts'] id = self.__get_id(op) self.all_with_noop(ns=op['ns'], ts=ts, op=op['op'], id=id, raw=op) #可以指定處理插入監控，更新監控或者刪除監控等 time.sleep(self.poll_time) if not cursor.alive: break except AutoReconnect: time.sleep(self.poll_time) |
循環這個start函數，在all_with_noop這里就可以編寫相應的監控處理邏輯。
這樣就可以實現一個簡易的準實時Mongo數據庫操作監控器，下一步就可以配合其他操作來對新入庫的程序進行相應處理。
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作能帶來一定的幫助，如果有疑問大家可以留言交流，謝謝大家對服務器之家的支持。
原文鏈接：http://phantom0301.cc/2017/05/08/MongoOplog/
