mysql线程池的实现原理浅析

create_new_thread(thd){ // 检查连接数是否过大 mysql_mutext_lock(&lock_connection_count); if (connetion_count >= max_connections + 1 || abort_loop){ mysql_mutext_unlock(&lock_connection_count); delete thd; // will close the connection return; } ++connection_count; if (connection_count > max_used_connections){ max_used_connections = connection_count; } mysql_mutext_unlock(&lock_connection_count); mysql_mutex_lock(&lock_thread_count); // 我在想如果连接很多很多，溢出了怎么办？？ thd->thread_id = thd->variables.pseudo_thread_id = thread_id++; thread_count++; // add_connection当然是个函数指针，thread_scheduler是一个放了这些函数指针的结构体 // 为了跨平台，却没有使用C++的多态，而是用了函数指针实现了多态 thread_scheduler->add_connection(thd); }

// 这个就是add_connection的实现
void create_thread_to_handle_connection(thd){
    // 注意这个判断，cache_thread_count大于wake_thread的时候才会进来，
    // 当线程结束的时候，线程cachd_thread_count++
    // 当发现有空闲，先wake_thread_count++，cachd_thread_count--
    // 成功启动线程后，wake_thread_count--，这时wake_thread_count和cachd_thread_count又归于平衡
    // 简单来说，cache_thread_count是用来记录可用线程数量的，而wake_thread是用来记录就绪线程数量的
    // 多数时候，wake_thread_count基本是0，而cache_thread_count则大于等于0
    if (cache_thread_count > wake_thread){
        thread_cache.push_back(thd);    
        wake_thread++;
        mysql_cond_signal(&cond_thread_cache);
    }
    else {
        // 如果没有cached thread则直接创建新的线程
        thread_created++;
        // threads这个list放着所有正在运行时的thd
        threads.append(thd);    
        thd->prior_thr_create_utime = thd->start_utime = my_micro_time();
        // 真正创建一个线程出来
        if (error = mysql_thread_create(key_thread_one_connection, &thd->real_id, &connection_attrib
                    handle_one_connection)) {
            // do some clear up
            // ignore the lock operations
            // 创建失败清理一下
            thread_count--;
            connection_count--;
            delete thd;
        }
    }
}

那么什么时候会进入if (connetion_count >= max_connections + 1 || abort_loop)这个if语句当中呢？就是有线程结束，并等待新的thd来的时候，看看线程结束的时候做了些什么：

// 线程结束时的回调
bool one_thread_per_connection_end(thd, bool put_in_cache){
    unlink_thd(thd); // delete thd
    if (put_in_cache) {
        // 关键函数cache_thread
        put_in_cache = cache_thread();
    }
    // 如果已经获得新thd，就不结束了
    if (put_in_cache){
        return 0;
    }
    
    // 否则结束线程
    my_thread_end();
    mysql_cond_broadcast(&cond_thread_count);

    pthread_exit(0);
    return 0;
}

其中的cache_thread是关键函数：

// 这个函数就是cache thread的关键，他的思路就是线程结束的时候，wait for cache_thread队列
// 取出新的thd，运行之
static bool cache_thread(){
    // 判断是否到达cache 线程的阈值
    if (cache_thread_count < thread_cache_size && !abort_loop && !kill_cache_threads){
        cache_thread_count++; // 进入cache状态    

        // wait for the signal to relive the thread
        // 开始wait for新的thd装进cache_thread list
        // 注意这里用的是while而不是if，因为
        // 1、cond_wait可能虚假唤醒，可能因为竞争，并没有真正获得新的thd
        // 2、若获取失败则再次等新的thd，总之就是尽可能的获得新的thd
        while (!abort_loop && !wake_thread && !kill_cached_threads){
            mysql_cond_wait(&cond_thread_cache, &lock_thread_count);    
            cached_thread_count--;
            // 此处没仔细看是什么逻辑，暂时跳过
            if (kill_cached_threads){
                mysql_cond_signal(&cond_flush_thread_cache);    
            }
            // 发现有就绪的thd，获取之，然后执行
            if (wake_thread){
                THD* thd;
                wake_thread--;
                thd = thread_cache.get();        
                thd->thread_stack = (char*)&thd;
                thd->store_globals();

                thd->mysys_var->abort = 0;
                thd->prior_thr_create_utime = thd->start_utime = my_micro_time();
                // 返回1使得线程不会结束，跑新的thd了
                return 1;
            }
        }
    }
    // 结束线程
    return 0;
}

分析cache thread思路完毕。

 

mysql线程池的实现原理浅析

标签：逻辑   art   个数   class   tor   max   如何   scheduler   空闲