ProxySQL源码分析5-MySQL_Thread：：run

整体的流程图如上，其中最重要的就是最后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(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(); }}