JavaGGDataSource

d

转自：来源忘记了。

 

/**
 * JavaGGDataSource.java 2011-3-4 上午09:21:05
 */
package test.datasource;

import java.io.PrintWriter;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import javax.sql.DataSource;

/**
 * @author linzq
 * 
 */
public class JavaGGDataSource implements DataSource
{

    // 连接队列

    private ConcurrentLinkedQueue<_Connection> connQueue    = new ConcurrentLinkedQueue<_Connection>();

    // 存放所有连接容器

    private List<_Connection>                  conns        = new ArrayList<_Connection>();
    private Driver                             driver       = null;
    private String                             jdbcUrl      = null;
    private String                             user         = null;
    private String                             password     = null;
    // -1为不限制连接数
    private int                                maxActive    = -1;
    private String                             driverClass  = null;
    // 默认为4小时,即4小时没有任何sql操作就把所有连接重新建立连接
    private int                                timeout      = 1000 * 60 * 60 * 4;
    private AtomicLong                         lastCheckout = new AtomicLong(
                                                                    System
                                                                            .currentTimeMillis());
    private AtomicInteger                      connCount    = new AtomicInteger();
    // 线程锁,主要用于新建连接和清空连接时
    private ReentrantLock                      lock         = new ReentrantLock();

    public void closeAllConnection()
    {

    }

    /**
     * 
     * 归还连接给连接池
     * 
     * 
     * 
     * @param conn
     * 
     *@date 2009-8-13
     * 
     *@author eric.chan
     */

    public void offerConnection(_Connection conn)
    {
        connQueue.offer(conn);
    }

    @Override
    public Connection getConnection() throws SQLException
    {
        return getConnection(user, password);

    }

    /**
     * 
     * 从池中得到连接，如果池中没有连接，则建立新的sql连接
     * 
     * 
     * 
     * @param username
     * 
     * @param password
     * 
     * @author eric.chan
     */

    @Override
    public Connection getConnection(String username, String password)

    throws SQLException
    {
        checkTimeout();
        _Connection conn = connQueue.poll();
        if (conn == null)
        {
            if (maxActive > 0 && connCount.get() >= maxActive)
            {
                for (;;)
                {
                    // 采用自旋方法 从已满的池中得到一个连接
                    conn = connQueue.poll();
                    if (conn != null)
                        break;
                    else
                        continue;
                }
            }
            lock.lock();
            try
            {
                if (maxActive > 0 && connCount.get() >= maxActive)
                {
                    // 处理并发问题
                    return getConnection(username, password);
                }
                Properties info = new Properties();
                info.put("user", username);
                info.put("password", password);
                Connection conn1 = loadDriver().connect(jdbcUrl, info);
                conn = new _Connection(conn1, this);
                int c = connCount.incrementAndGet();// 当前连接数加1
                conns.add(conn);
                System.out.println("info : init no. " + c + " connectioned");
            } finally
            {
                lock.unlock();
            }
        }
        lastCheckout.getAndSet(System.currentTimeMillis());
        return conn.getConnection();
    }

    /**
     * 
     * 检查最后一次的连接时间
     * 
     * 
     * 
     * @throws SQLException
     * 
     *@date 2009-8-13
     * 
     *@author eric.chan
     */

    private void checkTimeout() throws SQLException
    {

        long now = System.currentTimeMillis();
        long lt = lastCheckout.get();
        if ((now - lt) > timeout)
        {
            _Connection conn = null;
            lock.lock();
            try
            {
                if (connCount.get() == 0)
                    return;
                while ((conn = connQueue.poll()) != null)
                {
                    System.out.println("connection " + conn + " close ");
                    conn.close();
                    conn = null;
                }
                for (_Connection con : conns)
                {
                    con.close();
                }
                conns.clear();
                System.out.println("info : reset all connections");
                // 重置连接数计数器
                connCount.getAndSet(0);
                lastCheckout.getAndSet(System.currentTimeMillis());
            } finally
            {
                lock.unlock();
            }
        }
    }

    /**
     * 
     * 
     * 
     * @return
     * 
     *@date 2009-8-13
     * 
     *@author eric.chan
     */

