【从入门到放弃-MySQL】数据库连接过程分析

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 高可用系列,价值2615元额度,1个月
简介: 前言上周出现了几次连接超时、连接池满还有dbc连接事务模板失败的问题。所以有必要深入了解下MySQL的连接过程。正好,上周研究了怎么用Clion调试MySQL源码,接下来通过调试来研究一下吧。服务端启动sql/main.

前言

上周出现了几次连接超时、连接池满还有dbc连接事务模板失败的问题。所以有必要深入了解下MySQL的连接过程。

正好,上周研究了怎么用Clion调试MySQL源码,接下来通过调试来研究一下吧。

服务端

启动

sql/main.cc

extern int mysqld_main(int argc, char **argv);

int main(int argc, char **argv) { return mysqld_main(argc, argv); }
  • main:入口文件,仅调用了mysqld_main函数

sql/mysqld.cc

int mysqld_main(int argc, char **argv)
#endif
{
if (my_init())  // init my_sys library & pthreads
  {
    ...
  }
  ...
  if (load_defaults(MYSQL_CONFIG_NAME, load_default_groups, &argc, &argv,
                    &argv_alloc)) {
    ...
  }
  mysqld_socket_acceptor->connection_event_loop();

  mysqld_exit(signal_hand_thr_exit_code);
}
  • mysql_main:MySQL服务端启动逻辑的主要处理函数
  • my_init:系统库和线程初始化
  • load_defaults:加载my.cnf各参数
  • connection_event_loop:循环监听套接字。

sql/conn_handler/connection_acceptor.h

  /**
    Connection acceptor loop to accept connections from clients.
  */
  void connection_event_loop() {
    Connection_handler_manager *mgr =
        Connection_handler_manager::get_instance();
    while (!connection_events_loop_aborted()) {
      Channel_info *channel_info = m_listener->listen_for_connection_event();
      if (channel_info != NULL) mgr->process_new_connection(channel_info);
    }
  }
  • connection_event_loop:通过socket_connection.cc::listen_for_connection_event循环监听,直到有新的连接,开始connection_handler_manager.cc::process_new_connection新连接的处理过程。

新连接

服务端一直处于监听状态,当有新连接请求时,调用process_new_connection处理新连接。

sql/conn_handler/connection_handler_manager.cc

void Connection_handler_manager::process_new_connection(
    Channel_info *channel_info) {
  if (connection_events_loop_aborted() ||
      !check_and_incr_conn_count(channel_info->is_admin_connection())) {
    channel_info->send_error_and_close_channel(ER_CON_COUNT_ERROR, 0, true);
    delete channel_info;
    return;
  }

  if (m_connection_handler->add_connection(channel_info)) {
    inc_aborted_connects();
    delete channel_info;
  }
}
  • connection_events_loop_aborted:先判断是否已取消监听
  • check_and_incr_conn_count:再判断(会加锁)是否现有连接数是否大于连接最大值(连接池满),未满,则将线程数加一,满了则拒绝连接。(注意,这里的判断逻辑使MySQL的实际最大连接数是max_connections + 1)
  • add_connection:调用add_connection添加连接

sql/conn_handler/connection_handler_pre_thread.cc

bool Per_thread_connection_handler::add_connection(Channel_info *channel_info) {
  int error = 0;
  my_thread_handle id;

  DBUG_ENTER("Per_thread_connection_handler::add_connection");

  // Simulate thread creation for test case before we check thread cache
  DBUG_EXECUTE_IF("fail_thread_create", error = 1; goto handle_error;);

  if (!check_idle_thread_and_enqueue_connection(channel_info))
    DBUG_RETURN(false);

  /*
    There are no idle threads avaliable to take up the new
    connection. Create a new thread to handle the connection
  */
  channel_info->set_prior_thr_create_utime();
  error =
      mysql_thread_create(key_thread_one_connection, &id, &connection_attrib,
                          handle_connection, (void *)channel_info);
#ifndef DBUG_OFF
handle_error:
#endif  // !DBUG_OFF

  if (error) {
    ...
    //错误处理,略
  }

  Global_THD_manager::get_instance()->inc_thread_created();
  DBUG_PRINT("info", ("Thread created"));
  DBUG_RETURN(false);
}
  • 调用check_idle_thread_and_enqueue_connection查看是否有空闲的线程,有则将本次连接信息加入等待队列,并给空闲线程发送唤醒信号;否则新建线程处理本次连接
  • 在新线程中,调用handle_connection函数开始进行逻辑处理。
