重要数据结构

Rpl_info 的基类，保存了一些错误信息，如 IO/SQL thread last errorclass Slave_reporting_capability
{// 获取last errorError const& last_error() const { return m_last_error; }}Master_info 、Relay_log_info 的基类，很多重要的锁和信号量都在这里class Rpl_info : public Slave_reporting_capability
{/*为了避免死锁，需要按照以下顺序加锁run_lock, data_lock, relay_log.LOCK_log, relay_log.LOCK_indexrun_lock, sleep_lockrun_lock, info_thd_lockinfo_thd_lock 保护对 info_thd 的操作读操作需要获取 info_thd_lock 或 run_lock写操作需要获取 info_thd_lock 和 run_lockdata_lock : 保护对数据的读写run_lock  : 保护运行状态，变量 slave_running, slave_run_idsleep_lock: 对slave_sleep做互斥，防止同时进入多个slave_sleep*/mysql_mutex_t data_lock, run_lock, sleep_lock, info_thd_lock;/*data_cond: data_lock 保护的数据被修改时发出此信号楼，只有 Relay_log_info 会使用start_cond: sql thread start (start slave 先启动 io thread，后启动 sql thread，sql thread 启动代表全部启动成功)stop_cond: sql/io thread stopsleep_cond: slave被kill，目前只发现5.6中有应用，5.7 中没找到发出此信号量的代码*/mysql_cond_t data_cond, start_cond, stop_cond, sleep_cond;/* 关联 io/sql/worker thread 的 thdMaster_info 对应 io threadRelay_log_info 对应 sql threadSlave_worker 对应 worker thread */THD *info_thd;/* 已初始化 */bool inited;/* 已终止slave */volatile bool abort_slave;/*当前slave运行状态，io thread 即 mi->slave_running有三种状态MYSQL_SLAVE_NOT_RUN         0MYSQL_SLAVE_RUN_NOT_CONNECT 1MYSQL_SLAVE_RUN_CONNECT     2 sql thread 即 mi->slave_running 只有 0 1两种状态*/volatile uint slave_running;/* 用于判断slave thread 是否启动的变量，每次启动时递增1 */volatile ulong slave_run_id;/* repository's handler */Rpl_info_handler *handler;/* 唯一标识一条 info 记录的 id，标识一行或一个文件Master_info 和 Relay_log_info 可以通过 channel 标识唯一Worker_info 需要通过 {id, channel} 标识唯一*/uint internal_id;/* 通道名，多源复制使用 */char channel[CHANNEL_NAME_LENGTH+1];/* 实现了两个虚函数，读写 Rpl_info */virtual bool read_info(Rpl_info_handler *from)= 0;virtual bool write_info(Rpl_info_handler *to)= 0;
}/*Master_info 用户 IO thread主要保存以下信息：连接到Master的用户信息当前 master log name当前 master log offset一些其他控制变量Master_info 读取 master.info repository 初始化，通常是表或文件通过函数 mi_init_info() 进行初始化调用 flush_info() 可以将 master.info 写入磁盘，每次从master读取数据都需要刷盘
*/
class Master_info : public Rpl_info
{/* 前面保存了 user、host 等连接信息 *//* 下面是连接信息的 set/get 函数 *//* 和 master 的连接 */MYSQL* mysql;/* 对应的 Relay_log_info */        Relay_log_info *rli;/* IO 线程复制延迟 */long clock_diff_with_master;/* 心跳间隔 */float heartbeat_period;         // interface with CHANGE MASTER or master.info/* 收到心跳次数 */ulonglong received_heartbeats;  // counter of received heartbeat events/* 上次心跳时间 */time_t last_heartbeat;/* 上次收到event的时间 */time_t last_recv_event_timestamp;/* 忽略复制的server_id */Server_ids *ignore_server_ids;/* master server_id */ulong master_id;/* FORMAT_DESCRIPTION_EVENT前的checksum 算法 */binary_log::enum_binlog_checksum_alg checksum_alg_before_fd;/* 初始化 Master_info, 里面调用read_info() 从 Rpl_info_handler 中读取信息 */int mi_init_info();/* 清理 Master_info，里面会调用 Rpl_info_handler 的 eng_info() */void end_info();/* Master_info 信息落盘，每次从主库读取数据后都会执行 */int flush_info(bool force= FALSE);/*从master收到的 Format_description_log_event 写在 relay log 末尾IO thread 开始时创建，IO thread 结束时销毁IO thread 收到一个 Format_description_log_event 时更新每次rotate时，IO thread 写Format_description_log_event到新relay log每次执行FLUSH LOGS，client 写Format_description_log_event到新relay log*/Format_description_log_event *mi_description_event;/* 最近一个GTID，可能是未完成事务的GTID，用于事务结束时写入 Retrieved_Gtid_Set */Gtid last_gtid_queued;/* 用于判断事务边界 */Transaction_boundary_parser transaction_parser;/* channel lock，以下操作需要持有写锁START SLAVE;STOP SLAVE;CHANGE MASTER;RESET SLAVE;end_slave();*/Checkable_rwlock *m_channel_lock;/* channel 被引用的次数，只有为0时可以删除channel */Atomic_int32 references;
}

/*主要保存以下信息:当前relay log当前relay log offsetmaster log name与上次更新对应的主库日志序列sql thread 其他信息初始化过程和 Master_info 类似以下情况下 relay.info table/file 需要更新：1. relay log file rotated2. SQL thread stopped3. while processing a Xid_log_event4. after a Query_log_event(commit or rollback)5. after processing any statement written to the binary log without a transaction context.并行复制相关代码留作以后分析，本次暂不涉及
*/
class Relay_log_info : public Rpl_info
{/* 备份状态标志位，用于标志是否在语句中 */enum enum_state_flag {/* 在语句中 */IN_STMT,/** Flag counter.  Should always be last */STATE_FLAGS_COUNT};/* 是否复制相同server_id的event，一般是false */bool replicate_same_server_id;/* 正在执行或最后一个执行的GTID，用来填充 performance_schema.replication_applier_status_by_worke 的 last_seen_transaction 列*/Gtid_specification currently_executing_gtid;/* 读取下面变量时，必须受data_lock保护 *//* 当前relay log的文件描述符 */File cur_log_fd;/* reay_log 对象，MYSQL_BIN_LOG类留作以后分析*/MYSQL_BIN_LOG relay_log;/* 主要用于查询log_pos */LOG_INFO linfo;/*cur_log 指向 relay_log.get_log_file() 或者 cache_buf取决于relay_log_file是热日志，还是需要打开冷日志cache_buf 在打开冷日志时适用*/IO_CACHE cache_buf,*cur_log;/* 标识是否正在recovery */bool is_relay_log_recovery;/* 下面的变量可以不加锁读 *//*restart slave 时需要访问临时表。这个变量值在 init/end 时修改SQL thread 只读*/TABLE *save_temporary_tables;/* 对应的 Master_info */Master_info *mi;/* 打开临时表的数量 */Atomic_int32 channel_open_temp_tables;/* 保存 relay_log.get_open_count() */uint32 cur_log_old_open_count;/* init_info() 曾经失败，RESET SLAVE 可以修复错误 */bool error_on_rli_init_info;/* 这里跳过一些 Group replication 相关变量 *//* received gtid set */Gtid_set gtid_set;/* 标识此对象是否属于SQL线程（属于SQL线程为0） */bool rli_fake;/* 标识 retrieved GTID set 是否已被初始化 */bool gtid_retrieved_initialized;/* 上一个错误的GTID */Gtid last_sql_error_gtid;/* 日志空间限制，日志空间总量(用于sys_var:relay_log_space，控制relay log空间) */ulonglong log_space_limit,log_space_total;/* 是否忽略日志空间限制 */bool ignore_log_space_limit;/* 需要清理空间时，SQL线程指示IO线程rotate logs */bool sql_force_rotate_relay;/* 上次主库记录binlog的时间 */time_t last_master_timestamp;/* 上次执行event的时间 */time_t last_exec_event_timestamp;/* 跳过error event */volatile uint32 slave_skip_counter;/* 标记是否需要中断pos_wait，change master 和 reset slave时需要中断 */volatile ulong abort_pos_wait;/* log_space 相关信号量*/mysql_mutex_t log_space_lock;mysql_cond_t log_space_cond;/* 这里有一些 START SLAVE UNTIL 相关变量 *//* 重试事务次数(trans_retries是重试次数上限)，重试事务计数（retried_trans记录重试了多少次） */ulong trans_retries, retried_trans;/* 延迟复制时间CHANGE MASTER TO MASTER_DELAY=X.由data_lock保护， SQL thread 读取SQL thread 运行时该变量不可写*/time_t sql_delay;/* sql_delay 结束时间 */time_t sql_delay_end;/* enum_state_flag 的标志位 */uint32 m_flags;
}

函数分析

mysql_execute_command() 当做入口开始分析，可以看出change_master需要SUPER权限

case SQLCOM_CHANGE_MASTER:{if (check_global_access(thd, SUPER_ACL))goto error;res= change_master_cmd(thd);break;}/*函数具备以下功能更改接收/执行日志的配置/位点purge relay log删除 worker info（并行复制使用）
*//* 分析 change_master 函数会略过一部分逻辑 */
int change_master(THD* thd, Master_info* mi, LEX_MASTER_INFO* lex_mi,bool preserve_logs)
{/*如果SQL thread 和 IO thread已经停止，并且没有指定 relay_log_pos和relay_log_file会purge relay log*/bool need_relay_log_purge= 1;/* 为了修改 mysql.slave_master_info，需要无视read_only和super_read_only */thd->set_skip_readonly_check();/* channel 加读写锁，即将对channel做修改，函数结束时才会释放锁 */mi->channel_wrlock();/*对 mi->run_lock 和 rli->run_lock 加锁防止线程运行状态发生变化*/lock_slave_threads(mi);/* 设置thread_mask，用来标识 IO/SQL thread 的运行状态 */init_thread_mask(&thread_mask, mi, 0);/* 设置auto_position=1需要IO/SQL thread 都不在运行状态，否则报错退出 */if (thread_mask){if (lex_mi->auto_position != LEX_MASTER_INFO::LEX_MI_UNCHANGED){error= ER_SLAVE_CHANNEL_MUST_STOP;my_error(ER_SLAVE_CHANNEL_MUST_STOP, MYF(0), mi->get_channel());goto err;}/* 如果 SQL thread 和 IO thread 没有全部停止，不能purge relay log */need_relay_log_purge= 0;}/*下面是一些错误判断，都是很明显的错误1. 如果设置了auto_position，同时又指定了复制位点，如 relay_log_pos，报错退出2. auto_position 需要 GTID_MODE != OFF3. IO thread 运行时不能改变 IO thread 相关配置4. SQL thread 运行时不能改变 SQL thread 相关配置5. 如果指定了master_host，那么master_host不能是空串  *//* 记录当前状态 */THD_STAGE_INFO(thd, stage_changing_master); /* 标识停止的线程，给load_mi_and_rli_from_repositories()使用 */init_thread_mask(&thread_mask_stopped_threads, mi, 1);/* 从仓库加载 mi 和 rli 的配置只有停止状态的线程可以加载配置（SQL thread 对应 rli，IO thread 对应 mi）*/if (load_mi_and_rli_from_repositories(mi, false, thread_mask_stopped_threads))                                                                                                                            {error= ER_MASTER_INFO;my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));goto err;}/* 修改mi相关配置，并保存老配置save_ 变量中保存的老配置用于打印日志*/if (have_receive_option){strmake(saved_host, mi->host, HOSTNAME_LENGTH);strmake(saved_bind_addr, mi->bind_addr, HOSTNAME_LENGTH);saved_port= mi->port;strmake(saved_log_name, mi->get_master_log_name(), FN_REFLEN - 1);saved_log_pos= mi->get_master_log_pos();if ((error= change_receive_options(thd, lex_mi, mi)))                                                                                                                                                   {goto err;}}/* 打印日志，change master 的源值和目标值 */if (have_receive_option)sql_print_information("'CHANGE MASTER TO%s executed'. ""Previous state master_host='%s', master_port= %u, master_log_file='%s', ""master_log_pos= %ld, master_bind='%s'. ""New state master_host='%s', master_port= %u, master_log_file='%s', ""master_log_pos= %ld, master_bind='%s'.",mi->get_for_channel_str(true),saved_host, saved_port, saved_log_name, (ulong) saved_log_pos,saved_bind_addr, mi->host, mi->port, mi->get_master_log_name(),(ulong) mi->get_master_log_pos(), mi->bind_addr);/* 修改rli相关配置 */if (have_execute_option)change_execute_options(lex_mi, mi);/* 持久化master_info */if ((thread_mask & SLAVE_IO) == 0 && flush_master_info(mi, true)){error= ER_RELAY_LOG_INIT;my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");goto err;}if ((thread_mask & SLAVE_SQL) == 0){/* 记录全局变量 relay_log_purge */                                                                                                                                                            bool save_relay_log_purge= relay_log_purge;if (need_relay_log_purge){const char* errmsg= 0;/* purge relay logs */relay_log_purge= 1;THD_STAGE_INFO(thd, stage_purging_old_relay_logs);if (mi->rli->purge_relay_logs(thd,0 /* not only reset, but also reinit */,&errmsg)){error= ER_RELAY_LOG_FAIL;my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);goto err;}}else{const char* msg;relay_log_purge= 0;DBUG_ASSERT(mi->rli->inited);/*初始化 relay_log_pos */if (mi->rli->init_relay_log_pos(mi->rli->get_group_relay_log_name(),mi->rli->get_group_relay_log_pos(),true/*we do need mi->rli->data_lock*/,&msg, 0)){error= ER_RELAY_LOG_INIT;my_error(ER_RELAY_LOG_INIT, MYF(0), msg);goto err;}}/* 恢复全局变量 relay_log_purge 的值 */relay_log_purge= save_relay_log_purge;/* 清理until condition */mi->rli->clear_until_condition();/* relay_log_info 持久化到磁盘 */if (mi->rli->flush_info(true)){error= ER_RELAY_LOG_INIT;my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush relay info file.");goto err;}}/* 出错后跳到次数，释放之前申请的锁 */
err:unlock_slave_threads(mi);mi->channel_unlock();DBUG_RETURN(error);}

总结

change master 主要功能是修改 SQL 和 IO 线程的配置信息，执行时可能会purge relay log

没有特殊情况，建议指定auto_position=1，不要自己指定复制位点，避免数据丢失风险

如需对change master 做修改，需要注意在锁保护下修改变量，同时注意加锁顺序，避免死锁