分布式 | DBLE 后端连接模块源码解读

于由社区 路路发布

作者简介

路路，热爱技术、乐于分享的技术人，目前主要从事数据库相关技术的研究。

前言
对于数据库中间件来说，后端连接模块是重中之重，也是难点。连接出现问题通常很难排查，需要具备网络抓包以及 MySQL 协议的知识才能进行定位。
本文将从源码层面，力求详尽的分析后端连接模块。包括连接模块的设计、连接池的初始化、连接的创建以及获取等内容，希望能够帮助到大家。
将内容划分为：后端连接的方案设计、源码解读，两个方面来进行叙述。
一、后端连接模块方案设计
设计思想
DBLE 源于 Mycat，我们先来看下 Mycat 的连接池设计思想。
Mycat 为了最高效的利用后端的 MySQL 连接，采取了不同于 Cobar 也不同于传统 JDBC 连接池的做法。传统的做法是基于 Database 的连接池，即一个 MySQL 服务器上有 5 个 Database。如果数据库最大连接为 1000，则每个 Database 独占 200 个连接。这种模式的最大问题在于，将一个数据库所具备的最大 1000 个连接，隔离成了更新小的连接池，于是可能产生某个应用的连接不够，但其他应用的连接却很空闲，此类资源浪费的情况。而对于分片这种场景，这个缺陷则几乎是致命的。因为每个分片所对应的 Database 的连接数量被限制在了一个很小的范围内，从而导致系统并发能力的大幅降低。而 Mycat 则采用了基于 MySQL 实例的连接池模式，每个 Database 都可以用现有的 1000 个连接中的空闲连接（通过后续的源码解读，大家能够更清楚这里的设计）。
连接池的结构设计
DBLE 的后端连接池基本分为两个部分，一个是在空闲的部分，一个是等待响应的部分 。在空闲的部分基本结构符合在 schema.xml 配置文件里面的配置结构，如下为 DBLE 后端连接池的模块结构图（来源于 DBLE 官方文档）：

对照后端连接池模块的结构图，我们再来看下 schema.xml 配置文件：

上图中将配置文件中的主要结构与相关的类对应起来了，后续源码阅读也将主要从这几个类展开。继续看下连接池相关类的关系：