static void *handle_connection(void *arg) {
  Global_THD_manager *thd_manager = Global_THD_manager::get_instance();
  Connection_handler_manager *handler_manager =
      Connection_handler_manager::get_instance();
  Channel_info *channel_info = static_cast<Channel_info *>(arg);
  bool pthread_reused MY_ATTRIBUTE((unused)) = false;

  if (my_thread_init()) {
    ...
    //错误处理,略
  }

  for (;;) {
    THD *thd = init_new_thd(channel_info);
    if (thd == NULL) {
      ...
      //错误处理,略
    }

#ifdef HAVE_PSI_THREAD_INTERFACE
    if (pthread_reused) {
      ...
      //错误处理,略
    }
#endif

#ifdef HAVE_PSI_THREAD_INTERFACE
    /* Find the instrumented thread */
    PSI_thread *psi = PSI_THREAD_CALL(get_thread)();
    /* Save it within THD, so it can be inspected */
    thd->set_psi(psi);
#endif /* HAVE_PSI_THREAD_INTERFACE */
    mysql_thread_set_psi_id(thd->thread_id());
    mysql_thread_set_psi_THD(thd);
    mysql_socket_set_thread_owner(
        thd->get_protocol_classic()->get_vio()->mysql_socket);

    thd_manager->add_thd(thd);

    if (thd_prepare_connection(thd))
      handler_manager->inc_aborted_connects();
    else {
      while (thd_connection_alive(thd)) {
        if (do_command(thd)) break;
      }
      end_connection(thd);
    }
    close_connection(thd, 0, false, false);

    thd->get_stmt_da()->reset_diagnostics_area();
    thd->release_resources();

    // Clean up errors now, before possibly waiting for a new connection.
#ifndef HAVE_WOLFSSL
#if OPENSSL_VERSION_NUMBER < 0x10100000L
    ERR_remove_thread_state(0);
#endif /* OPENSSL_VERSION_NUMBER < 0x10100000L */
#endif
    thd_manager->remove_thd(thd);
    Connection_handler_manager::dec_connection_count();

#ifdef HAVE_PSI_THREAD_INTERFACE
    /*
      Delete the instrumentation for the job that just completed.
    */
    thd->set_psi(NULL);
    PSI_THREAD_CALL(delete_current_thread)();
#endif /* HAVE_PSI_THREAD_INTERFACE */

    delete thd;

    // Server is shutting down so end the pthread.
    if (connection_events_loop_aborted()) break;

    channel_info = Per_thread_connection_handler::block_until_new_connection();
    if (channel_info == NULL) break;
    pthread_reused = true;
    if (connection_events_loop_aborted()) {
      ...
      //错误处理,略
    }
  }

  my_thread_end();
  my_thread_exit(0);
  return NULL;
}
  • 会对连接进行thd_prepare_connection预处理操作,没问题后继续下面的逻辑。
  • 当连接未被关闭,就会一直do_command处理请求。
  • 当连接关闭,则走下面关闭逻辑

执行

sql/sql_parse.cc

bool do_command(THD *thd) {
  bool return_value;
  int rc;
  NET *net = NULL;
  enum enum_server_command command;
  COM_DATA com_data;
  DBUG_ENTER("do_command");
  DBUG_ASSERT(thd->is_classic_protocol());

  /*
    indicator of uninitialized lex => normal flow of errors handling
    (see my_message_sql)
  */
  thd->lex->set_current_select(0);

  /*
    XXX: this code is here only to clear possible errors of init_connect.
    Consider moving to prepare_new_connection_state() instead.
    That requires making sure the DA is cleared before non-parsing statements
    such as COM_QUIT.
  */
  thd->clear_error();  // Clear error message
  thd->get_stmt_da()->reset_diagnostics_area();

  /*
    This thread will do a blocking read from the client which
    will be interrupted when the next command is received from
    the client, the connection is closed or "net_wait_timeout"
    number of seconds has passed.
  */
  net = thd->get_protocol_classic()->get_net();
  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
  net_new_transaction(net);

  /*
    Synchronization point for testing of KILL_CONNECTION.
    This sync point can wait here, to simulate slow code execution
    between the last test of thd->killed and blocking in read().

    The goal of this test is to verify that a connection does not
    hang, if it is killed at this point of execution.
    (Bug#37780 - main.kill fails randomly)

    Note that the sync point wait itself will be terminated by a
    kill. In this case it consumes a condition broadcast, but does
    not change anything else. The consumed broadcast should not
    matter here, because the read/recv() below doesn't use it.
  */
  DEBUG_SYNC(thd, "before_do_command_net_read");

  /*
    Because of networking layer callbacks in place,
    this call will maintain the following instrumentation:
    - IDLE events
    - SOCKET events
    - STATEMENT events
    - STAGE events
    when reading a new network packet.
    In particular, a new instrumented statement is started.
    See init_net_server_extension()
  */
  thd->m_server_idle = true;
  rc = thd->get_protocol()->get_command(&com_data, &command);
  thd->m_server_idle = false;

  if (rc) {
    ...
    //错误处理,略
  }

  char desc[VIO_DESCRIPTION_SIZE];
  vio_description(net->vio, desc);
  DBUG_PRINT("info", ("Command on %s = %d (%s)", desc, command,
                      command_name[command].str));

  DBUG_PRINT("info", ("packet: '%*.s'; command: %d",
                      thd->get_protocol_classic()->get_packet_length(),
                      thd->get_protocol_classic()->get_raw_packet(), command));
  if (thd->get_protocol_classic()->bad_packet)
    DBUG_ASSERT(0);  // Should be caught earlier

  // Reclaim some memory
  thd->get_protocol_classic()->get_output_packet()->shrink(
      thd->variables.net_buffer_length);
  /* Restore read timeout value */
  my_net_set_read_timeout(net, thd->variables.net_read_timeout);

  return_value = dispatch_command(thd, &com_data, command);
  thd->get_protocol_classic()->get_output_packet()->shrink(
      thd->variables.net_buffer_length);

out:
  /* The statement instrumentation must be closed in all cases. */
  DBUG_ASSERT(thd->m_digest == NULL);
  DBUG_ASSERT(thd->m_statement_psi == NULL);
  DBUG_RETURN(return_value);
}
  • 主要的处理逻辑为dispatch_command,根据不同的command类型进行分发。
