整体的流程图如上,其中最重要的就是最后2步,分别处理data_stream和session
void MySQL_Thread::run() {
// 死循环
while (shutdown == 0) {
// processing_idles初始化是false,上次检查时间大于ping的周期
if (processing_idles == false &&
(last_processing_idles < curtime - mysql_thread___ping_interval_server_msec * 1000)) {
// 管理idle的连接池,step1(连接池管理)
// 如果机器不是online的,这个机器上的连接直接回收
// 如果服务器上的连接大于最大连接,回收
// 如果服务器上的连接超过了一定水位,回收
// 如果连接超时了connection_max_age,回收
// 管理idle的连接池,step2(多路复用)
// 检查这些连接与后端的ping的状态,如果正常ping通,并且连接满足复用的条件则保留,否则就是回收连接并返回-1
run___get_multiple_idle_connections(num_idles);
}
// 处理mirror任务
handle_mirror_queue_mysql_sessions();
// 根据ds的来管理session状态
// 根据ds的来管理mypolls timeout状态
// 根据ds的来管理监听events
ProcessAllMyDS_BeforePoll();
// 如果有新的客户端连接过来,那就增加poll的事件监听
while ((n = __sync_add_and_fetch(&mypolls.pending_listener_add, 0))) { // spin here
poll_listener_add(n);}
// 接受事件
rc = poll(mypolls.fds, mypolls.len, ttw);
// 处理断开的连接,删除对应的poll的事件监听
while ((n = __sync_add_and_fetch(&mypolls.pending_listener_del, 0))) { // spin here
if (static_cast<int>(n) == -1) {
for (unsigned int i = 0; i < mypolls.len; i++) {
if (mypolls.myds[i] && mypolls.myds[i]->myds_type == MYDS_LISTENER) {
poll_listener_del(mypolls.myds[i]->fd);
}
}
} else {
poll_listener_del(n);
}
assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_del, n, 0));
}
// 1、Scan_Sessions_to_Kill_All:将mysql_sessions、idle_mysql_sessions、resume_mysql_sessions、myexchange.idle_mysql_sessions、myexchange.resume_mysql_sessions全部遍历kill一遍
// kq.conn_ids:线程id相等、_sess->client_myds存在、username相等,标记_sess->killed=true;
// kq.query_ids:线程id相等、_sess->client_myds存在、username相等、_sess->mybe后端存在,标记_sess->killed=true;标记后端_sess->mybe->server_myds->wait_until=curtime;_sess->mybe->server_myds->kill_type=1;
// 2、将maintenance_loop=true
handle_kill_queues();
if (maintenance_loop) {
//根据mirror的任务情况调整GloMTH的status_variables,增加mirror_concurrency
run___cleanup_mirror_queue();
// this function is called by each thread to update global query statistics
GloQPro->update_query_processor_stats();
}
// 检查当前线程和全局线程的变量版本差异&更新
if (__sync_add_and_fetch(&__global_MySQL_Thread_Variables_version, 0) >
__thread_MySQL_Thread_Variables_version) {
refresh_variables();
}
// 遍历一遍mypoll
// 新建连接 listener_handle_new_connection
// 处理已有连接 process_data_on_data_stream
ProcessAllMyDS_AfterPoll();
//
process_all_sessions();
return_local_connections();
}
}