一、调用流程大概如下

handle_slave_sql->是否开启了slave_preserve_commit_order和log_slave_updates参数，开启的话需要设置提交顺序管理器if(opt_slave_preserve_commit_order&&rli->opt_slave_parallel_workers>0&&opt_bin_log&&opt_log_slave_updates)commit_order_mngr=newCommit_order_manager(rli->opt_slave_parallel_workers);//ordercommit管理器rli->set_commit_order_manager(commit_order_mngr);->如果是MTS则需要启动worker线程if(slave_start_workers(rli,rli->opt_slave_parallel_workers,&mts_inited)!=0)//启动worker线程{mysql_cond_broadcast(&rli->start_cond);mysql_mutex_unlock(&rli->run_lock);rli->report(ERROR_LEVEL,ER_SLAVE_FATAL_ERROR,ER(ER_SLAVE_FATAL_ERROR),"Failedduringslaveworkersinitialization");gotoerr;->检查reptable是否是事务类型的如果不是则报警告if(!rli->is_transactional())//是否是table或者是file类型是table类型则支持事物rli->report(WARNING_LEVEL,0,"Ifacrashhappensthisconfigurationdoesnotguaranteethattherelay""loginfowillbeconsistent");->初始化relaylog的访问位置if(rli->init_relay_log_pos(rli->get_group_relay_log_name(),rli->get_group_relay_log_pos(),true/*need_data_lock=true*/,&errmsg,1/*lookforadescription_event*/))//初始化relaylog的访问位置这个位置比较关键也就是从哪里开始读取我们的relaylog。如果出现错误将会导致读取的relaylog错误。因此我们需要保证repinfo的安全，如果设置了recoverrelaylog那么将会初始化为最新一个relaylog的开始位置，因为所有的未执行的binlogevent将会从新拉取，老的relaylog已经不重要了。后面再说。->GTIDevent没有办法使用sql_slave_skip_counter其具体含义参考：Log_event::do_shall_skipmysql>setglobalsql_slave_skip_counter=1;ERROR1858(HY000):sql_slave_skip_countercannotbesetwhentheserverisrunningwith@@GLOBAL.GTID_MODE=ON.Instead,foreachtransactionthatyouwanttoskip,generateanemptytransactionwiththesameGTIDasthetransaction进入循环知道SQL线程被杀死->进入状态stage_reading_event_from_the_relay_log->进行一段skipevent的判断和日志输出GTIDevent没有办法使用sql_slave_skip_counter其具体含义参考：Log_event::do_shall_skipmysql>setglobalsql_slave_skip_counter=1;ERROR1858(HY000):sql_slave_skip_countercannotbesetwhentheserverisrunningwith@@GLOBAL.GTID_MODE=ON.Instead,foreachtransactionthatyouwanttoskip,generateanemptytransactionwiththesameGTIDasthetransaction->exec_relay_log_event读取应用一个event的上层接口->next_event读取下一个Event完成MTS的检查点->获取开始位置rli->set_event_start_pos(my_b_tell(cur_log));->Log_event::read_log_event->如果是MTS是否需要进行检查点1、是否超过检查点周期周期检查在函数mts_checkpoint_routine内部set_timespec_nsec(&curr_clock,0);ulonglongdiff=diff_timespec(&curr_clock,&rli->last_clock);if(!force&&diff<period){/*Wedonotneedtoexecutethecheckpointnowbecausethetimeelapsedisnotenough.*/DBUG_RETURN(FALSE);}2、是否已经GAQ已经满了boolforce=(rli->checkpoint_seqno>(rli->checkpoint_group-1));//如果达到了GAQ的大小设置为force强制checkpoint->是否relaylog大小已经达到最大是否需要relaylog切换但是需要注意如果本事物没有结束不能进行切换

