前言

MongoShake是阿里云以Golang語言編寫的通用平臺型服務工具，它通過讀取MongoDB的Oplog操作日志來復制MongoDB的數據以實現特定需求。

MongoShake還提供了日志數據的訂閱和消費功能，可通過SDK、Kafka、MetaQ等方式的靈活對接，適用于日志訂閱、數據中心同步、Cache異步淘汰等場景。

官方地址：https://github.com/alibaba/MongoShake

使用場景

• MongoDB集群間數據的異步復制，免去業務雙寫開銷（數據災備）；
• MongoDB集群間數據的鏡像備份（當前1.0開源版本支持受限）；
• 日志離線分析；
• 日志訂閱；
• 數據路由，根據業務需求，結合日志訂閱和過濾機制，可以獲取關注的數據，達到數據路由的功能；
• Cache同步。日志分析的結果，知道哪些Cache可以被淘汰，哪些Cache可以進行預加載，反向推動Cache的更新；
• 基于日志的集群監控

基于mongoshake可以衍生出很多的場景，從而提供給開發者在架構設計中更多更靈活、更豐富的選擇

搭建步驟

mongoshake本身的使用不難，這個參考官方的配置即可，主要是根據自身的業務場景做好配置文件中各個參數的配置即可達到目的，下面以mongoshake一個較為常用的場景，即使用mongoshanke完成2個mongodb復制集群之間的數據同步

環境準備

• 兩臺服務器（阿里云、騰訊云服務器或虛擬機）
• 提前規劃相關的端口，并開放相關端口
• 提前下載mongodb以及mongoshake安裝包，兩臺服務器均上傳，本次mongodb版本為：mongodb-linux-x86_64-4.0.10.tgz，mongo-shake-v2.4.6.tar.gz

一、搭建mongodb復制集

選擇其中一臺服務器搭建一個mongodb的單機版復制集群，這里使用端口號區分，啟動的時候啟動多個實例即可（有條件的可以采用3個不同的服務器操作），按照下面的步驟依次執行即可：

1、創建3個目錄

mkdir replications
cd replications
mkdir myrs_27017 
mkdir myrs_27018
mkdir myrs_27019

2、在3個主目錄下，分別創建data和log目錄

mkdir data
cd data
mkdir db

在myrs_27017 ，myrs_27018 ，myrs_27019 3個目錄下分別做同樣的操作，

3、編輯mongod.conf配置文件

#數據保存路徑
dbpath=/usr/local/soft/mongo/replications/myrs_27017/data/db
#日志保存路徑
logpath=/usr/local/soft/mongo/replications/myrs_27017/log/mongod.log
#日志追加寫入
logappend=true
#復制集名稱
replSet=myrs
bind_ip=服務器IP（云服務器建議使用內網IP）
#mongo默認端口
port=27017
#操作日志容量
oplogSize=10000
noprealloc=true
#開啟子進程
fork=true

在另外的myrs_27018 ，myrs_27019兩個目錄下做同樣的操作，只需要分別修改下配置文件中的端口號即可

4、使用下面的命令依次啟動3個不同的mongodb實例

進入mongodb的bin目錄下，依次執行下面的命令

./mongod -f /usr/local/soft/mongo/replications/myrs_27017/mongod.conf

./mongod -f /usr/local/soft/mongo/replications/myrs_27018/mongod.conf

./mongod -f /usr/local/soft/mongo/replications/myrs_27019/mongod.conf

看到下面的信息表名啟動成功

5、創建集群

使用mongo的shell登錄到其中一個實例上，進入bin目錄下，執行下面命令

./mongo --host 服務器IP（建議使用內網IP） --port 27017

6、執行數集群信息的初始化操作

使用下面的命令執行

cfg={ _id:"myrs", members:[ {_id:0,host:'服務器IP:27017',priority:1}, {_id:1,host:'服務器IP:27018',priority:2}, {_id:2,host:'服務器IP:27019',arbiterOnly:true}]};

再使用：rs.initiate(cfg); 命令完成集群初始化

執行完畢后，可以使用： rs.status() 查看集群各個節點信息，打印出的信息太長，就不放截圖了，注意，執行完畢之后，當前節點會出現短暫的 secondary ，但是過一會兒就變成 primary了

7、將另外兩個節點加入集群

在上一步的窗口下依次執行下面的命令，將兩位2個mongo實例加入到集群中

rs.add("服務器IP:27018")  #加入第一個從節點
rs.add("服務器IP:27019")  #加入第二個仲裁節點

8、主節點上創建數據

在上一步的窗口下，即主節點上，給某個庫的某個集合下插入一條數據

登錄從節點，由于是復制集群，主節點上的數據必然會同步到從節點上，我們可以登陸進去查看是否同步成功

./mongo --host 服務器IP --port 27018

可以發現沒有權限，默認情況下，從節點是沒有讀寫權限的，需要做設置才行，可以在上面從節點的窗口執行下面的命令即可

rs.slaveOk()

執行完畢后，再次查詢即可看到主節點上插入的數據了

以上就是基于單機模式下搭建一個偽復制集群的全部過程，將同樣的操作在另外一個服務器上執行即可

在另一個集群上，我們在主節點上創建了下面一個test庫，并在庫下新建了一條數據

二、mongoshake配置

有了上面的兩個復制集群，mongoshake的使用就變得非常簡單了，mongoshake實現數據同步的基本原理是通過監聽mongodb的oplog，解析其中的事件，從而完成數據的讀取與寫入

mongoshake常用的同步包括，增量同步、全量同步、增量+全量同步，下面從列舉了mongoshake內部實現數據同步的業務機制

1、解壓mongoshake包

2、進入解壓后的目錄編輯并配置 collector.conf文件

下面列舉比較常用的幾個配置

#源數據庫地址,集群多個地址中間用逗號分割
mongo_urls = mongodb://源mongoIP:27017,源mongoIP:27018,源mongoIP:27019

# 通道模式。直接數據庫到數據庫
tunnel = direct
# 此處配置通道的地址，格式與mongo_urls對齊?！灸繕送降刂贰?
tunnel.address = mongodb://目標mongoIP:27017,目標mongoIP:27018,目標mongoIP:27019
# all 表示全量+增量，full表示僅全量，incr表示僅增量
sync_mode = all 
# raw是默認的類型，其采用聚合的模式進行寫入和
# 讀取，但是由于攜帶了一些控制信息，所以需要專門用receiver進行解析。
# json以json的格式寫入kafka，便于用戶直接讀取。
# bson以bson二進制的格式寫入kafka。
tunnel.message = raw

其他更多高級配置柯參考官方詳細說明進行了解，比如可以同步到kafka等

3、啟動mongoshake服務

在主目錄下，執行下述命令啟動同步任務，并打印日志信息

./collector.linux -conf=collector.conf -verbose

模擬測試

mongoshake服務啟動完畢后，這時可以去被同步的第一臺服務器上檢查數據是否同步成功

由于我們配置的是全量+增量的方式，因此服務一旦啟動，數據就全部同步過去了

這時候再次去第二臺服務器的test數據庫下再次新增一條數據

再返回到第一臺機器上檢查數據是否成功同步

可以看到，數據成功同步到第一臺機器上了