SQL 入库流程

• 服务器与 MySQL 建立连接
• 依次经过 MySQL 服务器内存中 Server 层的分析器、优化器、执行器
• 执行器根据执行计划操作 InnoDB 引擎
• InnoDB 从磁盘数据文件中将 data 读到缓冲池中
• 修改之前，会写入 undo log 将 data 存起来
• 然后将缓冲池中的 data 改成 new_data
• 写入一条 redo log 将修改后的 new_data 存起来
• 写入一条 binlog 将修改后的值 new_data 存起来
• 后台 IO 线程将缓冲池中被修改的值刷入磁盘

一、Buffer Pool

• InnoDB 里的重要结构，一块用于缓存 MySQL 磁盘数据的内存空间
• 没有 Buffer Pool 每次数据修改都要进行磁盘 IO 操作，有了 Buffer Pool 就可以将磁盘 IO 转换成内存操作，节省时间提高效率
• 如果断电，数据会全部丢失，这就需要用到 redo log、undo log、binlog

二、redo log

redo log 就是重做日志的意思

• 保证数据不丢失
  • 再修改之后，先将修改的数据记录到磁盘上的 redo log 中，就算断电后 Buffer Pool 的数据丢失，也可以从 redo log 中恢复
• WAL（Write-ahead loggin）预写式日志
  • 先预写日志后再将数据刷盘的机制
  • redo log 是磁盘顺序写，数据刷盘是随机写，顺序写比随机写效率高
• redo log buffer
  • 磁盘顺序写的效率已经很高效了，但是和内存操作还是有一定的差距，添加一个内存 buffer 可以优化
  • MySQL 是运行在操作系统上的，MySQL 挂了 Buffer Pool 会丢失，这时候 OS 的 caceh 没有丢失可以恢复，如果操作系统也挂了，OS 的 cache 也会丢失，天意不可违，该咋办咋办
• 落盘机制
  • innodb_flush_log_at_trx_commit = 1：实时写，实时刷
    • 每次事务提交之前，每次都会将数据从 redo log 刷到磁盘中
    • 效率最低，丢数据风险也低
  • innodb_flush_log_at_trx_commit = 0：延迟写，延迟刷
    • 每次事务提交时，只写数据到 redo log buffer 中，然后让后台线程定时将数据刷盘
    • 效率最高，丢数据风险最高
  • innodb_flush_log_at_trx_commit = 2：实时写，延迟刷
    • 每次事务提交之前，redo log 写到 OS cache 中
    • 效率比较高，丢失数据风险比较低，只要操作系统不挂，基本不会丢数据，推荐

三、undo log

• InnoDB 支持事务，事务可以回滚
• 在记录修改之前的数据过程，叫做记录 undo log
• undo 撤销、回滚，undo log 主要作用就是回滚数据
• undo log 默认存在全局表空间里面，可以简单理解为 undo log 也是记录在一个 MySQL 表里面，插入一条 undo log 和插入普通数据是类似的，也需要写 redo log

四、binlog

• redo log 记录的是修改之后的数据，提供崩溃恢复的能力
• undo log 记录修改之前的数据，提供回滚能力
• binlog 记录的是修改之后的数据，用于归档
• binlog sync_binlog 刷盘策略
  • sync_binlog = 0：每次提交事务前将 binlog 写入 OS cache，由操作系统控制刷盘
  • sync_binlog = 1：采用同步写磁盘的方式来写 binlog，不使用 OS cache
  • sync_binlog = N：每进行 n 次事务提交之后，调用一次 fsync 将 OS cache 中的 binlog 强制刷到磁盘
• redo log 和 binlog 区别
  • binlog 是逻辑日志，记录的是对哪一个表的哪一行做了什么修改
  • redo log 是物理日志，记录的是对哪个数据页中的哪个记录做了什么修改
  • binlog 是追加写，redo log 是循环写，日志文件有固定大小，会覆盖之前的数据
  • binlog 是 Server 层的日志，redo log 是 InnoDB 的日志

五、总结

