mysql做了主从同步之后，偶尔出现过几次主从同步报错或延迟，由于没有任何监控和报警机制，只有在应用程序报错的时候才能发现数据同步出问题了。所以写了个shell脚本用来检测mysql数据库的同步状态
#!/bin/bash
#monitor_mysql_slave status
USERNAME=xx
#如果密码中包含特殊字符，可以加引号，避免转义
PASSWORD='xx'
SLAVE_HOST=ip
SLAVE_PORT=port
MYSQL="mysql -u$USERNAME -p$PASSWORD"
SLAVE="$MYSQL -h $SLAVE_HOST -P $SLAVE_PORT"
EMAIL_ADDR=zhangsan@qq.com
datetime=`date +"%Y-%m-%d %H:%M:%S"`
#查看从服务器状态
MySQL_Status=`$SLAVE -e "SHOW SLAVE STATUS\G"  | grep -E "Running|Seconds_Behind_Master"|head -n3 `
#从服务器的二个进程信息
Slave_IO_Running=`echo $MySQL_Status | grep Slave_IO_Running |awk '{print $2}'`
Slave_SQL_Running=`echo $MySQL_Status |grep Slave_SQL_Running |awk '{print $4}'`
Seconds_Behind_Master=`echo $MySQL_Status |grep Seconds_Behind_Master |awk '{print $6}'`
if [ "$Slave_IO_Running" = "Yes" -a "$Slave_SQL_Running" = "Yes" -a "$Seconds_Behind_Master" -lt "10"]
then
echo "Replication OK"
else
SUBJECT="ERRORS in replication"
BODY="$datetime $SLAVE_HOST  MySQL slave replication error"
echo $SUBJECT
echo $BODY
#发送报警邮件
echo $BODY | mail -s "$SUBJECT" -t $EMAIL_ADDR
fi
编写好了之后，保存。授予可执行权限 chmod +x 文件名 ，然后执行即可。如果有错误，可使用sh -x 文件名 进行调试。
也可将该脚本加入到定时任务中，定时执行
使用crontab -e 命令，将下方的命令加入到定时任务中
定时任务每15分钟执行一次
*/15 * * * * /usr/local/xx.sh
重启crontab
这是个简单的检测MySQL同步状态的脚本，而且单纯的依靠 Slave_IO_Running和Slave_SQL_Running，Seconds_Behind_Master的值来判断主从同步是否延迟是不准确的。推荐使用专门的工具进行监控，比如percona-toolkit 等

需求：
工作中的业务数据，目前分别存储在美国的服务器和中国的服务器(mysql)，需要把对应的表实施数据同步。

解决方案canal(源码java)
这个工具还是挺强大的，主要实现其实类似于mysql的主从同步机制，监控数据库的bin-log日志从而达到数据同步。

这里简单介绍下它可做的同步流程：

一：既然说了是监控bin-log日志而达到的同步，肯定需要mysql开启bin-log，一般如果是用阿里云服务器和别的商家其实是默认支持的，不需要调整配置。

二：搭建canal,admin为web控制面板(里面可调整配置文件),server部署。两个服务均可使用docker部署。

三：基本操作搭建完成后，可选择合适的同步中间界，目前支持的消息队列为：kafka,rocketMq，另外支持Elasticsearch, 或者直接使用tcp连接来进行消费都是可行的,我们使用的是rocketMq。因为本身使用的消息队列就是rocketMq，所以也就把他当作通道了。

四：直接就讲rocketMq了，和kafka肯定是接近一致的，（tcp消费其实就是检测从节点收到的bin-log信息）.

这里讲下他的简单过程吧，中国服务器cn，美国服务器us,首先cnnal模拟us数据库的从库，从而让us数据库以为canal这边是自己的一个节点，当每次产生bin-log，都会往从节点进行发送(主从同步机制百度一大片),当canal拿到数据后，它自己也会做偏移值记录，再把对用的log发送到rocketMq,客户端在进行消费。

五：客户端消费的时候还是比较麻烦，当时没有直接拿到sql语句，只有对应的数据和对应的操作，相当于客户端这边还写了一个封装方法来进行对log的消费，从而新增到同步的数据库当中。

