六月婷婷综合激情-六月婷婷综合-六月婷婷在线观看-六月婷婷在线-亚洲黄色在线网站-亚洲黄色在线观看网站

明輝手游網中心:是一個免費提供流行視頻軟件教程、在線學習分享的學習平臺!

分享用MongoDB中oplog機制完成數據監控案例

[摘要]MongoDB 的Replication是通過一個日志來存儲寫操作的,這個日志就叫做oplog,而下面這篇文章主要給大家介紹了利用MongoDB中oplog機制實現準實時數據的操作監控的相關資料,需要的朋友可以參考借鑒,下面來一起看看吧。前言最近有一個需求是要實時獲取到新插入到MongoDB的數據...
MongoDB 的Replication是通過一個日志來存儲寫操作的,這個日志就叫做oplog,而下面這篇文章主要給大家介紹了利用MongoDB中oplog機制實現準實時數據的操作監控的相關資料,需要的朋友可以參考借鑒,下面來一起看看吧。

前言

最近有一個需求是要實時獲取到新插入到MongoDB的數據,而插入程序本身已經有一套處理邏輯,所以不方便直接在插入程序里寫相關程序,傳統的數據庫大多自帶這種觸發器機制,但是Mongo沒有相關的函數可以用(也可能我了解的太少了,求糾正),當然還有一點是需要python實現,于是收集整理了一個相應的實現方法。

一、引子

首先可以想到,這種需求其實很像數據庫的主從備份機制,從數據庫之所以能夠同步主庫是因為存在某些指標來做控制,我們知道MongoDB雖然沒有現成觸發器,但是它能夠實現主從備份,所以我們就從它的主從備份機制入手。

二、OPLOG

首先,需要以master模式來打開mongod守護,命令行使用–master,或者配置文件增加master鍵為true。

此時,我們可以在Mongo的系統庫local里見到新增的collection——oplog,此時oplog.$main里就會存儲進oplog信息,如果此時還有充當從數據庫的Mongo存在,就會還有一些slaves的信息,由于我們這里并不是主從同步,所以不存在這些集合。

分享用MongoDB中oplog機制實現數據監控實例

再來看看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地址: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中oplog機制實現數據監控實例的詳細內容,更多請關注php中文網其它相關文章!


學習教程快速掌握從入門到精通的SQL知識。




主站蜘蛛池模板: 婷婷视频网 | 天堂√在线官网 | 亚欧精品一区二区三区 | 中文字幕色在线 | 日本久操| 青青色在线 | 揉女胸视频www网站免费观看 | 亚洲a视频在线观看 | 日本大蕉香蕉大视频在线观看 | 四虎精品成人免费视频 | 日韩一级特黄 | 日本爱爱小视频 | 婷婷中文字幕 | 青青草视频在线免费观看 | 欧美一级视频免费观看 | 永久免费看毛片 | 性香港xxxxx免费视频播放 | 在线免费视频一区二区 | 欧美一卡二卡3卡4卡无卡免费 | 热综合一本伊人久久精品 | 日本爽爽爽爽爽爽在线观看免 | 最新国产视频 | 欧美一级淫片免费播放口 | 天天爽天天狼久久久综合 | 日操夜干 | 四虎国产精品免费久久影院 | 使劲儿操 | 天天干天天在线 | 最近最新中文字幕1页 | 天天做天天爱夜夜爽女人爽宅 | 天天色天天综合网 | 欧美小视频在线观看 | 亚洲产在线精品第一站不卡 | 自偷自偷自亚洲首页精品 | 亚洲国产成人久久综合野外 | 四虎影视网 | 亚洲高清在线播放 | 亚洲观看视频 | 日本免费高清一区 | 香蕉视频网页 | 思思久久99热只有精品 |