[TOC]
在A、B两地区机房，分别部署了mysql服务。由于业务需求，需要将A地区的数据准实时同步到B地区。
如上述场景，根据目前基础组件支持及业务需求现状，提出了以下方案:
(1) 将源库的 binlog写入mq, 数据同步服务消费mq, 并在目标库执行 (核心方案)
- maxwell将binlog写入消息队列qbus/kafka
- 服务消费mq，进行数据解析，格式过滤，生成sql
- 服务将数据写入到目标库(支持增、删、改)
(2) 周期性的从源库同步数据到目标库 (降级方案)
-
定时轮询，根据表主键，查询出增量的源库数据
-
将增量数据进行数据解析，格式过滤，生成sql
-
服务将数据写入到目标库 (仅支持增)
(3) 周期性的检测两库数据状态
- 服务内部启动时钟周期，进行周期检测
- 目标库数据是否与源库一致, 是否正常同步
基于上述方案，开发了数据同步服务data-sync
#demo # conf/iot.sync.toml #主要部分示例如下: [ssh] [ssh.yushaolong] username = 'yushaolong' board_host = '110.111.112.113:11022' password = "yourpassword" # 与private_path 二选一 (优先) private_path = './conf/empty.private.key' # 与 password 二选一 [strategy] #运行策略 [strategy.sync_mq] #通过binlog写入mq, 消费mq同步 [[strategy.sync_mq.mysql]] name = "cn_to_aws_database" #注意只能使用字母，数字，下划线 kind = "kafka" conf_path = "./conf/kafka.conf" group = "iot_data_sync" topics = ["iot_mysql_binlog_test"] log_path = "./log/consumer.log" [strategy.sync_direct] #定时任务同步 #仅支持追加 append_duration = 60 # 单位s, 同步周期 [strategy.inspect] #巡检模式,不做同步 inspect_duration = 30 #单位:s, 巡检周期, 当与定时任务同时进行时,应大于 append_duration inspect_alarm_count = 5 #巡检5次,数据仍然不一致且[primaryId未曾变化],报警 [[mysql_task]] task_name = "mysql_database_name_prod_to_test" #注意命名需要全局唯一, 只能使用字母,数字,下划线, database = "database_name" data_mode = "PART" #ALL全同步; PART 指定表 include_tables = ["user"] #同步表,data_mode=PART时, 则使用该字段(优先) exclude_tables = [] #不同步表,data_mode=PART时, 则使用该字段 [mysql_task.regular] #过滤规则定义,在data_mode=PART/ALL均生效 [mysql_task.regular.table_name] #表 table_name = "table_name" #表名称 filter_cols = ["col1","col2"] #过滤字段, 生成的sql会过滤相应字段,不进行同步 upsert = true #当false时, 使用insert格式; 当true时,会insert on duplicate update [mysql_task.source] db_host = "110.111.112.113" db_port = 3306 db_username = "database_name" db_password = "database_password" enable_ssh = true #开启ssh隧道模式,直接使用tunnel_host+tunnel_port [mysql_task.source.ssh] ssh_username = "yushaolong" #使用隧道用户 tunnel_host = "127.0.0.1" #打通本地ip tunnel_port = 3306 [mysql_task.target] db_host = "114.115.116.117" db_port = 3307 db_username = "database_name" db_password = "database_password" enable_ssh = false #关闭ssh隧道模式 [mysql_task.target.ssh] [mysql_task.strategy] # 数据库策略 strategy_mode = 4 # 策略模式: 1仅mq; 2仅定时; 4仅巡检; 5 使用mq并巡检; 6 使用定时并巡检; 7 同时mq,定时,巡检 sync_mq_name = "cn_to_aws_database" # 当模式包含mq时, 该字段有效├── conf #源配置 ├── src #源码目录 │ ├── application │ │ └── strategy #策略 │ │ ├── inspect #策略:巡检模式 │ │ ├── syncdirect #策略:从源库直接同步 │ │ └── syncmq #策略:从MQ同步 │ ├── infrastructure │ │ ├── config #解析配置 │ │ ├── context #全局上下文 │ │ ├── dao #数据库映射对象 │ │ ├── drivers #数据库驱动组件 │ │ ├── monitor #监控组件 │ │ ├── mq #消息队列组件 │ │ ├── repo │ │ │ ├── condition #库: 条件管理 │ │ │ ├── format #库: 数据格式 │ │ │ ├── serializer #库: 序列化 │ │ │ └── stream #库: 流 │ │ └── ssh #ssh隧道组件 │ ├── main │ ├── server │ │ └── http #对外http服务 │ └── service #业务处理 │ └── customfilter #业务:定制化过滤 └── tools #shell工具 项目使用了counter,gauge,histogram三项指标来检测服务运行状态。
(1) counter 指标监控项:
- 服务内部panic
- 数据同步状态异常
(2) gauge 指标监控项:
- 当前服务运行的策略模式
- 当前巡检失败次数
(3) histogram指标监控项:
- 进程goroutine数量、gc次数、gc耗时、heap分配、stack分配
- 服务依赖资源的错误、耗时监控
- 策略运行的错误、耗时监控
同时服务暴露了pprof的http端口，可以查看更详细的程序运行状态。
-
批量写优化
-
mongo支持