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

 更新时间:2017年05月10日 14:06:03   作者:零の杂货铺  
MongoDB 的Replication是通过一个日志来存储写操作的,这个日志就叫做oplog,而下面这篇文章主要给大家介绍了利用MongoDB中oplog机制实现准实时数据的操作监控的相关资料,需要的朋友可以参考借鉴,下面来一起看看吧。

前言

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

一、引子

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

二、OPLOG

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

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

再来看看oplog结构:

"ts" : Timestamp(6417682881216249, 1), 时间戳
"h" : NumberLong(0), 长度
"v" : 2, 
"op" : "n", 操作类型
"ns" : "", 操作的库和集合
"o2" : "_id" update条件
"o" : {} 操作值,即document

这里需要知道op的几种属性:

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如下:

#!/usr/bin/python
import pymongo
import re
import time
from pprint import pprint # pretty printer
from pymongo.errors import AutoReconnect

class 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):
    pass

class OplogPrinter(OplogWatcher):
  def all(self, **kw):
    pprint (kw)
    print #newline

if __name__ == '__main__':
  OplogPrinter()

首先是实现一个数据库的初始化,设定一个延迟时间(准实时):

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

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流,谢谢大家对脚本之家的支持。

相关文章

  • MongoDB设置登录账号、密码及权限的详细过程

    MongoDB设置登录账号、密码及权限的详细过程

    这篇文章主要给大家介绍了关于MongoDB设置登录账号、密码及权限的详细过程,文中通过代码以及图文介绍的非常详细,对大家学习或者使用MongoDB具有一定的参考借鉴价值,需要的朋友可以参考下
    2023-09-09
  • Windows下MongoDB的下载安装、环境配置教程图解

    Windows下MongoDB的下载安装、环境配置教程图解

    这篇文章主要介绍了Windows下MongoDB的下载安装、环境配置教程详解,本文给大家介绍的非常详细,具有一定的参考借鉴价值 ,需要的朋友可以参考下
    2019-06-06
  • 在MongoDB中实现大于小于查询功能

    在MongoDB中实现大于小于查询功能

    MongoDB是一个高性能、开源、无模式的文档型数据库,它使用BSON(Binary JSON)作为存储格式,支持丰富的查询语法,本文将详细介绍如何在MongoDB中实现大于小于查询,文中有详细的代码示例供大家参考,需要的朋友可以参考下
    2024-08-08
  • C#实现窗体中的各个控件同比自动放缩大小

    C#实现窗体中的各个控件同比自动放缩大小

    这篇文章主要介绍了C#实现窗体中的各个控件同比自动放缩大小,实现方式主要是利用panel控件为主题,对于每个控件的大小位置和字体这几个属性进行记录,然后根据窗体改变的大小同时放缩,需要的朋友可以参考下
    2014-10-10
  • MongoDB创建和查询视图的方式

    MongoDB创建和查询视图的方式

    本文整理mongodb的官方文档,介绍mongodb的视图创建和查询,在Mongodb中,允许使用两种方式来创建视图,本文结合实例代码给大家介绍的非常详细,感兴趣的朋友一起看看吧
    2023-12-12
  • MongoDB的分片集群基本配置教程

    MongoDB的分片集群基本配置教程

    MongoDB拥有经典的Sharding架构能将数据分散存储在数个服务器上以作集群,这里我们就来看一下MongoDB的分片集群基本配置教程:
    2016-07-07
  • MongoDB使用profile分析慢查询的步骤

    MongoDB使用profile分析慢查询的步骤

    这篇文章主要介绍了MongoDB profile分析慢查询的示例,帮助大家更好的理解和学习使用MongoDB数据库,感兴趣的朋友可以了解下
    2021-04-04
  • MongoDB各种查询操作详解

    MongoDB各种查询操作详解

    这篇文章主要介绍了MongoDB各种查询操作详解,包括比较查询、关联查询、数组查询等,需要的朋友可以参考下
    2014-03-03
  • 分享MongoDB修改oplog大小的4种方法

    分享MongoDB修改oplog大小的4种方法

    这篇文章主要介绍了分享MongoDB修改oplog大小的4种方法,文章基于MongoDB修改oplog大小展开其方法的介绍,需要的小伙伴可以参考一下
    2022-04-04
  • 基于Mongodb分布式锁解决定时任务并发执行问题

    基于Mongodb分布式锁解决定时任务并发执行问题

    这篇文章主要介绍了基于Mongodb分布式锁解决定时任务并发执行问题,网上有很多分布式锁的实现方案,基于redis、zk、等有很多,但是我的就是一个用了mysql和mongo的小应用,本文给大家详细讲解,需要的朋友可以参考下
    2023-04-04

最新评论