文章目錄

• • • 1.MySQL源碼
    • • 1. 主函數sql/mysqld.cc中，代碼如下：
      • 2.監聽連接: sql/mysqld.cc - handle_connections_sockets:
      • 3. 創建連接 sql/mysqld.cc create_new_thread/create_thread_to_handle_connection:
      • 4. 線程調度器thread_scheduler - create_thread_to_handle_connection
      • 5.handle_one_connection
      • 6.執行語句 sql/sql_parse.cc - do_command函數
      • 7.指令分發 sql/sql_parse.cc定義dispatch_command
      • 8.sql/sql_parse.cc mysql_parse函數負責解析SQL
      • 9.執行命令 mysql_execute_command
      • 10.接下來sql/sql_insert.cc中mysql_insert函數
      • 11.接著看真正寫數據的函數write_record (在sql/sql_insert.cc),精簡代碼如下:
    • 2.請求數據流

源碼才是王道。

1.MySQL源碼

---

1. 主函數sql/mysqld.cc中，代碼如下：

//標準入口函數 int main(int argc, char **argv) {//調用mysys/My_init.c->my_init()，初始化mysql內部的系統庫MY_INIT(argv[0]);//初始化日志功能logger.init_base(); //讀取配置文件信息load_defaults(conf_file_name, groups, &argc, &argv);//檢測啟動時的用戶選項 user_info = check_user(mysqld_user);//設置以該用戶運行set_user(mysqld_user, user_info);//初始化內部的一些組件，如table_cache, query_cache等。init_server_components();//初始化網絡模塊，創建socket監聽network_init();// 創建pid文件start_signal_handler();// 初始化mysql中的status變量init_status_vars();//創建manager線程start_handle_manager();//主要處理函數，處理新的連接并創建新的線程處理handle_connections_sockets(); }

2.監聽連接: sql/mysqld.cc - handle_connections_sockets:

