[直播回顾】 直击MySQL源码本色

table_map_event、table_map_event、rows_event,、rows_event

 handle_slave_io()

 |-my_thread_init()                  ← 0) 线程初始化

  |-init_slave_thread()

  |

  |-safe_connect()                    ← 1) 以标准的连接方式连上master

  |-get_master_version_and_clock()       并获取主库的所需信息

  |-get_master_uuid()

  |-io_thread_init_commands()

  |

  |-register_slave_on_master()       ← 2) 把自己注册到master上去

  | |-net_store_data()                    ←    设置数据包

  | |-simple_command()               ←    S把自己的ID、IP、端口、用户名提交给M，用于注册

  | |                                  ←    **上述会发送COM_REGISTER_SLAVE命令**

  |

  |                                 ###1BEGIN while循环中检测io_slave_killed()

  |

  |-request_dump()                  ← 3) 开始请求数据，向master请求binlog数据

  | |-RUN_HOOK()                    ←    调用relay_io->before_request_transmit()

  | |-int2store()                        ←    会根据是否为GTID作区分

  | |-simple_command()              ←    发送dump数据请求

  | |                               ←    **执行COM_BINLOG_DUMP_GTID/COM_BINLOG_DUMP命令**

  |

  |                                     ###2BEGIN while循环中检测io_slave_killed()

  |

  |-read_event()                    ← 4) 读取event并存放到本地relay log中

  | |-cli_safe_read()               ←    等待主库将binlog数据发过来

  |   |-my_net_read()

  |-RUN_HOOK()                      ←    调用relay_io->after_read_event()

  |

  |-queue_event()                   ← 5) 将接收到的event保存在relaylog中

  |-RUN_HOOK()                      ←    调用relay_io->after_queue_event()

  |-flush_master_info()

        bool dispatch_command(THD *thd, const COM_DATA *com_data,

              enum enum_server_command command)

{

  ... ...

  switch (command) {

      ... ...

#ifdef HAVE_REPLICATION

      case COM_REGISTER_SLAVE:    // 注册slave

          if (!register_slave(thd, (uchar*)packet, packet_length))

              my_ok(thd);

          break;

#endif

#ifdef EMBEDDED_LIBRARY

      case COM_BINLOG_DUMP_GTID:

          error= com_binlog_dump_gtid(thd, packet, packet_length);

          break;

      case COM_BINLOG_DUMP:

          error= com_binlog_dump(thd, packet, packet_length);

          break;

#endif

      ... ...

  }

  ... ...

}

dispatch_command()

 |-com_binlog_dump_gtid()              ← COM_BINLOG_DUMP_GTID

 |-com_binlog_dump()                     ← COM_BINLOG_DUMP

   |-kill_zombie_dump_threads()       ← 如果同一个备库注册，会移除跟该备库匹配的binlog dump线程

   |-mysql_binlog_send()               ← 上述两个命令都会执行到此处

     |                                              ← 会打开文件，在指定位置读取文件，将event按照顺序发给备库

     |-Binlog_sender::run()             ← 调用rpl_binlog_sender.cc中的发送

       |-init()

       | |-init_heartbeat_period()      ← 启动心跳

       | |-transmit_start()                ← RUN_HOOK()，binlog_transmit_delegate

       |

       |                              -###BEGIN while()循环，只要没有错误，线程未被杀死，则一直执行

       |-open_binlog_file()

       |-send_binlog()                   ← 发送二进制日志

       | |-send_events()

       |   |-after_send_hook()

       |     |-RUN_HOOK()               ← 调用binlog_transmit->after_send_event()钩子函数

       |

       |-set_last_file()

       |-end_io_cache()

       |-mysql_file_close()

       |-###END

int Rows_log_event::do_apply_event(Relay_log_info const *rli)

{

  ... ...

  table=

    m_table= const_cast<Relay_log_info*>(rli)->m_table_map.get_table(m_table_id);

  ... ...

  if (table)

  {

    ... ...

    if ((m_rows_lookup_algorithm != ROW_LOOKUP_NOT_NEEDED) &&

        !is_any_column_signaled_for_table(table, &m_cols))

    {

      error= HA_ERR_END_OF_FILE;

      goto AFTER_MAIN_EXEC_ROW_LOOP;

    }

    switch (m_rows_lookup_algorithm)

    {

      case ROW_LOOKUP_HASH_SCAN:

        do_apply_row_ptr= &Rows_log_event::do_hash_scan_and_update;

        break;

      case ROW_LOOKUP_INDEX_SCAN:

        do_apply_row_ptr= &Rows_log_event::do_index_scan_and_update;

        break;

      case ROW_LOOKUP_TABLE_SCAN:

        do_apply_row_ptr= &Rows_log_event::do_table_scan_and_update;

        break;

      case ROW_LOOKUP_NOT_NEEDED:

        DBUG_ASSERT(get_general_type_code() == binary_log::WRITE_ROWS_EVENT);

        /* No need to scan for rows, just apply it */

        do_apply_row_ptr= &Rows_log_event::do_apply_row;

        break;

      default:

        DBUG_ASSERT(0);

        error= 1;

        goto AFTER_MAIN_EXEC_ROW_LOOP;

        break;

    }

    do {

      error= (this->*do_apply_row_ptr)(rli);

      if (handle_idempotent_and_ignored_errors(rli, &error))

        break;

/* this advances m_curr_row */

      do_post_row_operations(rli, error);

    } while (!error && (m_curr_row != m_rows_end));

    ... ...

  }

  ... ...

}