/**
  Perform one connection-level (COM_XXXX) command.

  @param thd             connection handle
  @param command         type of command to perform
  @param com_data        com_data union to store the generated command

  @todo
    set thd->lex->sql_command to SQLCOM_END here.
  @todo
    The following has to be changed to an 8 byte integer

  @retval
    0   ok
  @retval
    1   request of thread shutdown, i. e. if command is
        COM_QUIT
*/
bool dispatch_command(THD *thd, const COM_DATA *com_data,
                      enum enum_server_command command) {
  ... //太长不看
  
  switch (command) {
    case ... //太长不看
    case COM_QUERY: {
      ... //太长不看
      mysql_parse(thd, &parser_state);

      ... //太长不看

      DBUG_PRINT("info", ("query ready"));
      break;
    }
    case ... //太长不看
    default:
      my_error(ER_UNKNOWN_COM_ERROR, MYF(0));
      break;
  }
}
  • 主要看COM_QUERY这个逻辑,我们要用到的DDL、DML都会走这个流程,这个流程中主要是调用mysql_parse方法
/**
  Parse a query.

  @param thd          Current session.
  @param parser_state Parser state.
*/

void mysql_parse(THD *thd, Parser_state *parser_state) {
  ... //太长不看

  mysql_reset_thd_for_next_command(thd);
  
  if (!err) {
    err = parse_sql(thd, parser_state, NULL);
    ... //太长不看
  }

  if (!err) {
    
    mysql_rewrite_query(thd);

    ... //太长不看
  }

  if (!err) {
    ...
    error = mysql_execute_command(thd, true);
    ...
  }
}
  • 主要是SQL语法解析和执行
  • mysql_reset_thd_for_next_command是对下一次执行做准备,重置线程各变量
  • mysql_rewrite_query看着像是SQL优化?待定 还没追进去,记个TODO
  • 词法解析前不应该有缓存吗?没有找到缓存的逻辑,记个TODO

关闭连接

sql/conn_handler/connection_handler_pre_thread.cc

Channel_info *Per_thread_connection_handler::block_until_new_connection() {
  Channel_info *new_conn = NULL;
  mysql_mutex_lock(&LOCK_thread_cache);
  if (blocked_pthread_count < max_blocked_pthreads && !shrink_cache) {
    /* Don't kill the pthread, just block it for reuse */
    DBUG_PRINT("info", ("Blocking pthread for reuse"));

    /*
      mysys_var is bound to the physical thread,
      so make sure mysys_var->dbug is reset to a clean state
      before picking another session in the thread cache.
    */
    DBUG_POP();
    DBUG_ASSERT(!_db_is_pushed_());

    // Block pthread
    blocked_pthread_count++;
    while (!connection_events_loop_aborted() && !wake_pthread && !shrink_cache)
      mysql_cond_wait(&COND_thread_cache, &LOCK_thread_cache);
    blocked_pthread_count--;

    if (shrink_cache && blocked_pthread_count <= max_blocked_pthreads) {
      mysql_cond_signal(&COND_flush_thread_cache);
    }

    if (wake_pthread) {
      wake_pthread--;
      if (!waiting_channel_info_list->empty()) {
        new_conn = waiting_channel_info_list->front();
        waiting_channel_info_list->pop_front();
        DBUG_PRINT("info", ("waiting_channel_info_list->pop %p", new_conn));
      } else {
        DBUG_ASSERT(0);  // We should not get here.
      }
    }
  }
  mysql_mutex_unlock(&LOCK_thread_cache);
  return new_conn;
}
  • 如果阻塞的线程数小于最大阻塞线程数,则此线程不回收,而是进入阻塞状态(等待),等待新连接来的时候重复使用。
  • 否则关闭线程。