    private Driver loadDriver()
    {
        if (driver == null)
        {
            try
            {
                driver = (Driver) Class.forName(driverClass).newInstance();
            } catch (ClassNotFoundException e)
            {
                System.out.println("error : can not find driver class " +
                        driverClass);
            } catch (Exception e)
            {
                e.printStackTrace();
            }
        }
        return driver;
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException
    {
        return null;
    }

    @Override
    public int getLoginTimeout() throws SQLException
    {
        return 0;
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException
    {
    }

    @Override
    public void setLoginTimeout(int seconds) throws SQLException
    {
    }

    @Override
    public boolean isWrapperFor(Class iface) throws SQLException
    {
        throw new SQLException("no Implemented isWrapperFor method");
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException
    {
        throw new SQLException("no Implemented unwrap method");
    }

    public String getJdbcUrl()
    {
        return jdbcUrl;
    }

    public void setJdbcUrl(String jdbcUrl)
    {
        this.jdbcUrl = jdbcUrl;
    }

    public String getUsername()
    {
        return user;
    }

    public void setUsername(String user)
    {
        this.user = user;
    }

    public String getPassword()
    {
        return password;
    }

    public void setPassword(String password)
    {
        this.password = password;
    }

    public String getDriverClass()
    {
        return driverClass;
    }

    public void setDriverClass(String driverClass)
    {
        this.driverClass = driverClass;
    }

    public int getTimeout()
    {
        return timeout;
    }

    public void setTimeout(int timeout)
    {
        this.timeout = timeout * 1000;
    }

    public void setMaxActive(int maxActive)
    {
        this.maxActive = maxActive;
    }

    public int getMaxActive()
    {
        return maxActive;
    }

}

/**
 * 数据连接的自封装 ,是java.sql.Connection的一个钩子,主要是处理close方法
 * 
 * @author linzq
 * 
 */
class _Connection implements InvocationHandler
{

    private final static String    CLOSE_METHOD_NAME = "close";

    private final Connection       conn;

    private final JavaGGDataSource ds;

    _Connection(Connection conn, JavaGGDataSource ds)
    {
        this.conn = conn;
        this.ds = ds;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args)
            throws Throwable
    {
        Object obj = null;
        // 判断是否调用了close的方法，如果调用close方法则把连接置为无用状态
        if (CLOSE_METHOD_NAME.equals(method.getName()))
        {
            // 归还连接给连接池
            ds.offerConnection(this);
        } else
        {
            // 运行非close的方法
            obj = method.invoke(conn, args);
        }
        return obj;
    }

    public Connection getConnection()
    {
        // 返回数据库连接conn的接管类，以便截住close方法
        Connection conn2 = (Connection) Proxy.newProxyInstance(conn.getClass()
                .getClassLoader(), new Class[] { Connection.class }, this);
        return conn2;
    }

    public void close() throws SQLException
    {
        // 调用真正的close方法,一但调用此方法就直接关闭连接
        if (conn != null && !conn.isClosed())
            conn.close();
    }
}

 

/**
 * TestGG.java 2011-3-4 上午10:02:14
 */
package test.datasource;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @author linzq
 * 
 */
public class TestGG
{

    /**
     * @param args
     */
    public static void main(String[] args)
    {
        JavaGGDataSource ds = new JavaGGDataSource();
        ds.setDriverClass("com.mysql.jdbc.Driver");
        ds.setJdbcUrl("jdbc:mysql://localhost:3306/test");
        ds.setUsername("root");
        ds.setPassword("123456");
        ds.setTimeout(300);
        // ds.setMaxActive(60);
        for (int i = 0; i < 20; i++)
        {
            new GG(ds).start();
        }
    }

    static class GG extends Thread
    {
        JavaGGDataSource ds = null;
        long             l  = System.currentTimeMillis();

        public GG(JavaGGDataSource ds)
        {
            this.ds = ds;
        }

        static final String sql       = "insert into testgg(col1,cols) values (?,?)";
        static final String selectsql = "select * from testgg where id=?";

        public void run()
        {
            for (int t = 0; t < 10000; t++)
            {
                Connection conn = null;
                try
                {
                    conn = ds.getConnection();
                    PreparedStatement ps = conn.prepareStatement(sql);
                    // 以下为insert
                    ps.setInt(1, 133664);
                    ps.setString(2, "ddd");
                    ps.executeUpdate();
                    ResultSet rs = ps.getGeneratedKeys();
                    // 以下为select
                    // 取得自增长ID
                    ps = conn.prepareStatement(selectsql);
                    if (rs.next())
                    {
                        // System.out.println(rs);
                        ps.setInt(1, rs.getInt("GENERATED_KEY"));// 表的字段名字也可以用字段下标值
                    }
                    rs = ps.executeQuery();
                    while (rs.next())
                    {
                        rs.getInt("id");
                        rs.getInt("col1");
                    }
                    rs.close();
                    ps.close();
                } catch (SQLException e)
                {
                    e.printStackTrace();
                } finally
                {
                    try
                    {
                        if (conn != null)
                        {
                            // ds.offerConnection(conn);
                            conn.close();
                        }
                    } catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println(System.currentTimeMillis() - l);
        }
    }
}

  数据库表结构：

CREATE TABLE `testgg` (
  `id` int(11) NOT NULL auto_increment,
  `col1` int(11) default NULL,
  `cols` varchar(200) default NULL,
  PRIMARY KEY  (`id`)
)

 

d