• Buffer Pool 是 MySQL 进程管理的一块内存空间，有减少磁盘 IO 次数的作用
• redo log 是 InnoDB 存储引擎的一种日志，主要是崩溃恢复，innodb_flush_log_at_trx_commit 控制三种刷盘策略，推荐 2
• undo log 是 InnoDB 存储引擎的一种日志，主要作用就是回滚
• binlog 是 MySQL Server 层的一种日志，主要作用是归档
• MySQL 挂了有两种情况
  • 操作系统挂了，MySQL 进程跟着挂
  • 操作系统没挂，MySQL 进程挂了

六、Java 监控 MySQL binlog

6.1 环境准备

• 查看 MySQL 是否开启 binlog
  • show variables like ‘log_bin’;
  • 默认是不开启的，会报 ERROR 1381 - You are not using binary loggin

• 开启 binlog
  • 修改 my.cnf（有些是 my.ini）

• 重启 MySQL

6.2 Java 代码

• 引入依赖

<dependency>
    <groupId>com.github.shyiko</groupId>
    <artifactId>mysql-binlog-connector-java</artifactId>
    <version>0.21.0</version>
</dependency>

• MySQL 消息类

public class MySQLRecord {

    private String database;
    private String table;
    private final List<UpdateRecord> records = new LinkedList<>();

    public MySQLRecord() {
    }

    public MySQLRecord(String database, String table) {
        this.database = database;
        this.table = table;
    }

    @Override
    public String toString() {
        return "{\"database\"=\"" + database + "\"," +
                "\"table\"=" + table + "\"," +
                "\"records\"=" + records + "}";
    }

    public void setRecord(UpdateRecord record) {
        records.add(record);
    }

    public void setRecords(List<UpdateRecord> records) {
        this.records.addAll(records);
    }

    public void setDatabase(String database) {
        this.database = database;
    }

    public void setTable(String table) {
        this.table = table;
    }

    public static class UpdateRecord {
        public String before;
        public String after;

        public UpdateRecord(String before, String after) {
            this.before = before;
            this.after = after;
        }

        @Override
        public String toString() {
            return "{\"before\"=\"" + before + "\"," +
                    "\"after\"=" + after + "\"}";
        }
    }

}

• 监控执行类

public class mysql {
    private static final Map<Long, HashMap<String, String>> dbMap = new HashMap<>();

    public static void main(String[] args) {

        BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "root", "123456");
        client.setServerId(1);

        client.registerEventListener(event -> {
            EventData data = event.getData();
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                if (!dbMap.containsKey(tableMapEventData.getTableId())) {
                    dbMap.put(tableMapEventData.getTableId(), new HashMap<String, String>(){{
                        put("database",tableMapEventData.getDatabase());
                        put("table", tableMapEventData.getTable());
                    }});
                }
            }
            if (data instanceof UpdateRowsEventData) {
                format(((UpdateRowsEventData) data).getTableId(), 0, ((UpdateRowsEventData) data).getRows());
            } else if (data instanceof WriteRowsEventData) {
                format(((WriteRowsEventData) data).getTableId(), 1, ((WriteRowsEventData) data).getRows());
            } else if (data instanceof DeleteRowsEventData) {
                format(((DeleteRowsEventData) data).getTableId(), -1, ((DeleteRowsEventData) data).getRows());
            }
        });

        try {
            client.connect();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 0 update, 1 insert, -1 delete
    public static void format(Long tableId, int type,  List<?> data) {
        MySQLRecord records = new MySQLRecord(dbMap.get(tableId).get("database"), dbMap.get(tableId).get("table"));
        for (Object row : data) {
            records.setRecord(new MySQLRecord.UpdateRecord(
                    type == 0 ? Arrays.toString((Object[]) ((Map.Entry<?, ?>) row).getKey()) :
                            type == 1 ? null : Arrays.toString((Object[]) row),
                    type == 0 ? Arrays.toString((Object[]) ((Map.Entry<?, ?>) row).getValue()) :
                            type == 1 ? Arrays.toString((Object[]) row) : null));
        }
        System.out.println(records);
    }

}