从上图可以看出 PhysicalDBPool 类关联了多个 PhysicalDatasource 类，PhysicalDatasource 类中也存在 PhysicalDBPool 类的引用，这与 schema.xml 配置文件中所展示的一样。
连接池模块还涉及到其他的一些类，我们这里一并看下：
关于上图的说明如下：
1. PhysicalDatasource 类引用了 ConMap 类来存储后端连接
2. ConMap 类中又引用了 ConQueue 类，ConQueue 类中才是真正存储了后端连接（ConQueue 类关联了多个 BackendConnection 类，BackendConnection 类为后端连接的抽象接口）
上述类图关系也可以从代码中看出：
在 DBLE 的连接池里，当前可用的 MySQL 连接是放到一个 HashMap 的数据结构里（对应 ConMap 类中的 items 变量），Key 为当前连接对应的 Database，另外还有二级分类，即按照连接是自动提交还是手动提交模式进行区分（对应 ConQueue 类中的 autoCommitCons 变量和 manCommitCons 变量），这个设计是为了高效的查询匹配可用的连接，具体逻辑如下：
当某个用户会话需要一个自动提交的，到分片 dn1 (对应 db1）的 SQL 连接的时候，连接池首先找是否有 db1 上的可用连接，如果有，看是否有自动提交模式的连接，找到就返回，否则返回 db1 上的手动提交模式的连接，若没有 db1 的可用连接，则随机返回一个其他 db 对应的可用连接，若没有可用连接，并且连接池还没达到上限，则创建一个新连接并返回，这个逻辑过程，我们会发现，用户会话得到的连接可能不是他原先想要的，比如 Database 不对应，或者事务模式不匹配，因此在执行具体的 SQL 之前，还有一个自动同步数据库连接的过程，包括事务隔离级别、事务模式、字符集、Database 等四个指标，同步完成以后，才会执行具体的 SQL 指令，这里就是为什么 Mycat 能够使用基于 MySQL 实例的连接池的原因。
上述逻辑对应的代码在 ConMap#tryTakeCon 中：
  1. public BackendConnection tryTakeCon(final String schema, boolean autoCommit) {

  3.         final ConQueue queue = items.get(schema == null ? KEY_STRING_FOR_NULL_DATABASE : schema);
  4.         BackendConnection con = null;
  5.         if (queue != null) {
  6.             con = tryTakeCon(queue, autoCommit);
  7.         }
  8.         if (con != null) {
  9.             return con;
  10.         } else {
  11.             for (ConQueue queue2 : items.values()) {
  12.                 if (queue != queue2) {
  13.                     con = tryTakeCon(queue2, autoCommit);
  14.                     if (con != null) {
  15.                         return con;
  16.                     }
  17.                 }
  18.             }
  19.         }
  20.         return null;

  22.     }

  24.     private BackendConnection tryTakeCon(ConQueue queue, boolean autoCommit) {

  26.         BackendConnection con;
  27.         if (queue != null && ((con = queue.takeIdleCon(autoCommit)) != null)) {
  28.             return con;
  29.         } else {
  30.             return null;
  31.         }
  32.     }
该方法是获取一个可用连接，代码的逻辑中，首先看对应的 Database 上是否有可用连接，如果有就立即返回，否则从其他的 Database 上找一个可用连接返回。
MySQLConnection 类为具体的 MySQL 连接对象， synAndDoExecute 方法则判断获取到的连接是否符合要求，若不符合要求，先同步状态，然后执行具体的SQL。
通过共享一个 MySQL 上的所有物理连接，并结合连接状态同步的特性，Mycat 的连接池做到了最佳的吞吐量，也在一定程度上提升了整个系统的并发支撑能力。
二、源码解读
了解了Dble后端连接池的相关设计思想，下面我们就从源码层面走读整个后端连接池的初始化、连接创建与获取的过程。
数据库连接池的数据结构的创建在启动时候配置加载阶段完成，DbleServer#startup方法中相关代码如下：
  1. public void startup() throws IOException {
  2.         this.config = new ServerConfig();
  3.         ……
  4. }

继续看 ServerConfig 类的构造方法：

  1. public ServerConfig() {
  2.         //创建连接池的逻辑在这一行代码
  3.         ConfigInitializer confInit = new ConfigInitializer(true, false);
  4.         this.system = confInit.getSystem();
  5.         this.users = confInit.getUsers();
  6.         this.schemas = confInit.getSchemas();
  7.         /*
  8.             初始化后的 datahost 数据结构存储在 ServerConfig 类的dataHosts：Map<String, PhysicalDBPool>变量中，其中 key 值为 schema.xml 中配置的 dataHost 的 name 值
  9.         */
  10.         this.dataHosts = confInit.getDataHosts();
  11.         /*
  12.             初始化后的 datanode 数据结构存储在 ServerConfig 类的dataNodes ：Map<String, PhysicalDBNode>变量中，其中key值为 schema.xml 中配置的 datanode 的 name 值
  13.         */
  14.         this.dataNodes = confInit.getDataNodes();
  15.         this.erRelations = confInit.getErRelations();
  16.         this.dataHostWithoutWR = confInit.isDataHostWithoutWH();
  17.         ConfigUtil.setSchemasForPool(dataHosts, dataNodes);
  18.         ……
  19. }
继续走读 ConfigInitializer 类的构造方法：
  1. public ConfigInitializer(boolean loadDataHost, boolean lowerCaseNames) {
  2.         //load server.xml
  3.         XMLServerLoader serverLoader = new XMLServerLoader(this);

  5.         //load rule.xml and schema.xml
  6.         SchemaLoader schemaLoader = new XMLSchemaLoader(lowerCaseNames, this);
  7.         //下面这两个方法分别初始化了 dataHosts：Map<String, PhysicalDBPool>和dataNodes：Map<String, PhysicalDBNode>变量
  8.         this.dataHosts = initDataHosts(schemaLoader);
  9.         this.dataNodes = initDataNodes(schemaLoader);
  10. }
继续查看 ConfigInitializer#initDataHosts 和 ConfigInitializer#initDataNodes 这两个方法：
  1. //这里即根据配置文件生成相应的 Map 结构
  2. private Map<String, PhysicalDBPool> initDataHosts(SchemaLoader schemaLoader) {
  3.         Map<String, DataHostConfig> nodeConf = schemaLoader.getDataHosts();
  4.         //create PhysicalDBPool according to DataHost
  5.         Map<String, PhysicalDBPool> nodes = new HashMap<>(nodeConf.size());
  6.         for (DataHostConfig conf : nodeConf.values()) {
  7.             //create PhysicalDBPool
  8.             PhysicalDBPool pool = getPhysicalDBPool(conf);
  9.             nodes.put(pool.getHostName(), pool);
  10.         }
  11.         return nodes;
  12.     }
  13. //这里也是根据配置文件生成相应的 Map 结构
  14. private Map<String, PhysicalDBNode> initDataNodes(SchemaLoader schemaLoader) {
  15.         Map<String, DataNodeConfig> nodeConf = schemaLoader.getDataNodes();
  16.         Map<String, PhysicalDBNode> nodes = new HashMap<>(nodeConf.size());
  17.         for (DataNodeConfig conf : nodeConf.values()) {
  18.             //PhysicalDBNode 类里存放了 PhysicalDBPool 类的引用
  19.             PhysicalDBPool pool = this.dataHosts.get(conf.getDataHost());
  20.             if (pool == null) {
  21.                 throw new ConfigException("dataHost not exists " + conf.getDataHost());
  22.             }
  23.             PhysicalDBNode dataNode = new PhysicalDBNode(conf.getName(), conf.getDatabase(), pool);
  24.             nodes.put(dataNode.getName(), dataNode);
  25.         }
  26.         return nodes;
  27.     }

到这里，后端连接相关的数据结构都已经初始化完成了（注意这里只是创建了相应的数据结构，里面还没有存放实际的后端MySQL连接），我们看到有 PhysicalDBPool 类和 PhysicalDBNode 类， PhysicalDBNode是 DBLE 分片（Datanode）的对应，引用一个连接池对象 PhysicalDBPool ， PhysicalDBPool 里面引用了真正的连接池对象 PhysicalDatasource，并且按照读节点和写节点分开引用，实现读写分类和节点切换的功能。

这两个类的关系可以看一下类图：

可以看到 PhysicalDBNode 类关联了 PhysicalDBPool 类，即对应每个 datanode 关联一个 datahost 连接池。
后端连接相关的数据结构都已经初始化完成了，但我们并没有看到创建连接放到对应的连接池里的代码逻辑，那创建连接的代码逻辑在哪里呢？
答案还是在 DbleServer#startup 方法中：
  1. public void startup() throws IOException {
  2.         ……
  3.         //关键代码在这一行（光看方法名真是不知道这个创建后端连接的方法）
  4.         pullVarAndMeta();
  5.         ……
来看一下 DbleServer#pullVarAndMeta 方法：
  1. private void pullVarAndMeta() throws IOException {
  2.         ……
  3.         //直接看关键代码，是下面这个方法进行初始化连接池的操作
  4.         initDataHost();
  5.         ……
继续走读 DbleServer#initDataHost 方法：
  1. private void initDataHost() {
  2.         // init datahost
  3.         Map<String, PhysicalDBPool> dataHosts = this.getConfig().getDataHosts();
  4.         LOGGER.info("Initialize dataHost ...");
  5.         //下面的代码便是根据配置文件里的 datahosts 开始遍历初始化相应的连接池了
  6.         for (PhysicalDBPool node : dataHosts.values()) {
  7.             //这里的 index 为配置多个 writehost 的时候，默认初始化哪一个，默认是 0，即第一个配置的 writehost
  8.             String index = dnIndexProperties.getProperty(node.getHostName(), "0");
  9.             if (!"0".equals(index)) {
  10.                 LOGGER.info("init datahost: " + node.getHostName() + "  to use datasource index:" + index);
  11.             }
  12.             //初始化连接池的方法
  13.             node.init(Integer.parseInt(index));
  14.             node.startHeartbeat();
  15.         }
  16.     }
现在跟着代码走到 PhysicalDBPool#init 方法：
  1. public int init(int index) {
  2.         if (!checkIndex(index)) {
  3.             index = 0;
  4.         }

  6.         for (int i = 0; i < writeSources.length; i++) {
  7.             int j = loop(i + index);
  8.             //initSource() 方法中初始化了连接池
  9.             if (initSource(j, writeSources[j])) {
  10.                 //activedIndex 表明了当前是哪个写节点的数据源在生效
  11.                 activeIndex = j;
  12.                 initSuccess = true;
  13.                 LOGGER.info(getMessage(j, " init success"));
  14.                 return activeIndex;
  15.             }
  16.         }

  18.         initSuccess = false;
  19.         LOGGER.warn(hostName + " init failure");
  20.         return -1;
  21.     }
下面我们就来详细分析下 PhysicalDBPool#initSource 方法：
  1. private boolean initSource(int index, PhysicalDatasource ds) {
  2.         if (ds.getConfig().isDisabled()) {
  3.             LOGGER.info(ds.getConfig().getHostName() + " is disabled, skipped");
  4.             return true;
  5.         }
  6.         //获取 datasource 的最小连接数，从这里也可以看出来，datahost 里配置的最小连接数范围针对下面配置的写节点、读节点
  7.         int initSize = ds.getConfig().getMinCon();
  8.         //最小连接数不能小于对应数据源的数据库 schema 数+1
  9.         if (initSize < this.schemas.length + 1) {
  10.             initSize = this.schemas.length + 1;
  11.             LOGGER.warn("minCon size is less than (the count of schema +1), so dble will create at least 1 conn for every schema and an empty schema conn");
  12.         }
  13.         //最大连接数不能小于最小连接数的判断
  14.         if (ds.getConfig().getMaxCon() < initSize) {
  15.             ds.getConfig().setMaxCon(initSize);
  16.             ds.setSize(initSize);
  17.             LOGGER.warn("maxCon is less than the initSize of dataHost:" + initSize + " change the maxCon into " + initSize);
  18.         }

  20.         LOGGER.info("init backend mysql source ,create connections total " + initSize + " for " + ds.getName() +
  21.                 " index :" + index);
  22.         //存放创建好的连接的数据结构
  23.         CopyOnWriteArrayList<BackendConnection> list = new CopyOnWriteArrayList<>();
  24.         //连接池初始化完成后的回调函数，后面会详细分析
  25.         GetConnectionHandler getConHandler = new GetConnectionHandler(list, initSize);
  26.         boolean hasConnectionInPool = false;
  27.         try {
  28.             //这里是初始化schema为null的一条连接
  29.             if (ds.getTotalConCount() <= 0) {
  30.                 ds.initMinConnection(null, true, getConHandler, null);
  31.             } else {
  32.                 LOGGER.info("connection with null schema do not create,because testConnection in pool");
  33.                 getConHandler.initIncrement();
  34.                 hasConnectionInPool = true;
  35.             }
  36.         } catch (Exception e) {
  37.             LOGGER.warn("init connection with schema null error", e);
  38.         }
  39.         //根据初始化连接池大小，正式初始化连接池中的连接
  40.         for (int i = 0; i < initSize - 1; i++) {
  41.             try {
  42.                 //PhysicalDataSource#initMinConnection方法为初始化单条后端连接的方法
  43.                 ds.initMinConnection(this.schemas[i % schemas.length], true, getConHandler, null);
  44.             } catch (Exception e) {
  45.                 LOGGER.warn(getMessage(index, " init connection error."), e);
  46.             }
  47.         }


  50.         long timeOut = System.currentTimeMillis() + 60 * 1000;

  52.         //等待所有初始化连接创建完成
  53.         while (!getConHandler.finished() && (System.currentTimeMillis() < timeOut)) {
  54.             try {
  55.                 Thread.sleep(100);
  56.             } catch (InterruptedException e) {
  57.                 /*
  58.                  * hardly triggered no error is needed
  59.                  */
  60.                 LOGGER.info("initError", e);
  61.             }
  62.         }
  63.         LOGGER.info("init result :" + getConHandler.getStatusInfo());
  64.         return !list.isEmpty() || hasConnectionInPool;
  65.     }
通过查看代码，我们知道 PhysicalDataSource#initMinConnection 方法为初始化单条后端连接的方法，继续查看该代码：
  1. public void initMinConnection(String schema, boolean autocommit, final ResponseHandler handler,
  2.                                   final Object attachment) throws IOException {
  3.         LOGGER.info("create new connection for " +
  4.                 this.name + " of schema " + schema);
  5.         //先判断是否还需创建新连接，防止创建连接过多
  6.         if (this.createNewCount()) {
  7.             //继续调用createNewConnection方法
  8.             createNewConnection(handler, attachment, schema);
  9.         }
  10.     }
跟着代码继续查看 PhysicalDataSource#createNewConnection方法：
  1. private void createNewConnection(final ResponseHandler handler, final Object attachment,
  2.                                      final String schema) throws IOException {
  3.         // 这里是异步创建连接
  4.         DbleServer.getInstance().getComplexQueryExecutor().execute(new Runnable() {
  5.             public void run() {
  6.                 try {
  7.                     //继续调用了createNewConnection方法，并且这里定义了内部类DelegateResponseHandler，注意这里的handler变量代表的是哪个类还记得吗？提醒一下大家这里是GetConnectionHandler类，这么多层调用很容易被搞迷糊了。
  8.                     createNewConnection(new DelegateResponseHandler(handler) {
  9.                         @Override
  10.                         public void connectionError(Throwable e, BackendConnection conn) {
  11.                             Map<String, String> labels = AlertUtil.genSingleLabel("data_host", hostConfig.getName() + "-" + config.getHostName());
  12.                             AlertUtil.alert(AlarmCode.CREATE_CONN_FAIL, Alert.AlertLevel.WARN, "createNewConn Error" + e.getMessage(), "mysql", config.getId(), labels);
  13.                             ToResolveContainer.CREATE_CONN_FAIL.add(hostConfig.getName() + "-" + config.getHostName());
  14.                             handler.connectionError(e, conn);
  15.                         }

  17.                         @Override
  18.                         public void connectionAcquired(BackendConnection conn) {
  19.                             takeCon(conn, handler, attachment, schema);
  20.                         }
  21.                     }, schema);
  22.                 } catch (IOException e) {
  23.                     handler.connectionError(e, null);
  24.                 }
  25.             }
  26.         });
  27.     }
上述方法异步提交了创建连接的任务，我们继续看下创建连接的任务里做了什么， PhysicalDatasource#createNewConnection 方法是个抽象方法：
  1. public abstract void createNewConnection(ResponseHandler handler, String schema) throws IOException;

具体实现看 MySQLDataSource#createNewConnection 方法：

  1. //这里调用了 MySQLConnectionFactory#make 方法创建连接
  2. public void createNewConnection(ResponseHandler handler, String schema) throws IOException {
  3.         //注意这里的 handler为DelegateResponseHandler，即在PhysicalDataSource#createNewConnection方法里定义的匿名内部类
  4.         factory.make(this, handler, schema);
  5.     }
MySQLConnectionFactory#make 我们也来看一下，这个方法涉及到网络方面了，我们都知道Dble的连接用了reactor设计模式，这里网络io层面我们就不再深入了，我们只需要知道连接通过网络建立之后，回调了哪个方法就行：
  1. public MySQLConnection make(MySQLDataSource pool, ResponseHandler handler,
  2.                                 String schema) throws IOException {

  4.         DBHostConfig dsc = pool.getConfig();
  5.         NetworkChannel channel = openSocketChannel(DbleServer.getInstance().isAIO());

  7.         MySQLConnection c = new MySQLConnection(channel, pool.isReadNode(), schema == null);
  8.         c.setSocketParams(false);
  9.         c.setHost(dsc.getIp());
  10.         c.setPort(dsc.getPort());
  11.         c.setUser(dsc.getUser());
  12.         c.setPassword(dsc.getPassword());
  13.         c.setSchema(schema);
  14.         //这里是连接建立后的回调类MySQLConnectionAuthenticator，注意这里的handler变量为PhysicalDataSource#createNewConnection方法里定义的匿名内部类DelegateResponseHandler
  15.         c.setHandler(new MySQLConnectionAuthenticator(c, handler));
  16.         c.setPool(pool);
  17.         c.setIdleTimeout(pool.getConfig().getIdleTimeout());
  18.         if (channel instanceof AsynchronousSocketChannel) {
  19.             ((AsynchronousSocketChannel) channel).connect(
  20.                     new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
  21.                     (CompletionHandler) DbleServer.getInstance().getConnector());
  22.         } else {
  23.             ((NIOConnector) DbleServer.getInstance().getConnector()).postConnect(c);
  24.         }
  25.         return c;
  26.     }
我们直接来看通过网络建立连接后的回调逻辑，进入 MySQLConnectionAuthenticator#handle 方法：
  1. public void handle(byte[] data) {
  2.           ……
  3.           //省略不相关的代码，在后端连接建立成功后，会直接调用PhysicalDataSource#createNewConnection方法里定义的匿名内部类的connectionAcquired方法
  4.           listener.connectionAcquired(source);
  5.           ……
  6.     }
转了好大的一圈，终于回来了，继续看 PhysicalDataSource#createNewConnection 方法：
  1. private void createNewConnection(final ResponseHandler handler, final Object attachment,
  2.                                      final String schema) throws IOException {
  3.         // aysn create connection
  4.         DbleServer.getInstance().getComplexQueryExecutor().execute(new Runnable() {
  5.             public void run() {
  6.                 try {
  7.                     createNewConnection(new DelegateResponseHandler(handler) {
  8.                         @Override
  9.                         public void connectionError(Throwable e, BackendConnection conn) {
  10.                             Map<String, String> labels = AlertUtil.genSingleLabel("data_host", hostConfig.getName() + "-" + config.getHostName());
  11.                             AlertUtil.alert(AlarmCode.CREATE_CONN_FAIL, Alert.AlertLevel.WARN, "createNewConn Error" + e.getMessage(), "mysql", config.getId(), labels);
  12.                             ToResolveContainer.CREATE_CONN_FAIL.add(hostConfig.getName() + "-" + config.getHostName());
  13.                             handler.connectionError(e, conn);
  14.                         }
  15.                         //回调的是这里的方法，这里的 handler 还记得吗？别忘了是 GetConnectionHandler 类
  16.                         @Override
  17.                         public void connectionAcquired(BackendConnection conn) {
  18.                             takeCon(conn, handler, attachment, schema);
  19.                         }
  20.                     }, schema);
  21.                 } catch (IOException e) {
  22.                     handler.connectionError(e, null);
  23.                 }
  24.             }
  25.         });
  26.     }
继续看 PhysicalDatasource#takeCon 方法：
  1. private void takeCon(BackendConnection conn,
  2.                          final ResponseHandler handler, final Object attachment,
  3.                          String schema) {
  4.         if (ToResolveContainer.CREATE_CONN_FAIL.contains(this.getHostConfig().getName() + "-" + this.getConfig().getHostName())) {
  5.             Map<String, String> labels = AlertUtil.genSingleLabel("data_host", this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  6.             AlertUtil.alertResolve(AlarmCode.CREATE_CONN_FAIL, Alert.AlertLevel.WARN, "mysql", this.getConfig().getId(), labels,
  7.                     ToResolveContainer.CREATE_CONN_FAIL, this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  8.         }
  9.         //这个方法是创建连接所属schema的连接队列
  10.         takeCon(conn, schema);
  11.         conn.setAttachment(attachment);
  12.         //调用了GetConnectionHandler类的connectionAcquired方法
  13.         handler.connectionAcquired(conn);
  14.     }
继续让我们看一下 GetConnectionHandler#connectionAcquired方法（这个变量从传入到现在，终于用上了）：
  1. public void connectionAcquired(BackendConnection conn) {
  2.         //将成功创建的连接放入成功连接列表
  3.         successCons.add(conn);
  4.         finishedCount.addAndGet(1);
  5.         LOGGER.info("connected successfully " + conn);
  6.         //重点！这里便是将新建的连接放入连接池的方法
  7.         conn.release();
  8.     }

继续看 MySQLConnection#release 方法：

  1. public void release() {
  2.         ……
  3.         complexQuery = false;
  4.         metaDataSynced = true;
  5.         attachment = null;
  6.         statusSync = null;
  7.         modifiedSQLExecuted = false;
  8.         isDDL = false;
  9.         testing = false;
  10.         setResponseHandler(null);
  11.         setSession(null);
  12.         logResponse.set(false);
  13.         //这里真正将连接放入连接池，该方法为`PhysicalDatasource#releaseChannel`方法
  14.         pool.releaseChannel(this);
  15.     }
让我们来到 PhysicalDatasource#releaseChannel 方法：
  1. public void releaseChannel(BackendConnection c) {
  2.         if (LOGGER.isDebugEnabled()) {
  3.             LOGGER.debug("release channel " + c);
  4.         }
  5.         //将连接放入连接池
  6.         returnCon(c);
  7.     }
继续看 PhysicalDatasource#returnCon 方法：
  1. private void returnCon(BackendConnection c) {
  2.         if (c.isClosedOrQuit()) {
  3.             return;
  4.         }
  5.         c.setAttachment(null);
  6.         c.setBorrowed(false);
  7.         c.setLastTime(TimeUtil.currentTimeMillis());

  9.         String errMsg = null;

  11.         boolean ok;
  12.         //根据连接的schema获取到ConQueue连接队列
  13.         ConQueue queue = this.conMap.createAndGetSchemaConQueue(c.getSchema());
  14.         //判断连接是自动提交还是手动提交，将连接放入相应的队列中
  15.         if (c.isAutocommit()) {
  16.             ok = queue.getAutoCommitCons().offer(c);
  17.         } else {
  18.             ok = queue.getManCommitCons().offer(c);
  19.         }
  20.         if (!ok) {
  21.             errMsg = "can't return to pool ,so close con " + c;
  22.         }
  23.         if (errMsg != null) {
  24.             LOGGER.info(errMsg);
  25.             c.close(errMsg);
  26.         }
  27.     }
到这里，一个连接从创建到放入连接池的逻辑都走通了，不得不感慨一句，DBLE 的调用真的是有点多，有点复杂呀！特别是异步回调，很容易把人搞蒙。
一不做二不休，我们再来看看连接获取的逻辑。
我们都知道 DBLE 是先对发送过来的 SQL 进行路由，然后根据路由结果到相应的节点执行 SQL 的。执行 SQL 当然要用连接了，我们下面便通过单节点下如何执行 SQL，来看连接的获取逻辑。
连接获取有两种情况：一是从上述创建的连接池中获取；二是新建连接（当连接池中的连接被用完的时候），下面分别说明。
单节点 SQL 的执行代码在 SingleNodeHandler#execute 方法中：
  1. public void execute() throws Exception {
  2.         connClosed = false;
  3.         this.packetId = (byte) session.getPacketId().get();
  4.         //这里是先判断会话有没有取过连接，如果有的话就不再取
  5.         if (session.getTargetCount() > 0) {
  6.             BackendConnection conn = session.getTarget(node);
  7.             if (conn == null && rrs.isGlobalTable() && rrs.getGlobalBackupNodes() != null) {
  8.                 for (String dataNode : rrs.getGlobalBackupNodes()) {
  9.                     RouteResultsetNode tmpNode = new RouteResultsetNode(dataNode, rrs.getSqlType(), rrs.getStatement());
  10.                     conn = session.getTarget(tmpNode);
  11.                     if (conn != null) {
  12.                         break;
  13.                     }
  14.                 }
  15.             }
  16.             node.setRunOnSlave(rrs.getRunOnSlave());
  17.             if (session.tryExistsCon(conn, node)) {
  18.                 execute(conn);
  19.                 return;
  20.             }
  21.         }

  23.         // 我们重点看下获取新连接的逻辑
  24.         node.setRunOnSlave(rrs.getRunOnSlave());
  25.         ServerConfig conf = DbleServer.getInstance().getConfig();
  26.         PhysicalDBNode dn = conf.getDataNodes().get(node.getName());
  27.         //PhysicalDBNode有相应PhysicalDBPool类的引用，这里调用了PhysicalDBNode#getConnection方法获取连接
  28.         dn.getConnection(dn.getDatabase(), session.getSource().isTxStart(), session.getSource().isAutocommit(), node, this, node);
  29.     }
查看 PhysicalDBNode#getConnection 方法：
  1. public void getConnection(String schema, boolean isMustWrite, boolean autoCommit, RouteResultsetNode rrs,
  2.                               ResponseHandler handler, Object attachment) throws Exception {
  3.         //判断是否必须走写节点
  4.         if (isMustWrite) {
  5.             //获取写节点的连接
  6.             getWriteNodeConnection(schema, autoCommit, handler, attachment);
  7.             return;
  8.         }
  9.         if (rrs.getRunOnSlave() == null) {
  10.             if (rrs.canRunINReadDB(autoCommit)) {
  11.                 dbPool.getRWBalanceCon(schema, autoCommit, handler, attachment);
  12.             } else {
  13.                 getWriteNodeConnection(schema, autoCommit, handler, attachment);
  14.             }
  15.         } else {
  16.             if (rrs.getRunOnSlave()) {
  17.                 if (!dbPool.getReadCon(schema, autoCommit, handler, attachment)) {
  18.                     LOGGER.info("Do not have slave connection to use, use master connection instead.");
  19.                     rrs.setRunOnSlave(false);
  20.                     rrs.setCanRunInReadDB(false);
  21.                     getWriteNodeConnection(schema, autoCommit, handler, attachment);
  22.                 }
  23.             } else {
  24.                 rrs.setCanRunInReadDB(false);
  25.                 getWriteNodeConnection(schema, autoCommit, handler, attachment);
  26.             }
  27.         }
  28.     }
我们看下如何获取写节点的连接的，也就是对应配置文件中 writehost 中相应的连接池中的连接，查看 PhysicalDBNode#getWriteNodeConnection 方法：
  1. private void getWriteNodeConnection(String schema, boolean autoCommit, ResponseHandler handler, Object attachment) throws IOException {
  2.         checkRequest(schema);
  3.         if (dbPool.isInitSuccess()) {
  4.             PhysicalDatasource writeSource = dbPool.getSource();
  5.             if (writeSource.getConfig().isDisabled()) {
  6.                 throw new IllegalArgumentException("[" + writeSource.getHostConfig().getName() + "." + writeSource.getConfig().getHostName() + "] is disabled");
  7.             }
  8.             writeSource.setWriteCount();
  9.             //调用了PhysicalDatasource类的getConnection方法获取连接
  10.             writeSource.getConnection(schema, autoCommit, handler, attachment);
  11.         } else {
  12.             throw new IllegalArgumentException("Invalid DataSource:" + dbPool.getActiveIndex());
  13.         }
  14.     }
跟着代码进入 PhysicalDatasource#getConnection 方法：
  1. public void getConnection(String schema, boolean autocommit, final ResponseHandler handler,
  2.                               final Object attachment) throws IOException {
  3.         /*
  4.              这里获取连接有两种情况：
  5.                 1.从上述创建的连接池中获取；
  6.                 2.新建连接。
  7.         */
  8.         //1.从连接池中获取连接
  9.         BackendConnection con = this.conMap.tryTakeCon(schema, autocommit);
  10.         if (con != null) {
  11.             takeCon(con, handler, attachment, schema);
  12.         } else {
  13.             if (!this.createNewCount()) {
  14.                 String maxConError = "the max active Connections size can not be max than maxCon for data host[" + this.getHostConfig().getName() + "." + this.getName() + "]";
  15.                 LOGGER.warn(maxConError);
  16.                 Map<String, String> labels = AlertUtil.genSingleLabel("data_host", this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  17.                 AlertUtil.alert(AlarmCode.REACH_MAX_CON, Alert.AlertLevel.WARN, maxConError, "dble", this.getConfig().getId(), labels);
  18.                 ToResolveContainer.REACH_MAX_CON.add(this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  19.                 throw new IOException(maxConError);
  20.             } else { //  2.新建连接
  21.                 if (ToResolveContainer.REACH_MAX_CON.contains(this.getHostConfig().getName() + "-" + this.getConfig().getHostName())) {
  22.                     Map<String, String> labels = AlertUtil.genSingleLabel("data_host", this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  23.                     AlertUtil.alertResolve(AlarmCode.REACH_MAX_CON, Alert.AlertLevel.WARN, "dble", this.getConfig().getId(), labels,
  24.                             ToResolveContainer.REACH_MAX_CON, this.getHostConfig().getName() + "-" + this.getConfig().getHostName());

  26.                 }
  27.                 LOGGER.info("no idle connection in pool,create new connection for " + this.name + " of schema " + schema);
  28.                 createNewConnection(handler, attachment, schema);
  29.             }
  30.         }
  31.     }
我们先来看下从连接池中获取连接的方法，ConMap#tryTakeCon 方法，这个方法的代码我们在连接池方案设计小节已经讲过了，这里在看一下吧：
  1. public BackendConnection tryTakeCon(final String schema, boolean autoCommit) {

  3.         final ConQueue queue = items.get(schema == null ? KEY_STRING_FOR_NULL_DATABASE : schema);
  4.         BackendConnection con = null;
  5.         if (queue != null) {
  6.             con = tryTakeCon(queue, autoCommit);
  7.         }
  8.         if (con != null) {
  9.             return con;
  10.         } else {
  11.             for (ConQueue queue2 : items.values()) {
  12.                 if (queue != queue2) {
  13.                     con = tryTakeCon(queue2, autoCommit);
  14.                     if (con != null) {
  15.                         return con;
  16.                     }
  17.                 }
  18.             }
  19.         }
  20.         return null;

  22.     }

  24.     private BackendConnection tryTakeCon(ConQueue queue, boolean autoCommit) {

  26.         BackendConnection con;
  27.         if (queue != null && ((con = queue.takeIdleCon(autoCommit)) != null)) {
  28.             return con;
  29.         } else {
  30.             return null;
  31.         }
  32.     }
tryTakeCon 方法是获取一个可用连接，代码的逻辑中，首先看对应的Database 上是否有可用连接，如果有就立即返回，否则从其他的 Database 上找一个可用连接返回。
MySQLConnection 类的 synAndDoExecute 方法会判断获取到的连接是否符合要求，若不符合要求，会先同步状态，然后执行具体的SQL，所以这里可以用其他database上的连接。
成功获取连接后，看下后续的执行逻辑，调用了 PhysicalDatasource#takeCon 方法：
  1. private void takeCon(BackendConnection conn,
  2.                          final ResponseHandler handler, final Object attachment,
  3.                          String schema) {
  4.         if (ToResolveContainer.CREATE_CONN_FAIL.contains(this.getHostConfig().getName() + "-" + this.getConfig().getHostName())) {
  5.             Map<String, String> labels = AlertUtil.genSingleLabel("data_host", this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  6.             AlertUtil.alertResolve(AlarmCode.CREATE_CONN_FAIL, Alert.AlertLevel.WARN, "mysql", this.getConfig().getId(), labels,
  7.                     ToResolveContainer.CREATE_CONN_FAIL, this.getHostConfig().getName() + "-" + this.getConfig().getHostName());
  8.         }
  9.         //这里仅做连接的一些状态记录
  10.         takeCon(conn, schema);
  11.         conn.setAttachment(attachment);
  12.         //这里的handler为SingleNodeHandler类（别忘了我们是一路从SingleNodeHandler类走过来的）
  13.         handler.connectionAcquired(conn);
  14.     }
继续看下 SingleNodeHandler#connectionAcquired 方法：
  1. public void connectionAcquired(final BackendConnection conn) {
  2.         //将后端连接与会话绑定，后续如果再用连接的话，则可以直接取出
  3.         session.bindConnection(node, conn);
  4.         //获取连接后，就是执行具体的sql了
  5.         execute(conn);
  6.     }

从连接池中获取连接的过程如上所述，流程还是比较清晰的。

那么当连接池中没有连接可用了（连接被用完）怎么办呢？我们此时便需要通过创建新的连接来执行 SQL，新建连接的操作与连接池初始化过程中新建连接的流程一样（忘记的话可以回过头再看一遍），唯一的不同是回调类不同，针对本例回调类是 SingleNodeHandler 类：
  1. //新建连接主要方法
  2. private void createNewConnection(final ResponseHandler handler, final Object attachment,
  3.                                      final String schema) throws IOException {
  4.         // aysn create connection
  5.         DbleServer.getInstance().getComplexQueryExecutor().execute(new Runnable() {
  6.             public void run() {
  7.                 try {
  8.                     createNewConnection(new DelegateResponseHandler(handler) {
  9.                         @Override
  10.                         public void connectionError(Throwable e, BackendConnection conn) {
  11.                             Map<String, String> labels = AlertUtil.genSingleLabel("data_host", hostConfig.getName() + "-" + config.getHostName());
  12.                             AlertUtil.alert(AlarmCode.CREATE_CONN_FAIL, Alert.AlertLevel.WARN, "createNewConn Error" + e.getMessage(), "mysql", config.getId(), labels);
  13.                             ToResolveContainer.CREATE_CONN_FAIL.add(hostConfig.getName() + "-" + config.getHostName());
  14.                             handler.connectionError(e, conn);
  15.                         }

  17.                         @Override
  18.                         public void connectionAcquired(BackendConnection conn) {
  19.                             //这里的handler为SingleNodeHandler，新建连接成功后，将回调此处代码
  20.                             takeCon(conn, handler, attachment, schema);
  21.                         }
  22.                     }, schema);
  23.                 } catch (IOException e) {
  24.                     handler.connectionError(e, null);
  25.                 }
  26.             }
  27.         });
  28.     }
PhysicalDataSource#takecon 方法中，在此处为调用了 SingleNodeHandler#connectionAcquired 方法：
  1. private void takeCon(BackendConnection conn,
  2.                          final ResponseHandler handler, final Object attachment,
  3.                          String schema) {
  4.         ……
  5.         //调用了SingleNodeHandler#connectionAcquired方法
  6.         handler.connectionAcquired(conn);
  7.     }
SingleNodeHandler#connectionAcquired 方法：
  1. //与从连接池中获取连接的操作一样，先是绑定连接到会话，然后具体执行 SQL
  2. public void connectionAcquired(final BackendConnection conn) {
  3.         session.bindConnection(node, conn);
  4.         execute(conn);
  5.     }

连接的获取流程到这里应该也已经清楚了。

但还有一个疑问，没有从连接池中获取到而新建的连接使用完之后怎么处理呢？我们还是以单节点的处理为例，看下SingleNodeHandler#okResponse方法，为什么看这个方法呢？因为 SQL执行成功后，会返回 OK 包到 DBLE，这里便会回调此方法（回调确实多呢）：
  1. public void okResponse(byte[] data, BackendConnection conn) {
  2.             ……
  3.             //删除掉多余代码，重点看下面这一行代码，这一行代码便是将新建的连接放回连接池的代码了
  4.             session.releaseConnectionIfSafe(conn, false);
  5.             ……
  6.         }
  7.     }

将新建连接放入连接池的具体代码在 NonBlockingSession#releaseConnectionIfSafe 方法中，其实还是调用的上述初始化连接池过程中将连接放入连接池中的代码，所以这里不再赘述了。

总结

本文先介绍了 DBLE 数据库中间件的后端连接的设计思想、后端连接的数据结构，然后从源码角度详细走读了连接的创建以及获取过程，代码中涉及到很多的异步调用，不小心就会把人搞晕。连接模块对于理解数据库中间件很重要，而且连接出现问题通常都是很难排查的，希望通过本文能帮助大家理解 DBLE 的后端连接模块。后续如果忘记相应的逻辑，还可以回过头翻阅一下。

社区近期动态

No.1

Mycat 问题免费诊断


诊断范围支持：

Mycat 的故障诊断、源码分析、性能优化

服务支持渠道：

  1. 技术交流群，进群后可提问

    QQ群（669663113）

  2. 社区通道，邮件&电话

    osc@actionsky.com

  3. 现场拜访，线下实地，1天免费拜访

关注“爱可生开源社区”公众号，回复关键字“Mycat”，获取活动详情。


No.2

社区技术内容征稿


征稿内容：

  1. 格式：.md/.doc/.txt

  2. 主题：MySQL、分布式中间件DBLE、数据传输组件DTLE相关技术内容

  3. 要求：原创且未发布过

  4. 奖励：作者署名；200元京东E卡+社区周边

投稿方式：

  1. 邮箱：osc@actionsky.com

  2. 格式：[投稿]姓名+文章标题

  3. 以附件形式发送，正文需注明姓名、手机号、微信号，以便小编及时联系

分类： DBLE 分布式中间件技术文章版本发布

相关文章

OceanBase

OB 运维 | Oracle 迁移到 OB 过程中有哪些不兼容的 DDL 语法？

本文将总结 Oracle 迁移到 OB 4.x 过程中，那些还不兼容的 阅读更多…

一问一实验（ChatDBA）

第 58 期：MySQL Slave 异常掉电后主从同步出现异常怎么排查？

社区王牌专栏《一问一实验：AI 版》改版以来已发布多期（51-57）， 阅读更多…

SQLE SQL审核工具

SQLE 3.2407.0 发布，全面提升多项目场景体验！

本周我们发布了 SQLE 3.2407.0 正式版！主要更新概览： 社 阅读更多…