#从库MTS多线程并行回放（一） --- 欢迎关注我的《深入理解MySQL主从原理 32讲 》，如下：  如果图片不能显示可查看下面链接： https://www.jianshu.com/p/d636215d767f 本节包含一个笔记如下： https://www.herecours.com/d/file/efpub/2026/01-01/20260301143347753680](https://upload-images.jianshu.io/upload_images/7398834-247112e730bd5e64.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240) ###三、步骤解析 下面对每一步进行解析如下： （1）如果是GTID\_LOG\_EVENT代表事物开始，将本事物加入到GAQ队列中（下一节会详细描述GAQ）。可参考函数Log\_event::get\_slave\_worker。 （2）将GTID\_LOG\_EVENT加入到curr\_group\_da队列中暂存。可参考函数Log\_event::get\_slave\_worker。 （3）获取GTID\_LOG\_EVENT中的last commit和seq number值。可参考函数Mts\_submode\_logical\_clock::schedule\_next\_event。 （4）获取current\_lwm值，这个值代表的是所有在GAQ队列上还没有提交完成事务中最早的那个事务的前一个已经提交事务的seq number，可能后面的事务已经提交完成了，听起来可能比较拗口但很重要，如果都提交完成了那么就是取最新提交的事务的seq number，下面的图表达的就是这个意思，这个图是源码中的。这个值的获取可参考函数Mts\_submode\_logical\_clock::get\_lwm\_timestamp。 ``` the last time index containg lwm +------+ | LWM | | | | V V V GAQ:x xoooooxxxxxXXXXX...X ^ ^ | | LWM+1（LWM代表的是检查点指向的位置） | + new current_lwm（这里就是current_lwm） <---- logical (commit) time ---- here `x' stands for committed, `X' for committed and discarded from the running range of the queue, `o' for not committed. ``` 我们可以先不看LWM部分，对于检查点的LWM后面在讨论。seq number从右向左递增，在GAQ中实际上有三种值： - X：已经做了检查点，在GAQ中出队的事物。 - x：已经提交完成的事物。 - o：没有提交完成的事物。 我们可以看到我们需要获取的current\_lwm并不是最新一次提交事物的seq number的值，而是最早未提交事物的前一个已经提交事物的seq number。这一点很重要，因为理解后就会知道大事务是如何影响MTS的并行回放的，同时中间的5个‘o’实际上就是所谓的‘gap’，关于‘gap’下一节还会详细描述。 （5）将GTID\_LOG\_EVENT中的last commit和当前current\_lwm进行比较。可以参考函数Mts\_submode\_logical\_clock::schedule\_next\_event。下面是大概的规则： - 如果last commit小于等于current\_lwm表示可以进行并行回放，继续。 - 如果last commit大于current\_lwm则表示不能进行并行回放。这个时候协调线程就需要等待了，直到小于等于的条件成立。成立后协调线程会被工作线程唤醒。等待期间状态被置为“Waiting for dependent transaction to commit”。 源码处也比较简单如下： ``` longlong lwm_estimate= estimate_lwm_timestamp(); //这个值 只有在 出现 下面等待的时候 才会设置 min_waited_timestamp ， //设置了min_waited_timestamp才会更新lwm_estimate if (!clock_leq(last_committed, lwm_estimate) && // @return true when a "<=" b,false otherwise last_committed<=lwm_estimate rli->gaq->assigned_group_index != rli->gaq->entry) { if (wait_for_last_committed_trx(rli, last_committed, lwm_estimate)) //等待上一次 组提交的完成 Waiting for dependent transaction to commit ``` （6）如果是QUERY\_EVENT则加入到curr\_group\_da队列中暂存。 （7）如果是MAP\_EVENT进行工作线程的分配。参考函数Mts\_submode\_logical\_clock::get\_least\_occupied\_worker，分配工作线程如下： - 如果有空闲的工作线程则分配完成，继续。 - 如果没有空闲的工作线程则等待空闲的工作线程。这种情况下状态会置为“Waiting for slave workers to process their queues”。 下面是分配的标准，其实也很简单： ``` for (Slave_worker **it= rli->workers.begin(); it != rli->workers.end(); ++it) { Slave_worker *w_i= *it; if (w_i->jobs.len == 0) //任务队列为0表示本Worker线程空闲可以分配 return w_i; } return 0; ``` （8）将GTID\_LOG\_EVENT和QUERY\_EVENT分配给工作线程。可参考append\_item\_to\_jobs函数。 前面工作线程已经分配了，这里就可以开始将Event分配给这个工作线程了。分配的时候需要检查工作线程的任务队列是否已满，如果满了需要等待，状态置为“Waiting for Slave Worker queue”。因为分配的单位是Event，对于一个事务而言可能包含很多Event，如果工作线程应用的速度赶不上协调线程入队的速度，可能导致任务队列的积压，因此任务队列被占满是可能的。任务队列的大小为16384如下： ``` mts_slave_worker_queue_len_max= 16384; ``` 下面是入队的部分代码： ``` while (worker->running_status == Slave_worker::RUNNING && !thd->killed && (ret= en_queue(&worker->jobs, job_item)) == -1) //如果已经满了 { thd->ENTER_COND(&worker->jobs_cond, &worker->jobs_lock, &stage_slave_waiting_worker_queue, &old_stage); //标记等待状态 worker->jobs.overfill= TRUE; worker->jobs.waited_overfill++; rli->mts_wq_overfill_cnt++; //标记队列满的次数 mysql_cond_wait(&worker->jobs_cond, &worker->jobs_lock); //等待唤醒 mysql_mutex_unlock(&worker->jobs_lock); thd->EXIT_COND(&old_stage); mysql_mutex_lock(&worker->jobs_lock); } ``` （9）MAP\_EVENT分配给工作线程，同上。 （10）DELETE\_EVENT分配给工作线程，同上。 （11）XID\_EVENT分配给工作线程，但是这里还需要额外的处理，主要处理一些和检查点相关的信息，这里关注一点如下： ``` ptr_group->checkpoint_log_name= my_strdup(key_memory_log_event, rli->get_group_master_log_name(), MYF(MY_WME)); ptr_group->checkpoint_log_pos= rli->get_group_master_log_pos(); ptr_group->checkpoint_relay_log_name=my_strdup(key_memory_log_event, rli->get_group_relay_log_name(), MYF(MY_WME)); ptr_group->checkpoint_relay_log_pos= rli->get_group_relay_log_pos(); ptr_group->ts= common_header->when.tv_sec + (time_t) exec_time; //Seconds_behind_master related .checkpoint //的时候会将这个值再次传递 mts_checkpoint_routine() ptr_group->checkpoint_seqno= rli->checkpoint_seqno; //获取seqno 这个值会在chkpt后减去偏移量 ``` 如果检查点处于这个事务上，那么这些信息会出现在表 slave\_worker\_info中，并且会出现在show slave status中。也就是说，show slave status中很多信息是来自MTS的检查点。下一节将详细描述检查点。 （12）如果上面Event的分配过程大于2分钟（120秒），可能会出现一个日志如下：  这个截图也是一个朋友问的问题。实际上这个日志可以算一个警告。实际上对应的源码为： ``` sql_print_information("Multi-threaded slave statistics%s: " "seconds elapsed = %lu; " "events assigned = %llu; " "worker queues filled over overrun level = %lu; " "waited due a Worker queue full = %lu; " "waited due the total size = %lu; " "waited at clock conflicts = %llu " "waited (count) when Workers occupied = %lu " "waited when Workers occupied = %llu", rli->get_for_channel_str(), static_cast<unsigned long> (my_now - rli->mts_last_online_stat), //消耗总时间 单位秒 rli->mts_events_assigned, //总的event分配的个数 rli->mts_wq_overrun_cnt, // worker线程分配队列大于 90%的次数 当前硬编码 14746 rli->mts_wq_overfill_cnt, //由于work 分配队列已满造成的等待次数 当前硬编码 16384 rli->wq_size_waits_cnt, //大Event的个数 一般不会存在 rli->mts_total_wait_overlap, //由于上一组并行有大事物没有提交导致不能分配worker线程的等待时间 单位纳秒 rli->mts_wq_no_underrun_cnt, //work线程由于没有空闲的而等待的次数 rli->mts_total_wait_worker_avail); //work线程由于没有空闲的而等待的时间 单位纳秒 ``` 因为经常看到朋友问这里详细说明一下它们的含义，从前面的分析中我们一共看到三个等待点： - “Waiting for dependent transaction to commit” 由于协调线程判定本事务由于last commit大于current\_lwm因此不能并行回放，协调线程处于等待，大事务会加剧这种情况。 - “Waiting for slave workers to process their queues” 由于没有空闲的工作线程，协调线程会等待。这种情况说明理论上的并行度是理想的，但是可能是参数slave\_parallel\_workers设置不够。当然设置工作线程的个数应该和服务器的配置和负载相结合考虑，因为第29节我们会看到线程是CPU调度最小的单位。 - “Waiting for Slave Worker queue” 由于工作线程的任务队列已满，协调线程会等待。这种情况前面说过是由于一个事务包含了过多的Event并且工作线程应用Event的速度赶不上协调线程分配Event的速度，导致了积压并且超过了16384个Event。 另外实际上还有一种等待如下： “Waiting for Slave Workers to free pending events”：由所谓的‘big event’造成的，什么是‘big event’呢，源码中描述为：event size is greater than slave\_pending\_jobs\_size\_max but less than slave\_max\_allowed\_packet。我个人认为出现的可能性不大，因此没做过多考虑。可以在函数append\_item\_to\_jobs中找到答案。 我们下面对应日志中的输出进行详细解释，如下： |指标|解释| |---|---| |seconds elapsed|整个分配过程消耗的时间，单位秒，超过120秒会出现这个日志。| |events assigned|本工作线程分配的Event数量。| |worker queues filled over overrun level|本工作线程任务队列中Event个数大于90%的次数。当前硬编码大于14746。| |waited due a Worker queue full|本工作线程任务队列已满的次数。当前硬编码大于16384。和前面第三点对应。| |waited due the total size|‘big event’的出现的次数。 |waited at clock conflicts|由于不能并行回放，协调线程等待的时间，单位纳秒。和前面第一点对应。| |waited (count) when Workers occupied|由于没有空闲的工作线程而等待的次数。对应前面第二点。| |waited when Workers occupied|由于没有空闲的工作线程而等待的时间。对应前面第二点。| 我们可以看到这个日志还是记录很全的，基本覆盖了前面我们讨论的全部可能性。那么我们再看看案例中的日志，waited at clock conflicts=91895169800 大约91秒。120秒钟大约91秒都因为不能并行回放而造成的等待，很明显应该考虑是否有大事物的存在。 ###四、并行回放判定的列子 下面是我主库使用WRITESET方式生成的一段binary log片段，我们主要观察lastcommit和seq number，通过分析来熟悉这种过程。  我们根据刚才说的并行判断规则，即： - 如果last commit小于等于current\_lwm表示可以进行并行回放，继续。 - 如果last commit大于current\_lwm则表示不能进行并行回放，需要等待。 ######具体解析如下： （last commit：22 seq number：23）这个事务会在（last commit：21 seq number：22）事务执行完成后执行因为（last commit：22<= seq number：22），后面的事务直到（last_commit：22 seq number：30），实际上都可以并行执行，我们先假设他们都执行完成了。我们继续观察随后的三个事务如下： - last\_committed：29 sequence\_number：31 - last\_committed：30 sequence\_number：32 - last\_committed：27 sequence\_number：33 我们注意到到这是基于WRITESET的并行复制下明显的特征。 last commit可能比上一个事务更小，这就是我们前面说的根据Writeset的历史MAP信息计算出来的。因此还是根据上面的规则它们三个是可以并行执行的。因为很明显： - last\_committed：29 <= current\_lwm：30 - last\_committed：30 <= current\_lwm：30 - last\_committed：27 <= current\_lwm：30 但是如果（last commit：22 seq number：30）这个事务之前有一个大事务没有执行完成的话，那么current\_lwm的取值将不会是30。比如（last commit：22 seq number：27）这个事务是大事务那么current\_lwm将会标记为26，上面的三个事务将会被堵塞，并且分配（last commit：29 seq number：31）的时候就已经堵塞了，原因如下： - last\_committed：29 > current\_lwm：26 - last\_committed：30 > current\_lwm：26 - last\_committed：27 > current\_lwm：26 我们再考虑一下基于WRITESET的并行复制下（last commit：27 seq number：33）这个事务，因为在我们并行规则下last commit越小获得并发的可能性越高。因此基于WRITESET的并行复制确实提高了从库回放的并行度，但正如第16节所讲主库会有一定的开销。 --- 第19节结束