利用MongoDB中oplog机制实现准实时数据的操作监控 - 主机乐

最近有一个需求是要实时获取到新插入到MongoDB的数据，而插入程序本身已经有一套处理逻辑，所以不方便直接在插入程序里写相关程序，传统的数据库大多自带这种触发器机制，但是Mongo没有相关的函数可以用（也可能我了解的太少了，求纠正），当然还有一点是需要python实现，于是收集整理了一个相应的实现方法。

一、引子

首先可以想到，这种需求其实很像数据库的主从备份机制，从数据库之所以能够同步主库是因为存在某些指标来做控制，我们知道MongoDB虽然没有现成触发器，但是它能够实现主从备份，所以我们就从它的主从备份机制入手。

二、OPLOG

首先，需要以master模式来打开mongod守护，命令行使用–master,或者配置文件增加master键为true。

此时，我们可以在Mongo的系统库local里见到新增的collection——oplog，此时oplog.$main里就会存储进oplog信息，如果此时还有充当从数据库的Mongo存在，就会还有一些slaves的信息，由于我们这里并不是主从同步，所以不存在这些集合。

再来看看oplog结构：

三、CODE

在Github上找到了别人的实现方式，不过它的函数库太老旧，所以在他的基础上进行修改。

Github地址：https://github.com/RedBeard0531/mongo-oplog-watcher

mongo_oplog_watcher.py如下：

#!/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 + '.' + collectionelif db is not None:self._ns_filter = re.compile(r'^%s\.' % db)else:self._ns_filter = Noneself.poll_time = poll_timeself.connection = connection or pymongo.Connection()if start_now:self.start()@staticmethoddef __get_id(op):id = Noneo2 = op.get('o2')if o2 is not None:id = o2.get('_id')if id is None:id = op['o'].get('_id')return iddef 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:breakexcept 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):passdef insert(self, ns, ts, id, obj, raw, **kw):passdef update(self, ns, ts, id, mod, raw, **kw):passdef delete(self, ns, ts, id, raw, **kw):passdef command(self, ns, ts, cmd, raw, **kw):passdef db_declare(self, ns, ts, **kw):passclass OplogPrinter(OplogWatcher):def all(self, **kw):pprint (kw)print #newlineif __name__ == '__main__':OplogPrinter()

首先是实现一个数据库的初始化，设定一个延迟时间（准实时）：

self.poll_time = poll_timeself.connection = connection or pymongo.MongoClient()

主要的函数是start() ,实现一个时间的比对并进行相应字段的处理：

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数据库操作监控器，下一步就可以配合其他操作来对新入库的程序进行相应处理。