目前双A数据库同步比较成熟的方案，这个canal算是扛把子了吧，阿里云好歹也是持续了10年在用（在github上自己说的）,不过这中间其实还有很多隐藏的隐患。比如：
1.阿里云的rocketMq建议默认存储三天，（这个其实可以自己也存一份数据，不过不建议）

2.canal挂了：这个好说，他自己记录了偏移值，重启后会从偏移值开始，只要自己别乱删记录，没啥问题。
2.1.消费记录失败时，rocketMq会出现异常情况.

消费情况失败会有两种处理:

处理一:返回失败，这样就会一直重复消费，一直失败，从而也会导致原本bin-log是同步的，到了消费这段就成了异步消费，成了无序的了，（之后的rocketMq记录也会消费）。

处理二:返回成功，那这个就有点猛了，失败了还返回成功，那就得自己去记录失败的值，在用一套体系来保证数据的同步。

还有一些问题就不一一列举了，虽然肯定都是有解决方案的，不过增大了程序的复杂度，加大了开发成本。这个时候就去观察有没有更好的解决方案。

后面发现了阿里封装canal，用java写了一个工具，名叫：otter

不得了，这个东西毕竟是阿里封装好的canal，还是比较成熟的，唯一的区别就是 otter智能 大大减少开发成本， canal也不是麻烦，肯定关于所有的风险处理都需要有保障，因为数据同步的准确性还是挺重要的，不过个人觉得otter过于智能如果中间有什么想要做特别处理的话，肯定直接用canal会更方便。

这个时候老大基于开发成本已经稳定性的考虑，选择了otter，果不其然之前花了两三天把canal弄好，算是白费了，不过不要紧安慰自己得有一套，起码熟悉了cannal的机制了。

otter：
在讲这个之前，再说一下双A数据库Id的解决方案，其实就是用的分布式数据库的解决方案，因为有并发的缘故，我们不能让Id重复，这个时候使用了redis做一个id存储，选择cn数据库存储奇数id，

us数据库存储偶数id，如果中途还有别的国家在部署服务器，那就按照余数来进行划分id.

回到otter，otter使用了 有。zk进行负载， node来进行操作 包括管理canal等。 同步可选择 mysql oracle。

好，下面就是一系列配置，部署好之后，大概最后的情况就是 美国服务器 部署了两个 node ，中国服务器 部署了 两个node，每个node是数据同步的一个方向，这个时候就成功的把生产测试数据库双向同步完成了，当然这中间有些地方需要配置，比如同步的库名保持一致，这个在同步字段时是需要的，其他的小细节都无伤大雅，github里面的错一般都有。

初次配置的话会有很多的配置细节需要注意，是会有些头疼，个人感觉网上在这方面的资料说明是比较散的，如果有啥疑问，可以评论滴滴。

后面公司需要把mysql数据同步到Elasticsearch，otter不支持，准备继续使用canal了。

前言
最近有一个需求是要实时获取到新插入到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上找到了别人的实现方式，不过它的函数库太老旧，所以在他的基础上进行修改。
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数据库操作监控器，下一步就可以配合其他操作来对新入库的程序进行相应处理。
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作能带来一定的帮助，如果有疑问大家可以留言交流，谢谢大家对脚本之家的支持。

