主从复制是MySQL源代码中比较重要的一部分代码，一直以来比较经典的复制逻辑是：Master Dump线程 -> Slave I/O线程(Relay log) -> Slave SQL线程。随着MySQL版本的不断演进，在已经比较长的主从复制链路中又出现了下面一些显著的变化：

• MySQL 5.6以后引入的MTS(Multi-Thread Slave)优化将传统的Slave SQL线程分解成了一个Dispatch线程和多个Worker线程；

• MySQL 5.7半同步复制插件中又引入了一个异步ACK线程，这个线程和Dump线程有着密切的关联；

• MySQL 5.6以后GTID的引入，又把主从复制的交互逻辑(传输循环以前)变得更为复杂，主要是GTID的解析和与file:pos复制模式的兼容；

以下是MySQL 5.7.10 Slave I/O线程的逻辑流程的关键点解析：

注意： 
HOOK是MySQL插件架构的基础，为保证总体流程的清晰，没有具体说明每一个HOOK的作用。

一、代码入口

在MySQL源代码sql/rpl_slave.cc文件中：

/**
  Slave IO thread entry point.

  @param arg Pointer to Master_info struct that holds information for
  the IO thread.

  @return Always 0.
*/
extern "C" void *handle_slave_io(void *arg)

二、关键逻辑

步骤01：(HOOK)thread_start

执行注册过的thread_start函数，在半同步插件中就实现了一个HOOK，做一些半同步插件的初始化工作：

if (RUN_HOOK(binlog_relay_io, thread_start, (thd, mi)))
{
  mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
    ER(ER_SLAVE_FATAL_ERROR), "Failed to run 'thread_start' hook");
    goto err;
}

步骤02：mysql_init

其实MySQL Slave的I/O线程和其它mysql cli一样，也是一个MySQL Master的客户端，因此这里首先创建一个用于连接Master的MYSQL句柄：

if (!(mi->mysql = mysql = mysql_init(NULL)))
{
  mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
               ER(ER_SLAVE_FATAL_ERROR), "error in mysql_init()");
  goto err;
}

步骤03：safe_connect

建立到Master的连接，其内部包含首先创建到Master TCP监听的TCP连接，然后进行用户权限验证，非法用户的连接会以失败告终：

// we can get killed during safe_connect
if (!safe_connect(thd, mysql, mi))
{
  sql_print_information("Slave I/O thread%s: connected to master '%s@%s:%d',"
  ...

步骤04：与Master交互

首先是根据Master的版本来创建Format_Description_Event、和Master对比时间戳、设置默认字符集等等；其次是查询对比Master的UUID，这样当故障转移时Master变化，Slave是可以感知到是从不同Master复制数据；最后，将当前Slave的UUID上报给Master, UUID是全局唯一的，Master也可以根据这个特性来做一些检验和错误预防。

注意： 
Format_Description_Event是binlog中非常重要的一种事件类型，主要的作用就是描述当前binlog文件的格式信息，通过在每一个binlog文件中包含这个事件，来兼容不同版本的binlog格式；同时允许用户在线修改一些和binlog相关的参数。解析一个binlog文件，比如mysqlbinlog等，都依赖这个格式描述符。

ret= get_master_version_and_clock(mysql, mi);
if (!ret)
  ret= get_master_uuid(mysql, mi);
if (!ret)
  ret= io_thread_init_commands(mysql, mi);

步骤05：注册Slave到Master

Slave上报自身用户复制的用户名、密码等信息，Master验证其是否具有进行复制的权限，COM_REGISTER_SLAVE命令的协议格式如下：

/*
 Register ourselves with the master.
*/
THD_STAGE_INFO(thd, stage_registering_slave_on_master);
if (register_slave_on_master(mysql, mi, &suppress_warnings))

步骤06：发起Dump请求

根据基于file:pos还是GTID复制模式的不同，Slave在Dump请求中附带位置信息, COM_BINLOG_DUMP_GTID命令的协议格式如下：

THD_STAGE_INFO(thd, stage_requesting_binlog_dump);
if (request_dump(thd, mysql, mi, &suppress_warnings))

步骤07：接收循环

上述步骤完成后，Slave I/O线程进入一个循环，不断接收TCP Socket的数据并做相应的处理，关键步骤分解如下：

1. 读取一个Binlog事件

THD_STAGE_INFO(thd, stage_waiting_for_master_to_send_event);
event_len= read_event(mysql, mi, &suppress_warnings);

2. (HOOK)after_read_event

if (RUN_HOOK(binlog_relay_io, after_read_event,
  (thd, mi,(const char*)mysql->net.read_pos + 1,
  event_len, &event_buf, &event_len)))

3. 写入到relay文件

bool synced= 0;
if (queue_event(mi, event_buf, event_len))
{
  mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
    ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
    "could not queue event from master");
    goto err;
}

其实queue_event是一个非常复杂冗长的处理逻辑，简单归纳为下面几个主要的子步骤：

• 获取relay文件锁并加锁

• 根据Format_Description_Event判断是否开启了binlog校验

• 读取每一个事件，做必要的验证，如果开启校验则执行校验

• 如果是GTID事件，则解析出对应的GTID

• 写relay文件，同样根据设置判断是否是要落盘(fsync)

• 是GTID事件，将GTID事件加入到queued_gtid集合中

• 通知SQL线程有新的binlog事件到达

4. (HOOK)after_queue_event

if (RUN_HOOK(binlog_relay_io, after_queue_event,
   (thd, mi, event_buf, event_len, synced)))
{
  mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
    ER(ER_SLAVE_FATAL_ERROR),
    "Failed to run 'after_queue_event' hook");
    goto err;
}

5. 写Master Info文件

注意： 
这里的写Master Info文件是在锁保护下进行的，另外，即便MySQL默认把这个文件的落盘设置为10000个Log事件fsync一次，但是还是会频繁的发起write 系统调用。

思考： 
如果调整一下binlog文件的格式，或者加入一种新的Log事件类型，这样是不是可以和其它事件合并在一起？

mysql_mutex_lock(&mi->data_lock);
if (flush_master_info(mi, FALSE))
{
  mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
    ER(ER_SLAVE_FATAL_ERROR),
    "Failed to flush master info.");
    mysql_mutex_unlock(&mi->data_lock);
  goto err;
}
mysql_mutex_unlock(&mi->data_lock);

6. relay空间占用检查

如果设置了relay日志的空间占用限制，则进行检查，剩余磁盘空间不足时等待。

if (rli->log_space_limit && rli->log_space_limit <
    rli->log_space_total &&
    !rli->ignore_log_space_limit)
  if (wait_for_relay_log_space(rli))
  {
    sql_print_error("Slave I/O thread aborted while waiting for relay" 
    " log space");
    goto err;
  }

7. 清理内存

清理内存，防止OOM!

思考： 
如何优化，复用内存是可行的，因为大多数复制架构中Slave I/O线程只有一条，预分配+复用可以优化绝大多数场景中的性能开销；唯一不足的是如果偶尔遇见一个特别大的binlog事件，是放大预分配的内存(采用什么策略缩放回去)，还是分片读取？

/*
After event is flushed to relay log file, memory used
by thread's mem_root is not required any more.
Hence adding free_root(thd->mem_root,...) to do the
cleanup, otherwise a long running IO thread can
cause OOM error.
*/
free_root(thd->mem_root, MYF(MY_KEEP_PREALLOC));

三、后记

从上面的分解可以看出：

• queue_event的逻辑有优化的空间，尤其是写relay日志的部分

• 写master info, 无论是写文件还是innodb表，都还可以优化