HDDataSource

2023-03-04 19:17:16 +00:00 · 2011-12-03 06:23:00 +08:00
parent d7ee8d45be
commit a910816eff
12 changed files with 284 additions and 22 deletions
@@ -190,6 +190,14 @@
 			<version>1.6.1</version>
 			<scope>provided</scope>
 		</dependency>
+		
+		<!-- for HADataSource  -->
+		<dependency>
+			<groupId>com.alibaba</groupId>
+			<artifactId>fastjson</artifactId>
+			<version>1.1.11</version>
+			<scope>provided</scope>
+		</dependency>

 		<dependency>
 			<groupId>mysql</groupId>
@@ -242,11 +250,6 @@
 			<version>0.9.1.2</version>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>com.alibaba</groupId>
-			<artifactId>fastjson</artifactId>
-			<version>1.0.6</version>
-			<scope>test</scope>
-		</dependency>
+	
 	</dependencies>
 </project>
@@ -329,9 +329,13 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            lock.unlock();
        }
    }
-
+    
    @Override
    public Connection getConnection() throws SQLException {
+        return getConnection(maxWait);
+    }
+
+    public Connection getConnection(long maxWaitMillis) throws SQLException {
        init();

        final int maxWaitThreadCount = getMaxWaitThreadCount();
@@ -349,7 +353,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        }

        for (;;) {
-            PoolableConnection poolalbeConnection = getConnectionInternal();
+            PoolableConnection poolalbeConnection = getConnectionInternal(maxWaitMillis);

            if (isTestOnBorrow()) {
                boolean validate = testConnectionInternal(poolalbeConnection.getConnection());
@@ -423,7 +427,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        }
    }