我们选择主主互热备做生产环境MYSQL高可用方案，该方案的优点是在压力大的时候，仍然可以通过设置负载均衡来做业务分发。而且最重要的是配置简单，两份配置都一样，改个IP就可以
就喜欢简单的东西。
目前考虑负债均衡分发可能有同步过程中引起数据不一致的问题，我们使用主主机制做数据即时同步，用KeepAliveD做数据库监控及故障自动切换。
实现：
1.数据库主从设置：
MYSQL安装完成后，mysql的配置修改为：
写道
[mysqld]
datadir=/usr/local/mysql/data
socket=/var/lib/mysql/mysql.sock
max_connection=1000
log-bin=mysql-bin
server-id=2
//这块两台服务器需要设置成不同
binlog-do-db=CU
binlog-do-db=FAE
binlog-do-db=HA
//它是用于主从同步时，给master用的配置参数。来说明需要复制那些库给从库。因为有时主从不是要把全部的DB都复制给从库。
binlog-ignore-db=mysql
//安全起见，无论如何不从主同步系统DB
replicate-do-db=CU
replicate-do-db=FAE
replicate-do-db=HA
//从主读取的库
replicate-ignore-db=mysql
//安全起见，无论如何不从主同步系统DB
log-slave-updates
#slave-skip-errors=all
#auto_increment_increment=2
#auto_increment_offset=2
//不是用负载均衡模式，用的是主主模式，所以这块不需要设置
[mysql.server]
user=mysql
basedir=/usr/local/mysql
[client]
socket=/var/lib/mysql/mysql.sock
配置好之后，在两台机器上分别给从建立账号，登录两台Mysql，执行以下语句：
写道
CHANGE MASTER TO
MASTER_HOST='192.***.***.***',
MASTER_USER='repl_user',
MASTER_PASSWORD='hj3****’；
注意的一点就是，使用sqlyog，一定要使用12.09以上的版本，否者主从同步会出现脏数据，因为sqlyog的老版本在修改数据库的时候，某些时候不会产生binlog，导致这些修改数据不会同步。
2.故障检测和主从切换。
我们使用keepalived和shell脚本搞定。
原理：两台数据库服务器上同时部署Keepalived。Keepalived所有节点会互相通信，互报平安。一旦抢到主的机器的Keepalived没有心跳了，从机器的Keepalived会把主的IP抢过来。Keepalived会定期调用shell脚本检测本机mysql是否可用。一旦本机mysq不可用，Keepalived会通知其他机器的Keepalived抢主的IP来实现故障切换。
Keepalived的配置：
写道
! Configuration File for keepalived
#Keepalived组的名称，各个组之间互相不影响
global_defs {
router_id cu_ha_router
}
#检测MYSQL的实例
vrrp_instance mysql {
state BACKUP # 主也配置为SLAVE
interface eth0 #绑定虚拟IP的网络接口
virtual_router_id 101 #VRRP组名，两个节点的设置必须一样，以指明各个节点属于同一VRRP组
priority 150 #节点的优先级(1-254之间)，备用节点必须比主节点优先级低
nopreempt # 不抢占，注意加上
advert_int 1 #组播信息发送间隔，两个节点设置必须一样
authentication { #设置验证信息，两个节点必须一致
auth_type PASS
auth_pass xmotor
}
#一定要注意这块子网掩码的计算，写错了可能会影响整台的网络访问
virtual_ipaddress {
172.*.*.201/20
}
track_script {
chk_mysql
}
#我们使用的是亚马逊云，不能使用组播模式(advertisement),所以必须使用单播模式。单播模式需要增加以下配置。我们做过测试，单播模式支持三台或以上的机器。
unicast_src_ip 172.31.2.1 #localIp
unicast_peer {
172.31.2.124 #Resource-02
#172.31.0.207 #Management
}
#以下四个脚本分别在机器变成主、从、失败、宕机的时候执行的脚本，目前我们只记录一条日志
notify_master "/home/keepalived/scripts/mysql_be_master.sh"
notify_backup "/home/keepalived/scripts/mysql_be_slave.sh"
notify_fault "/home/keepalived/scripts/mysql_fault.sh"
notify_stop "/home/keepalived/scripts/mysql_stop.sh"
}
#检测MYSQL函数
vrrp_script chk_mysql
{
script "/home/keepalived/scripts/mysql_chk.sh eth0 root root"
interval 2
timeout 2
fall 3
}
vrrp_script chk_redis
{
script "/home/keepalived/scripts/redis_check.sh 6391 eth0"
interval 2
timeout 2
fall 3
}
vrrp_script chk_mongo
{
script "/home/keepalived/scripts/mongo_chk.sh eth0"
interval 2
timeout 2
fall 3
}
vrrp_instance redis {
state BACKUP
interface eth0 #绑定虚拟IP的网络接口
virtual_router_id 102 #VRRP组名，两个节点的设置必须一样，以指明各个节点属于同一VRRP组
priority 100 #节点的优先级(1-254之间)，备用节点必须比主节点优先级低
advert_int 1 #组播信息发送间隔，两个节点设置必须一样
authentication { #设置验证信息，两个节点必须一致
auth_type PASS
auth_pass xmotor
}
virtual_ipaddress {
172.*.*.202/20
}
track_script {
chk_redis
}
unicast_src_ip 172.*.*.1 #localIp
unicast_peer {
172.*.*.124 #Resource-02
#172.*.*.207 #Management
}
notify_master "/home/keepalived/scripts/redis_be_master.sh 6391"
notify_backup "/home/keepalived/scripts/redis_be_slave.sh 172.*.*.* 6391"
notify_fault "/home/keepalived/scripts/redis_fault.sh 6391"
notify_stop "/home/keepalived/scripts/redis_stop.sh 6391"
}
vrrp_instance mongodb {
state BACKUP
interface eth0 #绑定虚拟IP的网络接口
virtual_router_id 103 #VRRP组名，两个节点的设置必须一样，以指明各个节点属于同一VRRP组
priority 100 #节点的优先级(1-254之间)，备用节点必须比主节点优先级低
advert_int 1 #组播信息发送间隔，两个节点设置必须一样
authentication { #设置验证信息，两个节点必须一致
auth_type PASS
auth_pass xmotor
}
unicast_src_ip 172.*.*.1 #localIp
unicast_peer {
172.*.*.124 #Resource-02
#172.*.*.207 #Management
}
virtual_ipaddress {
172.*.*.203/20
}
track_script {
chk_mongo
}
notify_master "/home/keepalived/scripts/mongo_be_master.sh"
notify_backup "/home/keepalived/scripts/mongo_be_slave.sh"
notify_fault "/home/keepalived/scripts/mongo_fault.sh"
notify_stop "/home/keepalived/scripts/mongo_stop.sh"
}
下面是检测mysql的脚本，可用返回0，不可用返回1。
写道
#!/bin/bash
#define mysql variable
NETBOARD=$1
MYSQL_USER=$2
MYSQL_PWD=$3
LOGFILE="/var/log/keepalived/mysql/mysql-check.log"
# 看3306端口是否开这
mysql_alive=`netstat -nl | awk 'NR>2{if ($4 ~ /.*:3306/) {print "Yes";exit 0}}'`
#记录日志
echo "[CHECK]" >> $LOGFILE
date >> $LOGFILE
# 取出网关的IP地址
gate=`route -n | grep $NETBOARD | grep UG | awk '{print $2}'`
# 查看网关是否通畅
ping -w 1 -c 1 $gate >/dev/null
ret=$?
#连不上网关，记录日志，返回1
if [ $ret != 0 ];then
echo "Network Gate is not connected" >> $LOGFILE 2>&1
exit 1
fi
Slave_IO_Running=""
Slave_SQL_Running=""
# 如果MYSQL端口是通的，登录mysql，执行mysql语句，检查mysql状态
if [ "$mysql_alive" == "Yes" ];then
Slave_IO_Running=`mysql -u${MYSQL_USER} -p${MYSQL_PWD} -e"show slave status\G" | grep "Slave_IO_Running:" | awk '{print $2}'`
Slave_SQL_Running=`mysql -u${MYSQL_USER} -p${MYSQL_PWD} -e"show slave status\G" | grep "Slave_SQL_Running:" | awk '{print $2}'`
if [ "$Slave_IO_Running" != "" -a "$Slave_SQL_Running" != "" ];then
echo "Success, Slave_IO_Running = $Slave_IO_Running, Slave_SQL_Running = $Slave_SQL_Running" >> $LOGFILE 2>&1
exit 0
else
mysql -u${MYSQL_USER} -p${MYSQL_PWD} -e"show slave status\G"
getSlaveStatusResult=$?
echo "Failed, Reason: $getSlaveStatusResult" >> $LOGFILE 2>&1
exit 1
fi
else
echo "Mysql Down" >> $LOGFILE 2>&1
exit 1
fi