打开一个会话Session
前文分析了MyBatis将配置文件转换为Java对象的流程,本文开始分析一下insert方法、update方法、delete方法处理的流程,至于为什么这三个方法要放在一起说,是因为:
- 从语义的角度,insert、update、delete都是属于对数据库的行进行更新操作
- 从实现的角度,我们熟悉的PreparedStatement里面提供了两种execute方法,一种是executeUpdate(),一种是executeQuery(),前者对应的是insert、update与delete,后者对应的是select,因此对于MyBatis来说只有update与select
示例代码为这段:
1 public long insertMail(Mail mail) { 2 SqlSession ss = ssf.openSession(); 3 try { 4 int rows = ss.insert(NAME_SPACE + "insertMail", mail); 5 ss.commit(); 6 if (rows > 0) { 7 return mail.getId(); 8 } 9 return 0; 10 } catch (Exception e) { 11 ss.rollback(); 12 return 0; 13 } finally { 14 ss.close(); 15 } 16 }
首先关注的是第2行的代码,ssf是SqlSessionFactory,其类型是DefaultSqlSessionFactory,上文最后已经分析过了,这里通过DefaultSqlSessionFactory来打开一个Session,通过Session去进行CRUD操作。
看一下openSession()方法的实现:
1 public SqlSession openSession() { 2 return openSessionFromDataSource(configuration.getDefaultExecutorType(), null, false); 3 }
顾名思义,从DataSource中获取Session,第一个参数的值是ExecutorType.SIMPLE,继续跟代码:
1 private SqlSession openSessionFromDataSource(ExecutorType execType, TransactionIsolationLevel level, boolean autoCommit) { 2 Transaction tx = null; 3 try { 4 final Environment environment = configuration.getEnvironment(); 5 final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment); 6 tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit); 7 final Executor executor = configuration.newExecutor(tx, execType); 8 return new DefaultSqlSession(configuration, executor, autoCommit); 9 } catch (Exception e) { 10 closeTransaction(tx); // may have fetched a connection so lets call close() 11 throw ExceptionFactory.wrapException("Error opening session. Cause: " + e, e); 12 } finally { 13 ErrorContext.instance().reset(); 14 } 15 }
第4行的代码,获取配置的环境信息Environment。
第5行的代码,从Environment中获取事物工厂TransactionFactory,由于<environment>中配置的是"JDBC",因此其真实类型是JdbcTransactionFactory,上文有说过。
第6行的代码,根据Environment中的DataSource(其实际类型是PooledDataSource)、TransactionIsolationLevel、autoCommit三个参数从TransactionFactory中获取一个事物,注意第三个参数autoCommit,它是openSession()方法中传过来的,其值为false,即MyBatis默认事物是不自动提交的。
第7行的代码,代码跟一下:
1 public Executor newExecutor(Transaction transaction, ExecutorType executorType) { 2 executorType = executorType == null ? defaultExecutorType : executorType; 3 executorType = executorType == null ? ExecutorType.SIMPLE : executorType; 4 Executor executor; 5 if (ExecutorType.BATCH == executorType) { 6 executor = new BatchExecutor(this, transaction); 7 } else if (ExecutorType.REUSE == executorType) { 8 executor = new ReuseExecutor(this, transaction); 9 } else { 10 executor = new SimpleExecutor(this, transaction); 11 } 12 if (cacheEnabled) { 13 executor = new CachingExecutor(executor); 14 } 15 executor = (Executor) interceptorChain.pluginAll(executor); 16 return executor; 17 }
这里总结一下:
- 根据ExecutorType获取一个执行器,这里是第10行的SimpleExecutor
- 如果满足第12行的判断开启缓存功能,则执行第13行的代码。第13行的代码使用到了装饰器模式,传入Executor,给SimpleExecutor装饰上了缓存功能
- 第15行的代码用于设置插件
这样就获取了一个Executor。最后将Executor、Configuration、autoCommit三个变量作为参数,实例化一个SqlSession出来,其实际类型为DefaultSqlSession。
insert方法执行流程
在看了openSession()方法知道最终获得了一个DefaultSqlSession之后,看一下DefaultSqlSession的insert方法是如何实现的:
1 public int insert(String statement, Object parameter) { 2 return update(statement, parameter); 3 }
看到虽然调用的是insert方法,但是最终统一都会去执行update方法,delete方法也是如此,这个开头已经说过了,这里证明了这一点。
接着继续看第2行的方法实现:
1 public int update(String statement, Object parameter) { 2 try { 3 dirty = true; 4 MappedStatement ms = configuration.getMappedStatement(statement); 5 return executor.update(ms, wrapCollection(parameter)); 6 } catch (Exception e) { 7 throw ExceptionFactory.wrapException("Error updating database. Cause: " + e, e); 8 } finally { 9 ErrorContext.instance().reset(); 10 } 11 }
第4行的代码根据statement从Configuration中获取MappedStatement,MappedStatement上文已经分析过了,存储在Configuration的mappedStatements字段中。
第5行的代码分为两部分,首先wrapCollection,顾名思义包装集合类,源码为:
1 private Object wrapCollection(final Object object) { 2 if (object instanceof Collection) { 3 StrictMap<Object> map = new StrictMap<Object>(); 4 map.put("collection", object); 5 if (object instanceof List) { 6 map.put("list", object); 7 } 8 return map; 9 } else if (object != null && object.getClass().isArray()) { 10 StrictMap<Object> map = new StrictMap<Object>(); 11 map.put("array", object); 12 return map; 13 } 14 return object; 15 }
这里做了三层处理:
- 如果参数是Collection(即集合)类型,放一个key为"collection"、value为参数的键值对
- 如果参数是List类型,放一个key为"list"、value为参数的键值对
- 如果参数是数组类型,放一个key为"array"、value为参数的键值对
将集合进行包装之后,就可以执行Executor的update方法了,Executor上面说了,是使用装饰器模式将SimpleExecutor加上了缓存功能的CacheExecutor,它的update方法实现为:
1 public int update(MappedStatement ms, Object parameterObject) throws SQLException { 2 flushCacheIfRequired(ms); 3 return delegate.update(ms, parameterObject); 4 }
第2行的代码是判断是否要求清缓存的,这里首先我们的示例配置文件mail.xml中没有配置<cache>,其次<insert>、<delete>、<update>、<select>中没有配置flushCache="true"属性,因此这一句代码不会执行任何操作。
第3行的代码delegate就是SimpleExecutor本身,因为是装饰器模式,因此会持有接口的引用,deletegate其类型就是Executor。继续跟代码,看一下update方法:
1 public int update(MappedStatement ms, Object parameter) throws SQLException { 2 ErrorContext.instance().resource(ms.getResource()).activity("executing an update").object(ms.getId()); 3 if (closed) { 4 throw new ExecutorException("Executor was closed."); 5 } 6 clearLocalCache(); 7 return doUpdate(ms, parameter); 8 }
前面的没什么好看的,继续跟第7行的代码:
1 public int doUpdate(MappedStatement ms, Object parameter) throws SQLException { 2 Statement stmt = null; 3 try { 4 Configuration configuration = ms.getConfiguration(); 5 StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null); 6 stmt = prepareStatement(handler, ms.getStatementLog()); 7 return handler.update(stmt); 8 } finally { 9 closeStatement(stmt); 10 } 11 }
第4行的代码获取MappedStatement中的Configuration对象。
第5行的代码获取Statement处理器StatementHandler接口实现类,Statement是Java原生的为JDBC设计的声明,StatementHandler接口实现类的真实类型为RoutingStatementHandler。
第6行和第7行的代码后文逐步分析,因为里面一点一点封装了我们平时写JDBC时的一些基本步骤,比如获取Connection,构建PreparedStatement、对execute后的结果进行处理等,先看一下prepareStatement的源码:
1 private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException { 2 Statement stmt; 3 Connection connection = getConnection(statementLog); 4 stmt = handler.prepare(connection, transaction.getTimeout()); 5 handler.parameterize(stmt); 6 return stmt; 7 }
后面逐步分析。
获取Connection
第一步,看下获取Connection的步骤。看一下上面getConnection方法如何实现:
1 protected Connection getConnection(Log statementLog) throws SQLException { 2 Connection connection = transaction.getConnection(); 3 if (statementLog.isDebugEnabled()) { 4 return ConnectionLogger.newInstance(connection, statementLog, queryStack); 5 } else { 6 return connection; 7 } 8 }
Connection从Transaction中获取,配置的是JDBC,这里代码进入JdbcTransaction的getConnection():
1 protected Connection getConnection(Log statementLog) throws SQLException { 2 Connection connection = transaction.getConnection(); 3 if (statementLog.isDebugEnabled()) { 4 return ConnectionLogger.newInstance(connection, statementLog, queryStack); 5 } else { 6 return connection; 7 } 8 }
先看一下第3行~第7行的代码,判断的意思是是否开启Statement的表达式,如果开启,那么第4行会给生成的Connection加上一个代理,代理的内容是在调用prepareStatement、prepareCall等方法前或者方法后打印日志,具体可见ConnectionLogger、PreparedStatementLogger、ResultSetLogger与StatementLogger的invoke方法。
接着继续跟第2行的代码:
1 public Connection getConnection() throws SQLException { 2 if (connection == null) { 3 openConnection(); 4 } 5 return connection; 6 }
跟一下第3行的代码:
1 protected void openConnection() throws SQLException { 2 if (log.isDebugEnabled()) { 3 log.debug("Opening JDBC Connection"); 4 } 5 connection = dataSource.getConnection(); 6 if (level != null) { 7 connection.setTransactionIsolation(level.getLevel()); 8 } 9 setDesiredAutoCommit(autoCommmit); 10 }
第6行~第8行的代码用于设置事物隔离级别,第9行的代码用于设置是否自动提交事物。下面跟一下第5行的代码getConnection()方法:
1 public Connection getConnection() throws SQLException { 2 return popConnection(dataSource.getUsername(), dataSource.getPassword()).getProxyConnection(); 3 }
这里简单提一下,在方法名中如果看到了"pop"、"push"字样,一定要把该方法使用的数据结构和栈联想起来,栈(stack)是一个先进先出数据结构,"pop"、"push"是栈特有的操作,前者将栈顶的数据推送出栈让调用者获取到,后者将数据压入栈顶。
后面的getProxyConnection()方法就是将获取到的Connection返回而已,没什么特殊的操作,这里跟一下popConnection方法实现,它位于PooledDataSource类中,这是由<dataSource>标签中的type属性决定的:
1 private PooledConnection popConnection(String username, String password) throws SQLException { 2 boolean countedWait = false; 3 PooledConnection conn = null; 4 long t = System.currentTimeMillis(); 5 int localBadConnectionCount = 0; 6 7 while (conn == null) { 8 synchronized (state) { 9 if (!state.idleConnections.isEmpty()) { 10 // Pool has available connection 11 conn = state.idleConnections.remove(0); 12 if (log.isDebugEnabled()) { 13 log.debug("Checked out connection " + conn.getRealHashCode() + " from pool."); 14 } 15 } else { 16 // Pool does not have available connection 17 if (state.activeConnections.size() < poolMaximumActiveConnections) { 18 // Can create new connection 19 conn = new PooledConnection(dataSource.getConnection(), this); 20 if (log.isDebugEnabled()) { 21 log.debug("Created connection " + conn.getRealHashCode() + "."); 22 } 23 } else { 24 // Cannot create new connection 25 PooledConnection oldestActiveConnection = state.activeConnections.get(0); 26 long longestCheckoutTime = oldestActiveConnection.getCheckoutTime(); 27 if (longestCheckoutTime > poolMaximumCheckoutTime) { 28 // Can claim overdue connection 29 state.claimedOverdueConnectionCount++; 30 state.accumulatedCheckoutTimeOfOverdueConnections += longestCheckoutTime; 31 state.accumulatedCheckoutTime += longestCheckoutTime; 32 state.activeConnections.remove(oldestActiveConnection); 33 if (!oldestActiveConnection.getRealConnection().getAutoCommit()) { 34 try { 35 oldestActiveConnection.getRealConnection().rollback(); 36 } catch (SQLException e) { 37 log.debug("Bad connection. Could not roll back"); 38 } 39 } 40 conn = new PooledConnection(oldestActiveConnection.getRealConnection(), this); 41 oldestActiveConnection.invalidate(); 42 if (log.isDebugEnabled()) { 43 log.debug("Claimed overdue connection " + conn.getRealHashCode() + "."); 44 } 45 } else { 46 // Must wait 47 try { 48 if (!countedWait) { 49 state.hadToWaitCount++; 50 countedWait = true; 51 } 52 if (log.isDebugEnabled()) { 53 log.debug("Waiting as long as " + poolTimeToWait + " milliseconds for connection."); 54 } 55 long wt = System.currentTimeMillis(); 56 state.wait(poolTimeToWait); 57 state.accumulatedWaitTime += System.currentTimeMillis() - wt; 58 } catch (InterruptedException e) { 59 break; 60 } 61 } 62 } 63 } 64 if (conn != null) { 65 if (conn.isValid()) { 66 if (!conn.getRealConnection().getAutoCommit()) { 67 conn.getRealConnection().rollback(); 68 } 69 conn.setConnectionTypeCode(assembleConnectionTypeCode(dataSource.getUrl(), username, password)); 70 conn.setCheckoutTimestamp(System.currentTimeMillis()); 71 conn.setLastUsedTimestamp(System.currentTimeMillis()); 72 state.activeConnections.add(conn); 73 state.requestCount++; 74 state.accumulatedRequestTime += System.currentTimeMillis() - t; 75 } else { 76 if (log.isDebugEnabled()) { 77 log.debug("A bad connection (" + conn.getRealHashCode() + ") was returned from the pool, getting another connection."); 78 } 79 state.badConnectionCount++; 80 localBadConnectionCount++; 81 conn = null; 82 if (localBadConnectionCount > (poolMaximumIdleConnections + 3)) { 83 if (log.isDebugEnabled()) { 84 log.debug("PooledDataSource: Could not get a good connection to the database."); 85 } 86 throw new SQLException("PooledDataSource: Could not get a good connection to the database."); 87 } 88 } 89 } 90 } 91 92 } 93 94 if (conn == null) { 95 if (log.isDebugEnabled()) { 96 log.debug("PooledDataSource: Unknown severe error condition. The connection pool returned a null connection."); 97 } 98 throw new SQLException("PooledDataSource: Unknown severe error condition. The connection pool returned a null connection."); 99 } 100 101 return conn; 102 }
这段方法很长,分解一下。
首先是第9行~第15行的判断,假使空闲的Connection列表不是空的,Connection就是空闲Connection列表的第一个Connection,且移除空闲Connection列表的第一个Connection,这也符合PooledDataSource的定义,有一个Connection池,对Connection进行复用而不是每次都new出来,这就是典型的栈的操作。但是这里有一点我认为MyBatis写得不是很好,List的实际类型是ArrayList,每次的移除操作是remove(0),ArrayList处理remove效率并不高尤其还是remove(0)的操作,因此这里替换成LinkedList会更好一些。
接着先看第23行~第63行的判断,它的判断逻辑是假如当前在使用的Connection数量大于或等于最大可用的Connection数量,那么获取当前正在使用的Connection列表中的第一个Connection做一个判断:
- 如果当前Connection执行时间已经超过了指定的Connection最大超时时间,那么原Connection如果不是自动Commit的,数据回滚,新建一个Connection,原Connection失效
- 如果当前Connection执行时间没有超过指定的Connection最大超时时间,那么使用wait方法等待
最后回到第17行~第23行的判断,即当前在使用的Connection数量小于最大可用的Connection数量,那么此时直接new一个PooledConnection出来,看一下PooledDataSource的getConnection()方法实现:
1 public Connection getConnection() throws SQLException { 2 return doGetConnection(username, password); 3 }
继续跟代码doGetConnection方法:
1 private Connection doGetConnection(String username, String password) throws SQLException { 2 Properties props = new Properties(); 3 if (driverProperties != null) { 4 props.putAll(driverProperties); 5 } 6 if (username != null) { 7 props.setProperty("user", username); 8 } 9 if (password != null) { 10 props.setProperty("password", password); 11 } 12 return doGetConnection(props); 13 }
这里就是先设置一下配置的属性,继续跟第12行的方法实现:
1 private Connection doGetConnection(Properties properties) throws SQLException { 2 initializeDriver(); 3 Connection connection = DriverManager.getConnection(url, properties); 4 configureConnection(connection); 5 return connection; 6 }
到了这里就是我们比较熟悉的代码了。
第2行的代码意思是MyBatis维护了一个Driver池registeredDrivers,如果我们的Driver不在Driver池里面,那么会尝试使用Class.forName方法初始化一下,成功的话加入Driver池中。
第3行的代码不说了,使用DriverManager的getConnection方法获取Connection,第4行的代码配置一下Connection,主要就是设置一下自动提交属性与事物隔离级别。
最后将生成的Connection返回出去,完成生成Connection的流程。
为Connection生成代理
上面解析了生成Connection的流程,代码到这里还没完还有一步,看一下PooledConnection的构造方法:
1 public PooledConnection(Connection connection, PooledDataSource dataSource) { 2 this.hashCode = connection.hashCode(); 3 this.realConnection = connection; 4 this.dataSource = dataSource; 5 this.createdTimestamp = System.currentTimeMillis(); 6 this.lastUsedTimestamp = System.currentTimeMillis(); 7 this.valid = true; 8 this.proxyConnection = (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), IFACES, this); 9 }
这里第8行的代码会为生成的Connection创建一个代理,PooledConnection本身就实现了InvocationHandler接口,看一下代理内容是什么,invoke方法的实现:
1 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 2 String methodName = method.getName(); 3 if (CLOSE.hashCode() == methodName.hashCode() && CLOSE.equals(methodName)) { 4 dataSource.pushConnection(this); 5 return null; 6 } else { 7 try { 8 if (!Object.class.equals(method.getDeclaringClass())) { 9 // issue #579 toString() should never fail 10 // throw an SQLException instead of a Runtime 11 checkConnection(); 12 } 13 return method.invoke(realConnection, args); 14 } catch (Throwable t) { 15 throw ExceptionUtil.unwrapThrowable(t); 16 } 17 } 18 }
这一步操作主要是为了处理close方法的,看一下pushConnection方法的实现:
1 protected void pushConnection(PooledConnection conn) throws SQLException { 2 3 synchronized (state) { 4 state.activeConnections.remove(conn); 5 if (conn.isValid()) { 6 if (state.idleConnections.size() < poolMaximumIdleConnections && conn.getConnectionTypeCode() == expectedConnectionTypeCode) { 7 state.accumulatedCheckoutTime += conn.getCheckoutTime(); 8 if (!conn.getRealConnection().getAutoCommit()) { 9 conn.getRealConnection().rollback(); 10 } 11 PooledConnection newConn = new PooledConnection(conn.getRealConnection(), this); 12 state.idleConnections.add(newConn); 13 newConn.setCreatedTimestamp(conn.getCreatedTimestamp()); 14 newConn.setLastUsedTimestamp(conn.getLastUsedTimestamp()); 15 conn.invalidate(); 16 if (log.isDebugEnabled()) { 17 log.debug("Returned connection " + newConn.getRealHashCode() + " to pool."); 18 } 19 state.notifyAll(); 20 } else { 21 state.accumulatedCheckoutTime += conn.getCheckoutTime(); 22 if (!conn.getRealConnection().getAutoCommit()) { 23 conn.getRealConnection().rollback(); 24 } 25 conn.getRealConnection().close(); 26 if (log.isDebugEnabled()) { 27 log.debug("Closed connection " + conn.getRealHashCode() + "."); 28 } 29 conn.invalidate(); 30 } 31 } else { 32 if (log.isDebugEnabled()) { 33 log.debug("A bad connection (" + conn.getRealHashCode() + ") attempted to return to the pool, discarding connection."); 34 } 35 state.badConnectionCount++; 36 } 37 } 38 }
代码的逻辑简单来说就是当调用close方法的时候,如果当前空闲Connection列表中的Connection数量小于指定空闲Connection列表中的数量(第二个判断connectionTypeCode的值为275950209,不知道是干什么的),那么会为原Connection生成一个PooledConnection并加入空闲Connection列表中。
如果不满足上面的条件,那么就直接调用Connection的close()方法并且让原Connection失效。