mysql集群

docker 搭建mysql主从

1.先建立数据存放目录（/Users/wangyanyang/docker_mysql_info）

####master 
[mysqld]
server_id = 1
log-bin= mysql-bin
read-only=0
replicate-ignore-db=mysql
replicate-ignore-db=sys
replicate-ignore-db=information_schema
replicate-ignore-db=performance_schema
!includedir /etc/mysql/conf.d/
!includedir /etc/mysql/mysql.conf.d/
####slave
[mysqld]
server_id = 2
log-bin= mysql-bin
read-only=1
replicate-ignore-db=mysql
replicate-ignore-db=sys
replicate-ignore-db=information_schema
replicate-ignore-db=performance_schema
!includedir /etc/mysql/conf.d/
!includedir /etc/mysql/mysql.conf.d

2.构建两个mysql容器

master
docker run -d -e MYSQL_ROOT_PASSWORD=root --name mastermysql --restart always 
-v /Users/wangyanyang/docker_mysql_info/master/data:/var/lib/mysql 
-v /Users/wangyanyang/docker_mysql_info/master/conf:/etc/mysql/conf.d 
-p 3307:3306 mysql
slave
docker run -d -e MYSQL_ROOT_PASSWORD=root --name slavemysql1 --restart always 
-v /Users/wangyanyang/docker_mysql_info/slave1/data:/var/lib/mysql 
-v /Users/wangyanyang/docker_mysql_info/slave1/conf:/etc/mysql/conf.d 
-p 3308:3306 mysql

3.配置master

进入容器
docker exec -it ea159e960099 /bin/bash
登录数据库
mysql -u root -p

创建用户, identified with 'mysql_native_password' 表示使用该用户的密码加密方式。
create user 'reader'@'%' identified with 'mysql_native_password' by 'reader';
赋予用户相应的权限.
grant all privileges on *.* to 'reader'@'%';
刷新 MySql 系统权限。
flush privileges;

查看状态，记住File、Position的值.
show master status;

4.配置slave

//设置主库链接
CHANGE MASTER TO MASTER_HOST='10.12.254.39',    -- 主库的 IP 地址
Master_Port=3307,    -- 主库端口
MASTER_USER='reader',    --指定用户名
MASTER_PASSWORD='reader',    -- 密码
MASTER_LOG_FILE='mysql-bin.000003',    -- 从库需要从主库上备份的文件名称
MASTER_LOG_POS=0;    -- 从库开始备份的位置，此时设置为 0，也就是从初始位置开始备份

//启动从库同步
start slave;

5.测试主从同步即可。
详见: https://learnku.com/articles/35374

主从复制原理

master数据写⼊，更新binlog
master创建⼀个dump线程向slave推送binlog
slave连接到master的时候，会创建⼀个IO线程接收binlog，并记录到relay log中继⽇志中
slave再开启⼀个sql线程读取relay log事件并在slave执⾏，完成同步
slave记录⾃⼰的binglog

主从延迟问题

原因

随机重放: 从库中sql线程读取从relay.log日志中读取数据, 并写入磁盘过程是随机的, 效率不是太高, 会存在主从延迟的问题。
主库并发高: 某一时刻，大量写请求打到主库上，意味着要不断对 binlog 进行写入，此时从库中的 SQL 线程就会应接不暇，自然会产生主从延迟。
锁等待: 对于 SQL单线程来说，当遇到阻塞时就会一直等待，直到执行成功才会继续进行。如果某一时刻从库因为查询产生了锁等待的情况，此时只有当前的操作执行完成后才会进行下面的操作，产生了主从延迟的情况。

解决方案

并行复制: sql线程读取relay.log日志并写入磁盘, 采用多线程的方式进行. (mysql5.6之后提供)

还可以通过控制主库并发, 强制读主库来解决延迟。

读写分离(mybatis插件简单实现)

1.定义多数据源(一主一从)
2.实现AbstractRoutingDataSource来动态的使用数据源
3.实现mybatis plugin来动态的选择数据源

引入多数据源

实现AbstractRoutingDataSource

spring提供了AbstractRoutingDataSource, 提供了动态选择数据源的功能。替换原有的单一数据源后，即可实现读写分离


@Component("dynamicDataSource")
public class DynamicDataSource extends AbstractRoutingDataSource {

    @Resource(name = "normalDataSource")
    private DataSource normalDataSource;

    @Resource(name = "writeDataSource")
    private DataSource writeDataSource;

    @Resource(name = "readDataSource")
    private DataSource readDataSource;

    // 调用数据源时：determineCurrentLookupKey()方法返回了需要使用的数据源对应的key
    @Override
    protected Object determineCurrentLookupKey() {
        DynamicDataSourceGlobal dynamicDataSourceGlobal = DynamicDataSourceHolder.getDataSource();
        if (dynamicDataSourceGlobal == null
                || dynamicDataSourceGlobal == DynamicDataSourceGlobal.NORMAL) {
            return DynamicDataSourceGlobal.NORMAL.name();
        }
        if (dynamicDataSourceGlobal == DynamicDataSourceGlobal.WRITE) {
            return DynamicDataSourceGlobal.WRITE.name();
        }
        return DynamicDataSourceGlobal.READ.name();
    }