-    private PoolableConnection getConnectionInternal() throws SQLException {
+    private PoolableConnection getConnectionInternal(long maxWait) throws SQLException {
        PoolableConnection poolalbeConnection;

        try {
@@ -54,11 +54,6 @@ public class HADataSource extends MultiDataSource implements DataSource {
        throw new UnsupportedOperationException();
    }

-    @Override
-    public Connection getConnection() throws SQLException {
-        return new MultiDataSourceConnection(this, createConnectionId());
-    }
-
    public Balancer getBalancer() {
        return balancer;
    }
@@ -4,22 +4,74 @@ import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;

 import com.alibaba.druid.pool.DataSourceAdapter;
 import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.pool.ha.valid.DefaultValidDataSourceChecker;
+import com.alibaba.druid.pool.ha.valid.ValidDataSourceChecker;

 public abstract class MultiDataSource extends DataSourceAdapter {

-    private Properties          properties       = new Properties();
+    private Properties               properties                       = new Properties();

-    private final AtomicInteger connectionIdSeed = new AtomicInteger();
-    private final AtomicInteger statementIdSeed  = new AtomicInteger();
+    private final AtomicInteger      connectionIdSeed                 = new AtomicInteger();
+    private final AtomicInteger      statementIdSeed                  = new AtomicInteger();
+
+    private ValidDataSourceChecker   validDataSourceChecker           = new DefaultValidDataSourceChecker();
+    private long                     validDataSourceCheckPeriodMillis = 3000;
+
+    private int                      schedulerThreadCount             = 3;
+    private ScheduledExecutorService scheduler;
+
+    private boolean                  inited                           = false;
+    private final Lock               lock                             = new ReentrantLock();

    public MultiDataSource(){

    }

+    public void init() {
+        if (inited) {
+            return;
+        }
+
+        lock.lock();
+        try {
+            if (inited) {
+                return;
+            }
+
+            scheduler = Executors.newScheduledThreadPool(schedulerThreadCount);
+            scheduler.scheduleAtFixedRate(new ValidTask(), validDataSourceCheckPeriodMillis,
+                                          validDataSourceCheckPeriodMillis, TimeUnit.MILLISECONDS);
+            inited = true;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public long getValidDataSourceCheckPeriodMillis() {
+        return validDataSourceCheckPeriodMillis;
+    }
+
+    public void setValidDataSourceCheckPeriodMillis(long validDataSourceCheckPeriodMillis) {
+        this.validDataSourceCheckPeriodMillis = validDataSourceCheckPeriodMillis;
+    }
+
+    public ValidDataSourceChecker getValidDataSourceChecker() {
+        return validDataSourceChecker;
+    }
+
+    public void setValidDataSourceChecker(ValidDataSourceChecker validDataSourceChecker) {
+        this.validDataSourceChecker = validDataSourceChecker;
+    }
+
    public int createConnectionId() {
        return connectionIdSeed.getAndIncrement();
    }
@@ -30,16 +82,31 @@ public abstract class MultiDataSource extends DataSourceAdapter {

    public abstract List<DruidDataSource> getDataSources();

-    public abstract void setDataSources(List<DruidDataSource> dataSources);
-
    public Properties getProperties() {
        return properties;
    }

-    @Override
-    public abstract Connection getConnection() throws SQLException;
+    public Connection getConnection() throws SQLException {
+        init();
+
+        return new MultiDataSourceConnection(this, createConnectionId());
+    }

    public abstract Connection getConnectionInternal(MultiDataSourceConnection conn, String sql) throws SQLException;

    public abstract void handleNotAwailableDatasource(DruidDataSource dataSource);
+
+    class ValidTask implements Runnable {
+
+        @Override
+        public void run() {
+            for (DruidDataSource dataSource : getDataSources()) {
+                boolean isValid = validDataSourceChecker.isValid(dataSource);
+                if (isValid) {
+                    handleNotAwailableDatasource(dataSource);
+                }
+            }
+        }
+
+    }
 }
@@ -49,7 +49,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect

    private Date                  connectedTime    = null;

-    public MultiDataSourceConnection(HADataSource haDataSource, int id){
+    public MultiDataSourceConnection(MultiDataSource haDataSource, int id){
        this.haDataSource = haDataSource;
        this.id = id;
    }
@@ -0,0 +1,40 @@
+package com.alibaba.druid.pool.ha.cobar;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.pool.ha.MultiDataSource;
+import com.alibaba.druid.pool.ha.MultiDataSourceConnection;
+
+
+public class CobarDataSource extends MultiDataSource {
+    private List<DruidDataSource> dataSources = new ArrayList<DruidDataSource>();
+
+    @Override
+    public List<DruidDataSource> getDataSources() {
+        return dataSources;
+    }
+
+    public void setDataSources(List<DruidDataSource> dataSources) {
+        this.dataSources = dataSources;
+    }
+
+    @Override
+    public Connection getConnection() throws SQLException {
+        return new MultiDataSourceConnection(this, createConnectionId());
+    }
+
+    @Override
+    public Connection getConnectionInternal(MultiDataSourceConnection conn, String sql) throws SQLException {
+        return null;
+    }
+
+    @Override
+    public void handleNotAwailableDatasource(DruidDataSource dataSource) {
+        
+    }
+
+}
@@ -0,0 +1,5 @@
+package com.alibaba.druid.pool.ha.config;
+
+public interface ConfigLoader {
+    MultiDataSourceConfig load();
+}
@@ -0,0 +1,33 @@
+package com.alibaba.druid.pool.ha.config;
+
+public class DruidDataSourceConfig {
+
+    private String url;
+    private String user;
+    private String password;
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getUser() {
+        return user;
+    }
+
+    public void setUser(String user) {
+        this.user = user;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+}
@@ -0,0 +1,18 @@
+package com.alibaba.druid.pool.ha.config;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultiDataSourceConfig {
+
+    private List<DruidDataSourceConfig> dataSources = new ArrayList<DruidDataSourceConfig>();
+
+    public List<DruidDataSourceConfig> getDataSources() {
+        return dataSources;
+    }
+
+    public void setDataSources(List<DruidDataSourceConfig> dataSources) {
+        this.dataSources = dataSources;
+    }
+
+}
@@ -0,0 +1,17 @@
+package com.alibaba.druid.pool.ha.config;
+
+import java.net.URL;
+
+public class URLConnectionConfigLoader {
+
+    private final URL configServerURL;
+
+    public URLConnectionConfigLoader(URL configServerURL){
+        this.configServerURL = configServerURL;
+    }
+
+    public URL getConfigServerURL() {
+        return configServerURL;
+    }
+
+}
@@ -0,0 +1,72 @@
+package com.alibaba.druid.pool.ha.valid;
+
+import java.sql.Connection;
+import java.sql.Statement;
+
+import com.alibaba.druid.logging.Log;
+import com.alibaba.druid.logging.LogFactory;
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.util.JdbcUtils;
+
+public class DefaultValidDataSourceChecker implements ValidDataSourceChecker {
+
+    private final static Log LOG           = LogFactory.getLog(DefaultValidDataSourceChecker.class);
+
+    private long             maxWaitMillis = 3000;
+
+    private String           validateSql   = "SELECT 1";
+
+    public long getMaxWaitMillis() {
+        return maxWaitMillis;
+    }
+
+    public void setMaxWaitMillis(long maxWaitMillis) {
+        this.maxWaitMillis = maxWaitMillis;
+    }
+
+    public String getValidateSql() {
+        return validateSql;
+    }
+
+    public void setValidateSql(String validateSql) {
+        this.validateSql = validateSql;
+    }
+
+    @Override
+    public boolean isValid(DruidDataSource dataSource) {
+        Connection conn = null;
+
+        try {
+            conn = dataSource.getConnection(maxWaitMillis);
+
+            if (conn == null) {
+                return false;
+            }
+
+            return isValidConnection(dataSource, conn);
+
+        } catch (Exception ex) {
+            LOG.error("check datasource valid errror", ex);
+        } finally {
+            JdbcUtils.close(conn);
+        }
+
+        return false;
+    }
+
+    public boolean isValidConnection(DruidDataSource dataSource, Connection conn) {
+        Statement stmt = null;
+
+        try {
+            stmt = conn.createStatement();
+            stmt.execute(validateSql);
+        } catch (Exception ex) {
+            LOG.error("check datasource valid errror", ex);
+        } finally {
+            JdbcUtils.close(stmt);
+        }
+
+        return true;
+    }
+
+}
@@ -0,0 +1,8 @@
+package com.alibaba.druid.pool.ha.valid;
+
+import com.alibaba.druid.pool.DruidDataSource;
+
+
+public interface ValidDataSourceChecker {
+    boolean isValid(DruidDataSource dataSource);
+}