/*Ifwehavereachedthelimitoftherelayspaceandwe如果我们达到relay_log_space_limit上限需要通知IOTHREAD进行切换清理空间```aregoingtosleep,waitingformoreevents:1.Ifoutsideagroup,SQLthreadaskstheIOthreadtoforcearotationsothattheSQLthreadpurgeslogsnexttimeitprocessesanevent(thusspaceisfreed).2.Ifinagroup,SQLthreadaskstheIOthreadtoignorethelimitandqueuesyetonemoreeventsothattheSQLthreadfinishesthegroupandisareabletorotateandpurgesometimesoon.*/if(rli->log_space_limit&&rli->log_space_limit<rli->log_space_total){/*forcerotationifnotinanunfinishedgroup*/if(!rli->is_parallel_exec()){rli->sql_force_rotate_relay=!rli->is_in_group();//如果不是一组就需要切换}else{rli->sql_force_rotate_relay=(rli->mts_group_status!=Relay_log_info::MTS_IN_GROUP);}/*askforonemoreevent*/rli->ignore_log_space_limit=true;//是一组不能切换}```->如果读取了当前relaylog的全部的relaylogevent，->如果是当前relaylog->空闲状态下等待io线程的唤醒，如果是MTS还需要定期醒来进行检查点，如下：```if(rli->is_parallel_exec()&&(opt_mts_checkpoint_period!=0||DBUG_EVALUATE_IF("check_slave_debug_group",1,0))){intret=0;structtimespecwaittime;ulonglongperiod=static_cast<ulonglong>(opt_mts_checkpoint_period*1000000ULL);ulongsignal_cnt=rli->relay_log.signal_cnt;mysql_mutex_unlock(log_lock);do{/*Atthispointthecoordinatorhasnojobtodelegatetoworkers.However,workersareexecutingtheirassignedjobsandassuchthecheckpointroutinemustbeperiodicallyinvoked.*/(void)mts_checkpoint_routine(rli,period,false,true/*need_data_lock=true*/);//TODO:ALFRANIOERRORmysql_mutex_lock(log_lock);if(DBUG_EVALUATE_IF("check_slave_debug_group",1,0))period=10000000ULL;set_timespec_nsec(&waittime,period);ret=rli->relay_log.wait_for_update_relay_log(thd,&waittime);}while((ret==ETIMEDOUT||ret==ETIME)/*todo:remove*/&&signal_cnt==rli->relay_log.signal_cnt&&!thd->killed);}else{rli->relay_log.wait_for_update_relay_log(thd,NULL);//等待relaylog更改的信号SQLTHREAD会等待在这里}```->如果不是当前relaylog那么SQL线程应用或者分发完成完成后就可以清理了并且参数relay_log_purge需要设置为1if(rli->relay_log.purge_first_log(rli,rli->get_group_relay_log_pos()==rli->get_event_relay_log_pos()&&!strcmp(rli->get_group_relay_log_name(),rli->get_event_relay_log_name())))//做relaylog的清理->如果是单SQL现成获取event的时间这一步就是获取计算延迟的重要因素,但是注意MTS不是在这里实在检查点里面last_master_timestamp```rli->last_master_timestamp=ev->common_header->when.tv_sec+//eventheader的timestamp(time_t)ev->exec_time;//获取event的timestamp作为计算last_master_timestamp的基础数据queryevent才有的执行时间DBUG_ASSERT(rli->last_master_timestamp>=0);//但是对于MTS来讲应该注意是最后一个XIDEVENT的时间不是这里设置的在mts_checkpoint_routine里面```->如果GITD_MODE且AUTO_POSITION且是MTS需要由协调线程进行半事物的恢复(partialtransaction)构造回滚EVENT进行恢复，而对已非MTS会在gtidevent做回滚。这种情况可能出现在：-AUTO_POSITION情况下如果重连，会重新发送已经传输的Event。-AUTO_POSITION情况下如果从库异常宕机重启，并且recovery_relay_log=0的情况下,会重新发送已经传输的Event,并且relaylogpos不会重置因此我们前面在IO线程和DUMP线程中已经讨论了，每次sql线程的启动都会通过GTID去重新寻找需要拉取的位置。coord_handle_partial_binlogged_transaction(rli,ev)->apply_event_and_update_pos非MTS完成应用MTS完成分发->进行skipevent操作->维护skipcounter计数器if(reason==Log_event::EVENT_SKIP_COUNT){--rli->slave_skip_counter;//维护skipcountskip_event=TRUE;}我们看到slave_skip_counter是以event为单位的，但是对于最后一个event如果跨事务了那么整个事物都需要跳过。但是skip在GTID模式下是不能用的。->如果不能跳过的事务就需要应用了。MTS则完成分发->完成延迟应用逻辑sql_delay_event(ev,thd,rli)->ev->apply_event(rli);这里单SQL线程应用MTS完成分发，分发方式参考前面->是否是进行MTSrecoveryif(rli->is_mts_recovery())根据bitmap设置进行跳过处理if(rli->is_mts_recovery())//如果是恢复这个地方就是前面恢复扫描出来的位置{boolskip=bitmap_is_set(&rli->recovery_groups,rli->mts_recovery_index)&&(get_mts_execution_mode(::server_id,rli->mts_group_status==Relay_log_info::MTS_IN_GROUP,rli->current_mts_submode->get_type()==MTS_PARALLEL_TYPE_DB_NAME)==EVENT_EXEC_PARALLEL);if(skip){DBUG_RETURN(0);}else{DBUG_RETURN(do_apply_event(rli));}}