参考文献:
https://www.cnblogs.com/FateTHarlaown/p/8676166.html

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
5天前
|
Java 数据库连接 测试技术
SpringBoot入门(4) - 添加内存数据库H2
SpringBoot入门(4) - 添加内存数据库H2
27 4
SpringBoot入门(4) - 添加内存数据库H2
|
7天前
|
Java 数据库连接 测试技术
SpringBoot入门(4) - 添加内存数据库H2
SpringBoot入门(4) - 添加内存数据库H2
16 2
SpringBoot入门(4) - 添加内存数据库H2
|
2天前
|
存储 Java 关系型数据库
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践,包括连接创建、分配、复用和释放等操作,并通过电商应用实例展示了如何选择合适的连接池库(如HikariCP)和配置参数,实现高效、稳定的数据库连接管理。
8 2
|
3天前
|
关系型数据库 MySQL Linux
在 CentOS 7 中通过编译源码方式安装 MySQL 数据库的详细步骤,包括准备工作、下载源码、编译安装、配置 MySQL 服务、登录设置等。
本文介绍了在 CentOS 7 中通过编译源码方式安装 MySQL 数据库的详细步骤,包括准备工作、下载源码、编译安装、配置 MySQL 服务、登录设置等。同时,文章还对比了编译源码安装与使用 RPM 包安装的优缺点,帮助读者根据需求选择最合适的方法。通过具体案例,展示了编译源码安装的灵活性和定制性。
22 2
|
6天前
|
存储 关系型数据库 MySQL
MySQL vs. PostgreSQL:选择适合你的开源数据库
在众多开源数据库中,MySQL和PostgreSQL无疑是最受欢迎的两个。它们都有着强大的功能、广泛的社区支持和丰富的生态系统。然而,它们在设计理念、性能特点、功能特性等方面存在着显著的差异。本文将从这三个方面对MySQL和PostgreSQL进行比较,以帮助您选择更适合您需求的开源数据库。
29 4
|
8天前
|
存储 人工智能 Java
Neo4j从入门到精通:打造高效知识图谱数据库 | AI应用开发
在大数据和人工智能时代,知识图谱作为一种高效的数据表示和查询方式,逐渐受到广泛关注。本文从入门到精通,详细介绍知识图谱及其存储工具Neo4j,涵盖知识图谱的介绍、Neo4j的特点、安装步骤、使用方法(创建、查询)及Cypher查询语言的详细讲解。通过本文,读者将全面了解如何利用Neo4j处理复杂关系数据。【10月更文挑战第14天】
39 6
|
10天前
|
存储 关系型数据库 MySQL
基于案例分析 MySQL 权限认证中的具体优先原则
【10月更文挑战第26天】本文通过具体案例分析了MySQL权限认证中的优先原则,包括全局权限、数据库级别权限和表级别权限的设置与优先级。全局权限优先于数据库级别权限,后者又优先于表级别权限。在权限冲突时,更严格的权限将被优先执行,确保数据库的安全性与资源合理分配。
|
1天前
|
监控 关系型数据库 MySQL
数据库优化:MySQL索引策略与查询性能调优实战
【10月更文挑战第27天】本文深入探讨了MySQL的索引策略和查询性能调优技巧。通过介绍B-Tree索引、哈希索引和全文索引等不同类型,以及如何创建和维护索引,结合实战案例分析查询执行计划,帮助读者掌握提升查询性能的方法。定期优化索引和调整查询语句是提高数据库性能的关键。
5 0
|
2天前
|
监控 关系型数据库 MySQL
数据库优化:MySQL索引策略与查询性能调优实战
【10月更文挑战第26天】数据库作为现代应用系统的核心组件,其性能优化至关重要。本文主要探讨MySQL的索引策略与查询性能调优。通过合理创建索引(如B-Tree、复合索引)和优化查询语句(如使用EXPLAIN、优化分页查询),可以显著提升数据库的响应速度和稳定性。实践中还需定期审查慢查询日志,持续优化性能。
12 0
|
10天前
|
SQL 数据库
SQL数据库基础语法入门
[link](http://www.vvo.net.cn/post/082935.html)