    // AbstractRoutingDataSource内部维护了一个Map<Object, Object>的Map 
    // 在初始化过程中，我们将多个数据源加入到这个map 
    @Override
    public void afterPropertiesSet() {
        //将所有数据源放到Map中
        Map<Object, Object> targetDataSources = new HashMap<>(4);
        targetDataSources.put(DynamicDataSourceGlobal.NORMAL.name(), normalDataSource);
        targetDataSources.put(DynamicDataSourceGlobal.WRITE.name(), writeDataSource);
        targetDataSources.put(DynamicDataSourceGlobal.READ.name(), readDataSource);
        //动态数据源
        //设置默认数据源
        setDefaultTargetDataSource(normalDataSource);
        //设置可通过路由key,切换的数据源Map集
        setTargetDataSources(targetDataSources);
        super.afterPropertiesSet();
    }
}

当前线程需要使用的数据源对应的key，是在DynamicDataSourceHolder类中维护

实现mybatis plugin

当前线程使用的数据源对应的key, 这个key需要在mybatis plugin根据sql类型来确定。

package com.dmall.spring;

import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.executor.Executor;
import org.apache.ibatis.executor.keygen.SelectKeyGenerator;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.mapping.SqlCommandType;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.session.ResultHandler;
import org.apache.ibatis.session.RowBounds;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;


@Intercepts({
        @Signature(type = Executor.class, method = "update", args = {
                MappedStatement.class, Object.class}),
        @Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class}),
        @Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class})
})
public class DynamicPlugin implements Interceptor {
    private static final String REGEX = ".*insert\\u0020.*|.*delete\\u0020.*|.*update\\u0020.*";
    private static final Map<String, DynamicDataSourceGlobal> cacheMap = new ConcurrentHashMap<>();

    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();
        if (!synchronizationActive) {
            Object[] objects = invocation.getArgs();
            MappedStatement ms = (MappedStatement) objects[0];
            DynamicDataSourceGlobal dynamicDataSourceGlobal = null;
            if ((dynamicDataSourceGlobal = cacheMap.get(ms.getId())) == null) {
                //读方法
                if (ms.getSqlCommandType().equals(SqlCommandType.SELECT)) {
                    //!selectKey 为自增id查询主键(SELECT LAST_INSERT_ID() )方法，使用主库
                    if (ms.getId().contains(SelectKeyGenerator.SELECT_KEY_SUFFIX)) {
                        dynamicDataSourceGlobal = DynamicDataSourceGlobal.WRITE;
                    } else {
                        BoundSql boundSql = ms.getSqlSource().getBoundSql(objects[1]);
                        String sql = boundSql.getSql().toLowerCase(Locale.CHINA).replaceAll("[\\t\\n\\r]", " ");
                        if (sql.matches(REGEX)) {
                            dynamicDataSourceGlobal = DynamicDataSourceGlobal.WRITE;
                        } else {
                            dynamicDataSourceGlobal = DynamicDataSourceGlobal.READ;
                        }
                    }
                } else {
                    dynamicDataSourceGlobal = DynamicDataSourceGlobal.WRITE;
                }
                System.out.println("设置方法[{ " + ms.getId() + "}] use [{" + dynamicDataSourceGlobal.name() + "}] Strategy, SqlCommandType [{" + ms.getSqlCommandType().name() + "}]..");
                cacheMap.put(ms.getId(), dynamicDataSourceGlobal);
            }
            DynamicDataSourceHolder.putDataSource(dynamicDataSourceGlobal);
        }
        return invocation.proceed();
    }

    @Override
    public Object plugin(Object target) {
        if (target instanceof Executor) {
            return Plugin.wrap(target, this);
        } else {
            return target;
        }
    }
    @Override
    public void setProperties(Properties properties) {
        //
    }
}

分库分表

详见 https://www.yuque.com/wangyanyang/qz7fwa/ee92bw

不停机扩容

第⼀阶段: 在线双写,查询⾛⽼库

建⽴好新的库表结构，数据写⼊久库的同时，也写⼊拆分的新库
2. 数据迁移，使⽤数据迁移程序，将旧库中的历史数据迁移到新库
3. 使⽤定时任务，新旧库的数据对⽐，把差异补齐

第⼆阶段: 在线双写,查询⾛新库
完成了历史数据的同步和校验
2. 把对数据的读切换到新库

第三阶段：旧库下线

旧库不再写⼊新的数据
2. 经过⼀段时间，确定旧库没有请求之后，就可以下线⽼库

java知识体系

mysql 高可用