pthread_handler_t handle_connections_sockets(void *arg __attribute__((unused))) {// unix_socket在network_init中被打開FD_SET(unix_sock,&clientFDs); // abort_loop是全局變量，在某些情況下被置為1表示要退出。while (!abort_loop) { // 需要監聽的socketreadFDs=clientFDs; // select異步監聽select((int) max_used_connection,&readFDs,0,0,0); // 接受請求new_sock = accept(sock, my_reinterpret_cast(struct sockaddr *)(&cAddr),&length);// 創建mysqld任務線程描述符，它封裝了一個客戶端連接請求的所有信息thd= new THD; // 網絡操作抽象層vio_tmp=vio_new(new_sock, VIO_TYPE_SOCKET, VIO_LOCALHOST); // 初始化任務線程描述符的網絡操作my_net_init(&thd->net,vio_tmp)); // 創建任務線程create_new_thread(thd); } }

3. 創建連接 sql/mysqld.cc create_new_thread/create_thread_to_handle_connection:

static void create_new_thread(THD *thd) {NET *net=&thd->net;// 看看當前連接數是不是超過了系統配置允許的最大值，如果是就斷開連接。if (connection_count >= max_connections + 1 || abort_loop) { close_connection(thd, ER_CON_COUNT_ERROR, 1);delete thd;}++connection_count;// 將新連接加入到thread_scheduler的連接隊列中。thread_scheduler.add_connection(thd); }

4. 線程調度器thread_scheduler - create_thread_to_handle_connection

void create_thread_to_handle_connection(THD *thd) {//看當前工作線程緩存(thread_cache)中有否空余的線程if (cached_thread_count > wake_thread) { thread_cache.append(thd);// 有的話則喚醒一個線程來用pthread_cond_signal(&COND_thread_cache);} else {threads.append(thd);//沒有可用空閑線程則創建一個新的線程pthread_create(&thd->real_id,&connection_attrib, handle_one_connection, (void*) thd))); } }

5.handle_one_connection

pthread_handler_t handle_one_connection(void *arg) {// 初始化線程預處理操作thread_scheduler.init_new_connection_thread(); //載入一些Session級變量setup_connection_thread_globals(thd); for (;;) { //初始化LEX詞法解析器lex_start(thd); // 進行連接身份驗證login_connection(thd); // 初始化線程Status,即show status看到的prepare_new_connection_state(thd); // 處理命令do_command(thd); //沒事做了關閉連接,丟入線程池end_connection(thd); } }

6.執行語句 sql/sql_parse.cc - do_command函數

bool do_command(THD *thd) {NET *net= &thd->net;//讀取客戶端發送的報文packet_length = my_net_read(net);packet = (char*) net->read_pos;// 解析客戶端傳過來的命令類型command = (enum enum_server_command) (uchar) packet[0]; // 根據命令類型進行分發dispatch_command(command, thd, packet+1, (uint) (packet_length-1)); }

7.指令分發 sql/sql_parse.cc定義dispatch_command

bool dispatch_command(enum enum_server_command command, THD *thd, char* packet, uint packet_length) {NET *net = &thd->net;thd->command = command; switch (command) { //判斷命令類型case COM_INIT_DB: ...;case COM_TABLE_DUMP: ...;case COM_CHANGE_USER: ...;...case COM_QUERY: //如果是Queryalloc_query(thd, packet, packet_length); //從網絡數據包中讀取Query并存入thd->querymysql_parse(thd, thd->query, thd->query_length, &end_of_stmt); //送去解析} }

8.sql/sql_parse.cc mysql_parse函數負責解析SQL

void mysql_parse(THD *thd, const char *inBuf, uint length, const char ** found_semicolon) {//初始化線程解析描述符lex_start(thd); // 看query cache中有否命中，有就直接返回結果，否則進行查找if (query_cache_send_result_to_client(thd, (char*) inBuf, length) <= 0) { Parser_state parser_state(thd, inBuf, length); // 解析SQL語句 parse_sql(thd, & parser_state, NULL); // 執行語句mysql_execute_command(thd); } }

9.執行命令 mysql_execute_command

int mysql_execute_command(THD *thd) {// 解析過后的SQL語句的語法結構LEX *lex= thd->lex; // 該語句要訪問的表的列表TABLE_LIST *all_tables = lex->query_tables; switch (lex->sql_command) {...case SQLCOM_INSERT://insert語句insert_precheck(thd, all_tables);mysql_insert(thd, all_tables, lex->field_list, lex->many_values, lex->update_list, lex->value_list, lex->duplicates, lex->ignore);break; ... case SQLCOM_SELECT: // select 語句// 檢查用戶對數據表的訪問權限check_table_access(thd, lex->exchange ? SELECT_ACL | FILE_ACL : SELECT_ACL, all_tables, UINT_MAX, FALSE); // 執行select語句execute_sqlcom_select(thd, all_tables); break;} }

10.接下來sql/sql_insert.cc中mysql_insert函數

bool mysql_insert(THD *thd,TABLE_LIST *table_list, // 該INSERT要用到的表List<Item> &fields, // 使用的項....) {// 這里的鎖只是防止表結構修改open_and_lock_tables(thd, table_list); mysql_prepare_insert(...);//里面還有trigger，錯誤，view之類的雜七雜八的東西，我們都忽略foreach value in values_list {write_record(...);} }

11.接著看真正寫數據的函數write_record (在sql/sql_insert.cc),精簡代碼如下:

int write_record(THD *thd, TABLE *table,COPY_INFO *info) { // 寫數據記錄//如果是REPLACE或UPDATE則替換數據if (info->handle_duplicates == DUP_REPLACE || info->handle_duplicates == DUP_UPDATE) { table->file->ha_write_row(table->record[0]);table->file->ha_update_row(table->record[1], table->record[0]));} else {table->file->ha_write_row(table->record[0]);} } //存儲引擎抽象的Handler API ! int handler::ha_write_row(uchar *buf) { write_row(buf); // 調用具體的實現binlog_log_row(table, 0, buf, log_func)); // 寫binlog }

---

2.請求數據流

總結

以上是生活随笔為你收集整理的MySql源码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。