ha

src/main/java/com/alibaba/druid/filter/stat/StatFilter.java

@@ -160,18 +160,21 @@ public class StatFilter extends FilterEventAdapter implements StatFilterMBean {
             connectStat.afterConnected(nanoSpan);
             dataSourceStat.getConnectionStat().afterConnected(nanoSpan);
 
-            JdbcConnectionStat.Entry statEntry = getConnectionInfo(connection);
-
-            dataSourceStat.getConnections().put(connection.getId(), statEntry);
-
-            statEntry.setConnectTime(new Date(startTime));
-            statEntry.setConnectTimespanNano(nanoSpan);
-            statEntry.setEstablishNano(System.nanoTime());
-            statEntry.setEstablishTime(nowTime);
-            statEntry.setConnectStackTrace(new Exception());
-
-            connectStat.setActiveCount(dataSourceStat.getConnections().size());
-            dataSourceStat.getConnectionStat().setActiveCount(dataSourceStat.getConnections().size());
+            if (connection != null) {
+                JdbcConnectionStat.Entry statEntry = getConnectionInfo(connection);
+    
+                dataSourceStat.getConnections().put(connection.getId(), statEntry);
+    
+                statEntry.setConnectTime(new Date(startTime));
+                statEntry.setConnectTimespanNano(nanoSpan);
+                statEntry.setEstablishNano(System.nanoTime());
+                statEntry.setEstablishTime(nowTime);
+                statEntry.setConnectStackTrace(new Exception());
+                
+                
+                connectStat.setActiveCount(dataSourceStat.getConnections().size());
+                dataSourceStat.getConnectionStat().setActiveCount(dataSourceStat.getConnections().size());
+            }
         }
         return connection;
     }

src/main/java/com/alibaba/druid/pool/ha/DataSourceHolder.java

