最近在写的数据迁移工具完成的差不多了,今天将连接池换成C3P0,发现一个问题,就是配置了多个数据源的C3P0在同时获取不同数据源的Connection时会发生死锁。 1.运行如下的代码,用JProfiler测试,会发现死锁的情况: 代码: package com.highgo.test.c3p0dea
最近在写的数据迁移工具完成的差不多了,今天将连接池换成C3P0,发现一个问题,就是配置了多个数据源的C3P0在同时获取不同数据源的Connection时会发生死锁。
1.运行如下的代码,用JProfiler测试,会发现死锁的情况:
代码:
package com.highgo.test.c3p0deadlock; import java.sql.SQLException; import com.mchange.v2.c3p0.ComboPooledDataSource; //加锁source个postgre的ComboPooledDataSource的getConnection用一个锁 public class Test { public static void main(String[] args) throws InterruptedException { ComboPooledDataSource source = new ComboPooledDataSource("source"); ComboPooledDataSource source2 = new ComboPooledDataSource("source"); ComboPooledDataSource postgres = new ComboPooledDataSource("postgres"); ComboPooledDataSource postgres2 = new ComboPooledDataSource("postgres"); new Thread(new SourceGetConn(source), "source").start(); // new Thread(new SourceGetConn(source2), "source2").start(); // Thread.sleep(1000); new Thread(new DestGetConn(postgres), "postgres").start(); // new Thread(new DestGetConn(postgres2), "postgres2").start(); } } class SourceGetConn implements Runnable { private ComboPooledDataSource source = null; public SourceGetConn(ComboPooledDataSource source) { this.source = source; } @Override public void run() { while (true) { try { Thread.sleep(1000); source.getConnection(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } } class DestGetConn implements Runnable { private ComboPooledDataSource postgres = null; public DestGetConn(ComboPooledDataSource source) { this.postgres = source; } @Override public void run() { while (true) { try { Thread.sleep(1000); postgres.getConnection(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } }
可以看到source和postgre两个进程都被一个没有记录的对象锁住了。
2.将上边的代码的Thread.sleep注释去掉,在运行,是不会有死锁问题的,于是查看C3P0的源代码,ComboPooledDataSource@getConnection是继承自AbstractPoolBackedDataSource#getConnection,代码如下:
public Connection getConnection() throws SQLException { PooledConnection pc = getPoolManager().getPool().checkoutPooledConnection(); return pc.getConnection(); } public Connection getConnection(String username, String password) throws SQLException { PooledConnection pc = getPoolManager().getPool(username, password).checkoutPooledConnection(); return pc.getConnection(); }
先看这个PoolManager,AbstractPoolBackedDataSource#getPoolManager方法的实现如下,是线程安全的
private synchronized C3P0PooledConnectionPoolManager getPoolManager() throws SQLException { if (poolManager == null) { ConnectionPoolDataSource cpds = assertCpds(); poolManager = new C3P0PooledConnectionPoolManager(cpds, null, null, this.getNumHelperThreads(), this.getIdentityToken(), this.getDataSourceName()); if (logger.isLoggable(MLevel.INFO)) logger.info("Initializing c3p0 pool... " + this.toString( true ) /* + "; using pool manager: " + poolManager */); } return poolManager; }从上边的代码也可以看出,一个DataSource实例,只保持一个PoolManager的引用。
public synchronized C3P0PooledConnectionPool getPool(String username, String password, boolean create) throws SQLException { if (create) return getPool( username, password ); else { DbAuth checkAuth = new DbAuth( username, password ); C3P0PooledConnectionPool out = (C3P0PooledConnectionPool) authsToPools.get(checkAuth); if (out == null) throw new SQLException("No pool has been initialized for databse user '" + username + "' with the specified password."); else return out; } }再看C3P0PooledConnectionPool#checkoutPooledConnection();
public PooledConnection checkoutPooledConnection() throws SQLException { //System.err.println(this + " -- CHECKOUT"); try { PooledConnection pc = (PooledConnection) this.checkoutAndMarkConnectionInUse(); pc.addConnectionEventListener( cl ); return pc; } catch (TimeoutException e) { throw SqlUtils.toSQLException("An attempt by a client to checkout a Connection has timed out.", e); } catch (CannotAcquireResourceException e) { throw SqlUtils.toSQLException("Connections could not be acquired from the underlying database!", "08001", e); } catch (Exception e) { throw SqlUtils.toSQLException(e); } }返回一个C3P0PooledConnection 实例;C3P0PooledConnection 这个类里的方法都是线程安全的。ComboPooledDataSource@getConnection的最后一站就是C3P0PooledConnection#getConnection;如下:
public synchronized Connection getConnection() throws SQLException { if ( exposedProxy != null) { //DEBUG //System.err.println("[DOUBLE_GET_TESTER] -- double getting a Connection from " + this ); //new Exception("[DOUBLE_GET_TESTER] -- Double-Get Stack Trace").printStackTrace(); //origGet.printStackTrace(); // System.err.println("c3p0 -- Uh oh... getConnection() was called on a PooledConnection when " + // "it had already provided a client with a Connection that has not yet been " + // "closed. This probably indicates a bug in the connection pool!!!"); logger.warning("c3p0 -- Uh oh... getConnection() was called on a PooledConnection when " + "it had already provided a client with a Connection that has not yet been " + "closed. This probably indicates a bug in the connection pool!!!"); return exposedProxy; } else { return getCreateNewConnection(); } }从上边的源码分析可以看出,一个ComboPooledDataSource实例的ComboPooledDataSource@getConnection是线程安全的,可以放心调用;可以测试一下,将最开始的代码稍微修改下,如下:
package com.highgo.test.c3p0deadlock; import java.sql.SQLException; import com.mchange.v2.c3p0.ComboPooledDataSource; //加锁source个postgre的ComboPooledDataSource的getConnection用一个锁 public class Test { public static void main(String[] args) throws InterruptedException { ComboPooledDataSource source = new ComboPooledDataSource("source"); // ComboPooledDataSource source2 = new ComboPooledDataSource("source"); ComboPooledDataSource postgres = new ComboPooledDataSource("postgres"); // ComboPooledDataSource postgres2 = new ComboPooledDataSource("postgres"); new Thread(new SourceGetConn(source), "source").start(); new Thread(new SourceGetConn(source), "source2").start(); // Thread.sleep(1000); // new Thread(new DestGetConn(postgres), "postgres").start(); // new Thread(new DestGetConn(postgres2), "postgres2").start(); } } class SourceGetConn implements Runnable { private ComboPooledDataSource source = null; public SourceGetConn(ComboPooledDataSource source) { this.source = source; } @Override public void run() { while (true) { try { Thread.sleep(1000); source.getConnection(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } } class DestGetConn implements Runnable { private ComboPooledDataSource postgres = null; public DestGetConn(ComboPooledDataSource source) { this.postgres = source; } @Override public void run() { while (true) { try { Thread.sleep(1000); postgres.getConnection(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } }将一个ComboPooledDataSource实例,传给两个线程分别getConnection,getConnection的过程没有加锁的情况下是可以运行的,完全没有问题。
稍微总结一下:
C3P0在一个ComboPooledDataSource实例的getConnection方法是线程安全的
C3P0在一个数据源的多个ComboPooledDataSource实例的getConnection方法也是线程安全的
C3P0在多个数据源的多个ComboPooledDataSource不同时调用getConnection的情况下,不会发生死锁(基于概率,若干时间之后,肯定会发生死锁)
C3P0在多个数据源的多个ComboPooledDataSource实例的getConnection方法同时(相邻的两行代码)调用时,会发生死锁现象,如1中所述
4.总结:
属于不同数据源的多个ComboPooledDataSource实例的getConnection方法调用要互斥
测试代码如下:
package com.highgo.test.c3p0deadlock; import java.sql.SQLException; import java.util.concurrent.locks.ReentrantLock; import com.mchange.v2.c3p0.ComboPooledDataSource; //加锁source个postgre的ComboPooledDataSource的getConnection用一个锁 public class Test2 { public static void main(String[] args) throws InterruptedException { ComboPooledDataSource source = new ComboPooledDataSource("source"); ComboPooledDataSource source2 = new ComboPooledDataSource("source"); ComboPooledDataSource postgres = new ComboPooledDataSource("postgres"); ComboPooledDataSource postgres2 = new ComboPooledDataSource("postgres"); ReentrantLock lock = new ReentrantLock(); new Thread(new SourceGetConn2(source, lock), "source").start(); new Thread(new SourceGetConn2(source2, lock), "source2").start(); Thread.sleep(1000); new Thread(new DestGetConn2(postgres, lock), "postgres").start(); new Thread(new DestGetConn2(postgres2, lock), "postgres2").start(); } } class SourceGetConn2 implements Runnable { private ComboPooledDataSource source = null; private ReentrantLock lock; public SourceGetConn2(ComboPooledDataSource source, ReentrantLock lock) { this.source = source; this.lock = lock; } @Override public void run() { while (true) { try { Thread.sleep(1000); lock.lock(); source.getConnection(); lock.unlock(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } } class DestGetConn2 implements Runnable { private ComboPooledDataSource postgres = null; private ReentrantLock lock; public DestGetConn2(ComboPooledDataSource source, ReentrantLock lock) { this.postgres = source; this.lock = lock; } @Override public void run() { while (true) { try { Thread.sleep(1000); lock.lock(); postgres.getConnection(); lock.unlock(); System.out.println("I get a Connection! I am in " + Thread.currentThread().getName()); } catch (InterruptedException | SQLException e) { e.printStackTrace(); } } } }
5.最后总结一个效率还可以的工具类
package com.highgo.hgdbadmin.myutil; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import com.mchange.v2.c3p0.ComboPooledDataSource; public class C3P0Util { public static String SOURCE = "source"; public static String POSTGRES = "postgres"; private ComboPooledDataSource source = null; private ComboPooledDataSource postgres = null; private static C3P0Util instance = null; private C3P0Util() { source = new ComboPooledDataSource("source"); postgres = new ComboPooledDataSource("postgres"); } public static final synchronized C3P0Util getInstance() { if (instance == null) { instance = new C3P0Util(); } return instance; } public synchronized Connection getConnection(String dataSource) throws SQLException { if ("source".equals(dataSource)) { return source.getConnection(); } else if ("postgres".equals(dataSource)) { return postgres.getConnection(); } return null; } public synchronized void close(Connection conn) { try { if (conn != null) { conn.close(); conn = null; } } catch (SQLException e) { } } public synchronized void close(Statement stat) { try { if (stat != null) { stat.close(); stat = null; } } catch (SQLException e) { } } public synchronized void close(ResultSet rest) { try { if (rest != null) { rest.close(); rest = null; } } catch (SQLException e) { } } public static void main(String[] args) { new Thread(new TestThread(), "test").start(); } private static class TestThread implements Runnable { private String dataSource = "source"; @Override public void run() { while (true) { try { Connection conn = C3P0Util.getInstance().getConnection(""); System.out.println("hello,this is " + dataSource); } catch (SQLException e) { // TODO Auto-generated catch block e.printStackTrace(); } if ("source".equals(dataSource)) { dataSource = "postgres"; } else { dataSource = "source"; } } } } }