下面还是摘取了handle_slave_sql的主要代码来分析，看起来比较简单就是包括两部：
1） sql线程slave_start_workers启动了一堆的线程池worker，来处理每次读取的事件
2） 后面sql线程exec_relay_log_event读取log分发到不同的worker
pthread_handler_t handle_slave_sql(void *arg)
{
if (slave_start_workers(rli, rli->opt_slave_parallel_workers, &mts_inited) != 0)
{
}
while (!sql_slave_killed(thd,rli)) {
if (exec_relay_log_event(thd,rli)) {
}
}
}
展开看sql线程是如何启动slave线程池的
int slave_start_workers(Relay_log_info *rli, ulong n, bool *mts_inited)
{
if (init_hash_workers(n)){}
for (i= 0; i < n; i++)
{
if ((error= slave_start_single_worker(rli, i)))
goto err;
}
}
init_hash_workers这个函数其实是初始化一个hash，将每个worker对应到一个db上面，索引就是db 名字，后面sql处理分发就是按照这个db 名字来的，也就是mysql主从同步是按照db来并发的。
slave_start_single_worker这个函数式真正的启动worker线程的地方，展开这个函数能发现
下面代码：
| | (error= mysql_thread_create(key_thread_slave_worker, &th,
| | | | | | | | | | &connection_attrib,
| | | | | | | | | | handle_slave_worker, (void*) w)))
{
| sql_print_error("Failed during slave worker thread create (errno= %d)",
| | | | | | | | | error);
| error= 1;
| goto err;
}
这个地方开始启动线程，线程处理函数handle_slave_worker，参数是Slave_worker类型指针
查看一下这个类定义的地方
class Slave_worker : public Relay_log_info
其实它是Relay_log_info的一个子类，这个地方主要是为了处理不同的event设计的，这样可以自动调用
不同的子类来处理不同的事件。
展开handle_slave_worker,这个线程其实是worker不同的再处理master推送过来的event:
pthread_handler_t handle_slave_worker(void *arg)
{
while (!error)
{
| | error= slave_worker_exec_job(w, rli);
}
}
int slave_worker_exec_job(Slave_worker *worker, Relay_log_info *rli)
{
job_item= pop_jobs_item(worker, job_item);
error= ev->do_apply_event_worker(worker);
}
pop_jobs_item是从队列取一个event，这个队列比较重要，后面还有看到sql线程会每次往
这个队列里面放一个event
do_apply_event_worker将取到的evnet执行,这个函数里面调用do_apply_event，这个其实是个
虚函数，每个不同event 有不同的实现。
到这里worker池的主要流程其实大略的看了一遍了，后面就是分析worker池的event是怎么拿到的。
回到handle_slave_sql，其实我们看到启动线程池之后，它就在不停的exec_relay_log_event
这个函数主要是调用apply_event_and_update_pos，主要功能就是两个函数实现：
exec_res= ev->apply_event(rli);
if (append_item_to_jobs(job_item, w, rli))
DBUG_RETURN(SLAVE_APPLY_EVENT_AND_UPDATE_POS_APPEND_JOB_ERROR);
第一个apply_event就是根据上面worker线程池初始化的时候的hash来得到本次事件处理所使用的
worker
第二个append_item_to_jobs就是将第一步获取的worker处理的事件放到队列里面，供第一步对应的
worker的线程来处理。
至此主从分析完了，主从分析非常复杂，上面仅仅是分析大略的流程，主要是为了以后遇到问题需要读代码的时候可以快速定位到相应的代码位置。
阅读(1825) | 评论(0) | 转发(0) |