@@ -16,6 +16,8 @@ public class DataSourceHolder implements Closeable {
 
     private boolean               fail              = false;
 
+    private int                   weight            = 1;
+
     public DataSourceHolder(DruidDataSource dataSource){
         if (dataSource == null) {
             throw new IllegalArgumentException("dataSource is null");
@@ -27,6 +29,14 @@ public class DataSourceHolder implements Closeable {
         connectCount.set(0);
     }
 
+    public int getWeight() {
+        return weight;
+    }
+
+    public void setWeight(int weight) {
+        this.weight = weight;
+    }
+
     public boolean isEnable() {
         return enable && (!fail);
     }

src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java

@@ -20,6 +20,7 @@ import javax.management.ObjectName;
 
 import com.alibaba.druid.filter.Filter;
 import com.alibaba.druid.pool.DataSourceAdapter;
+import com.alibaba.druid.pool.ha.config.ConfigLoader;
 import com.alibaba.druid.pool.ha.valid.DataSourceFailureDetecter;
 import com.alibaba.druid.pool.ha.valid.DefaultDataSourceFailureDetecter;
 import com.alibaba.druid.proxy.jdbc.DataSourceProxy;
@@ -27,31 +28,43 @@ import com.alibaba.druid.util.JdbcUtils;
 
 public abstract class MultiDataSource extends DataSourceAdapter implements MultiDataSourceMBean, DataSourceProxy {
 
-    private Properties                             properties                       = new Properties();
+    private Properties                              properties                       = new Properties();
 
-    private final AtomicLong                       connectionIdSeed                 = new AtomicLong();
-    private final AtomicLong                       statementIdSeed                  = new AtomicLong();
-    private final AtomicLong                       resultSetIdSeed                  = new AtomicLong();
-    private final AtomicLong                       transactionIdSeed                = new AtomicLong();
+    private final AtomicLong                        connectionIdSeed                 = new AtomicLong();
+    private final AtomicLong                        statementIdSeed                  = new AtomicLong();
+    private final AtomicLong                        resultSetIdSeed                  = new AtomicLong();
+    private final AtomicLong                        transactionIdSeed                = new AtomicLong();
 
-    protected DataSourceFailureDetecter               validDataSourceChecker           = new DefaultDataSourceFailureDetecter();
-    private long                                   validDataSourceCheckPeriodMillis = 3000;
+    protected DataSourceFailureDetecter             validDataSourceChecker           = new DefaultDataSourceFailureDetecter();
+    private long                                    validDataSourceCheckPeriodMillis = 3000;
 
-    private int                                    schedulerThreadCount             = 3;
-    private ScheduledExecutorService               scheduler;
+    private int                                     schedulerThreadCount             = 3;
+    private ScheduledExecutorService                scheduler;
 
-    private boolean                                inited                           = false;
-    private final Lock                             lock                             = new ReentrantLock();
+    private boolean                                 inited                           = false;
+    protected final Lock                            lock                             = new ReentrantLock();
 
     private ConcurrentMap<String, DataSourceHolder> dataSources                      = new ConcurrentHashMap<String, DataSourceHolder>();
 
-    private ObjectName                             objectName;
+    private ObjectName                              objectName;
 
-    private List<Filter>                           filters                          = new ArrayList<Filter>();
+    private List<Filter>                            filters                          = new ArrayList<Filter>();
 
-    private boolean                                enable;
+    private boolean                                 enable;
 
-    private String                                 name;
+    private String                                  name;
+
+    private long                                    totalWeight                      = 0;
+
+    private ConfigLoader                            configLoader;
+
+    public ConfigLoader getConfigLoader() {
+        return configLoader;
+    }
+
+    public void setConfigLoader(ConfigLoader configLoader) {
+        this.configLoader = configLoader;
+    }
 
     public String getName() {
         if (name == null) {
@@ -68,6 +81,10 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
         this.name = name;
     }
 
+    public boolean isIntited() {
+        return this.inited;
+    }
+
     public boolean isEnable() {
         return enable;
     }
@@ -88,7 +105,7 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
 
     }
 
-    public void init() {
+    public void init() throws SQLException {
         if (inited) {
             return;
         }
@@ -99,6 +116,8 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
                 return;
             }
 
+            initInternal();
+
             scheduler = Executors.newScheduledThreadPool(schedulerThreadCount);
             scheduler.scheduleAtFixedRate(new FailureDetectTask(), validDataSourceCheckPeriodMillis,
                                           validDataSourceCheckPeriodMillis, TimeUnit.MILLISECONDS);
@@ -110,6 +129,10 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
         }
     }
 
+    protected void initInternal() throws SQLException {
+
+    }
+
     public void resetStat() {
 
     }
@@ -167,10 +190,28 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
     public Map<String, DataSourceHolder> getDataSources() {
         return dataSources;
     }
+    
+    public DataSourceHolder getDataSourceHolder(String name) {
+        return dataSources.get(name);
+    }
+
+    public void addDataSource(String name, DataSourceHolder dataSourceHolder) {
+        dataSources.put(name, dataSourceHolder);
+        
+        this.totalWeight += dataSourceHolder.getWeight();
+    }
 
     public Properties getProperties() {
         return properties;
     }
+    
+    public void computeTotalWeight() {
+        int totalWeight = 0;
+        for (DataSourceHolder holder : this.dataSources.values()) {
+            totalWeight += holder.getWeight();
+        }
+        this.totalWeight = totalWeight;
+    }
 
     public Connection getConnection() throws SQLException {
         init();
@@ -178,11 +219,12 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
         return new MultiDataSourceConnection(this, createConnectionId());
     }
 
-    public abstract MultiConnectionHolder getConnectionInternal(MultiDataSourceConnection conn, String sql) throws SQLException;
+    public abstract MultiConnectionHolder getConnectionInternal(MultiDataSourceConnection conn, String sql)
+                                                                                                           throws SQLException;
 
     public void handleNotAwailableDatasource(DataSourceHolder dataSourceHolder) {
     }
-    
+
     public String[] getDataSourceNames() {
         return this.dataSources.keySet().toArray(new String[this.dataSources.size()]);
     }
@@ -221,14 +263,14 @@ public abstract class MultiDataSource extends DataSourceAdapter implements Multi
     public long createTransactionId() {
         return transactionIdSeed.incrementAndGet();
     }
-    
+
     public boolean restartDataSource(String name) {
         DataSourceHolder holder = this.getDataSources().get(name);
         if (holder != null) {
             holder.restart();
             return true;
         }
-        
+
         return false;
     }
 

src/main/java/com/alibaba/druid/pool/ha/cobar/CobarConfigLoader.java

@@ -0,0 +1,117 @@
+package com.alibaba.druid.pool.ha.cobar;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.sql.SQLException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+import com.alibaba.druid.pool.ha.DataSourceHolder;
+import com.alibaba.druid.pool.ha.config.ConfigLoader;
+import com.alibaba.druid.pool.ha.config.URLConnectionConfigLoader;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+
+public class CobarConfigLoader extends URLConnectionConfigLoader implements ConfigLoader {
+
+    private final CobarDataSource dataSource;
+
+    public CobarConfigLoader(CobarDataSource cobarDataSource) throws SQLException{
+        super(createURL(cobarDataSource.getUrl()));
+
+        this.dataSource = cobarDataSource;
+    }
+
+    public CobarDataSource getDataSource() {
+        return dataSource;
+    }
+
+    public static URL createURL(String jdbcUrl) throws SQLException {
+        if (!CobarDataSource.isCobar(jdbcUrl)) {
+            throw new SQLException("illegal cobar url");
+        }
+
+        String rest = jdbcUrl.substring("jdbc:cobar://".length());
+        String[] items = rest.split("/");
+        String loginServerIp = items[0];
+        String schemaName = items[1];
+
+        int port = 80;
+        {
+            int pos = loginServerIp.indexOf(':');
+            if (pos != -1) {
+                loginServerIp = loginServerIp.substring(0, pos);
+                port = Integer.parseInt(loginServerIp.substring(pos + 1));
+            }
+        }
+
+        String url = "http://" + loginServerIp;
+
+        if (port != 80) {
+            url += ":" + port;
+        }
+
+        url += "cobarStatusQuery?sid=" + schemaName;
+
+        try {
+            return new URL(url);
+        } catch (MalformedURLException e) {
+            throw new SQLException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void handleResponseMessage(String response) throws SQLException {
+        JSONObject json = JSON.parseObject(response);
+        String errorMessage = json.getString("error");
+        if (errorMessage != null) {
+            throw new SQLException("load config error, message : " + errorMessage);
+        }
+
+        JSONArray array = json.getJSONArray("cobarList");
+
+        Set<String> keys = new HashSet<String>();
+        for (int i = 0; i < array.size(); ++i) {
+            JSONObject item = array.getJSONObject(i);
+
+            String ip = item.getString("ip");
+            int port = item.getIntValue("port");
+            String schema = item.getString("schema");
+            int weight = item.getIntValue("weight");
+
+            String key = ip + ":" + port + "/" + schema;
+            keys.add(key);
+
+            DataSourceHolder holder = dataSource.getDataSourceHolder(key);
+
+            if (holder == null) {
+                String jdbcUrl = "jdbc:mysql://" + ip + ":" + port + "/" + schema;
+                holder = dataSource.createDataSourceHolder(jdbcUrl, weight);
+                dataSource.addDataSource(key, holder);
+            } else {
+                if (holder.getWeight() != weight) {
+                    holder.setWeight(weight);
+                    dataSource.computeTotalWeight();
+                }
+            }
+        }
+
+        int removeCount = 0;
+        Iterator<Map.Entry<String, DataSourceHolder>> iter = dataSource.getDataSources().entrySet().iterator();
+        while (iter.hasNext()) {
+            Map.Entry<String, DataSourceHolder> entry = iter.next();
+            if (!keys.contains(entry.getKey())) {
+                iter.remove();
+                dataSource.handleDataSourceDiscard(entry.getValue());
+                removeCount++;
+            }
+        }
+        if (removeCount != 0) {
+            dataSource.computeTotalWeight();
+        }
+    }
+
+}

src/main/java/com/alibaba/druid/pool/ha/cobar/CobarDataSource.java

@@ -0,0 +1,116 @@
+package com.alibaba.druid.pool.ha.cobar;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.alibaba.druid.filter.Filter;
+import com.alibaba.druid.logging.Log;
+import com.alibaba.druid.logging.LogFactory;
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.pool.ha.DataSourceHolder;
+import com.alibaba.druid.pool.ha.MultiConnectionHolder;
+import com.alibaba.druid.pool.ha.MultiDataSource;
+import com.alibaba.druid.pool.ha.MultiDataSourceConnection;
+import com.alibaba.druid.util.JdbcUtils;
+
+public class CobarDataSource extends MultiDataSource {
+
+    private final static Log LOG          = LogFactory.getLog(CobarDataSource.class);
+
+    private String           url;
+
+    private String           username;
+
+    private String           password;
+
+    private List<Filter>     proxyFilters = new ArrayList<Filter>();
+    private String           filters;
+
+    public CobarDataSource(){
+
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public List<Filter> getProxyFilters() {
+        return proxyFilters;
+    }
+
+    public void setProxyFilters(List<Filter> proxyFilters) {
+        this.proxyFilters = proxyFilters;
+    }
+
+    public String getFilters() {
+        return filters;
+    }
+
+    public void setFilters(String filters) {
+        this.filters = filters;
+    }
+
+    protected void initInternal() throws SQLException {
+        if (url == null || url.isEmpty()) {
+            throw new SQLException("");
+        }
+
+        if (isCobar(url)) {
+            this.setConfigLoader(new CobarConfigLoader(this));
+        } else {
+            DataSourceHolder holder = createDataSourceHolder(this.url, 1);
+            this.addDataSource("master", holder);
+        }
+    }
+
+    protected DataSourceHolder createDataSourceHolder(String url, int weight) throws SQLException {
+        DruidDataSource dataSource = new DruidDataSource();
+        dataSource.setUrl(url);
+        dataSource.setUsername(username);
+        dataSource.setPassword(password);
+        if (filters != null && !filters.isEmpty()) {
+            dataSource.setFilters(filters);
+        }
+        dataSource.setProxyFilters(proxyFilters);
+
+        DataSourceHolder holder = new DataSourceHolder(dataSource);
+        holder.setWeight(weight);
+        return holder;
+    }
+
+    protected void handleDataSourceDiscard(DataSourceHolder holder) {
+        LOG.debug("dataSource close");
+        JdbcUtils.close(holder);
+    }
+
+    public static boolean isCobar(String url) {
+        return url.startsWith("jdbc:cobar://");
+    }
+
+    @Override
+    public MultiConnectionHolder getConnectionInternal(MultiDataSourceConnection conn, String sql) throws SQLException {
+        return null;
+    }
+
+}

src/main/java/com/alibaba/druid/pool/ha/config/ConfigLoader.java

@@ -1,5 +1,7 @@
 package com.alibaba.druid.pool.ha.config;
 
+import java.sql.SQLException;
+
 public interface ConfigLoader {
-    MultiDataSourceConfig load();
+    void load() throws SQLException;
 }

src/main/java/com/alibaba/druid/pool/ha/config/URLConnectionConfigLoader.java

@@ -5,12 +5,11 @@ import java.io.Reader;
 import java.net.HttpURLConnection;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.sql.SQLException;
 
-import com.alibaba.druid.DruidRuntimeException;
 import com.alibaba.druid.util.JdbcUtils;
-import com.alibaba.fastjson.JSON;
 
-public class URLConnectionConfigLoader implements ConfigLoader {
+public abstract class URLConnectionConfigLoader implements ConfigLoader {
 
     private URL url;
     private int connectTimeout = 1000 * 3;
@@ -49,7 +48,7 @@ public class URLConnectionConfigLoader implements ConfigLoader {
     }
 
     @Override
-    public MultiDataSourceConfig load() {
+    public void load() throws SQLException {
         if (url == null) {
             throw new IllegalStateException("configServerURL is null");
         }
@@ -69,16 +68,17 @@ public class URLConnectionConfigLoader implements ConfigLoader {
 
                 reader = new InputStreamReader(conn.getInputStream());
                 responseMessage = JdbcUtils.read(reader);
+                
+                handleResponseMessage(responseMessage);
             } catch (Exception e) {
-                throw new DruidRuntimeException("load config error, url : " + url.toString());
+                throw new SQLException("load config error, url : " + url.toString());
             } finally {
                 JdbcUtils.close(reader);
                 conn.disconnect();
             }
         }
-
-        MultiDataSourceConfig config = JSON.parseObject(responseMessage, MultiDataSourceConfig.class);
-        return config;
     }
+    
+    protected abstract void handleResponseMessage(String response) throws SQLException;
 
 }