From 03b41811ea56e0618b541bc05c6d271364303aef Mon Sep 17 00:00:00 2001 From: wenshao Date: Tue, 20 Dec 2011 22:37:21 +0800 Subject: [PATCH] ha --- .../alibaba/druid/filter/stat/StatFilter.java | 27 ++-- .../druid/pool/ha/DataSourceHolder.java | 10 ++ .../druid/pool/ha/MultiDataSource.java | 82 +++++++++--- .../pool/ha/cobar/CobarConfigLoader.java | 117 ++++++++++++++++++ .../druid/pool/ha/cobar/CobarDataSource.java | 116 +++++++++++++++++ .../druid/pool/ha/config/ConfigLoader.java | 4 +- .../ha/config/URLConnectionConfigLoader.java | 16 +-- 7 files changed, 331 insertions(+), 41 deletions(-) create mode 100644 src/main/java/com/alibaba/druid/pool/ha/cobar/CobarConfigLoader.java create mode 100644 src/main/java/com/alibaba/druid/pool/ha/cobar/CobarDataSource.java diff --git a/src/main/java/com/alibaba/druid/filter/stat/StatFilter.java b/src/main/java/com/alibaba/druid/filter/stat/StatFilter.java index f163333eb1..6b97631c77 100644 --- a/src/main/java/com/alibaba/druid/filter/stat/StatFilter.java +++ b/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; } diff --git a/src/main/java/com/alibaba/druid/pool/ha/DataSourceHolder.java b/src/main/java/com/alibaba/druid/pool/ha/DataSourceHolder.java index e3ef2e3bfa..159a2835eb 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/DataSourceHolder.java +++ b/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); } diff --git a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java index d87f125243..bbe5d2e804 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java +++ b/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 dataSources = new ConcurrentHashMap(); - private ObjectName objectName; + private ObjectName objectName; - private List filters = new ArrayList(); + private List filters = new ArrayList(); - 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 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; } diff --git a/src/main/java/com/alibaba/druid/pool/ha/cobar/CobarConfigLoader.java b/src/main/java/com/alibaba/druid/pool/ha/cobar/CobarConfigLoader.java new file mode 100644 index 0000000000..8c55b633d5 --- /dev/null +++ b/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 keys = new HashSet(); + 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> iter = dataSource.getDataSources().entrySet().iterator(); + while (iter.hasNext()) { + Map.Entry entry = iter.next(); + if (!keys.contains(entry.getKey())) { + iter.remove(); + dataSource.handleDataSourceDiscard(entry.getValue()); + removeCount++; + } + } + if (removeCount != 0) { + dataSource.computeTotalWeight(); + } + } + +} diff --git a/src/main/java/com/alibaba/druid/pool/ha/cobar/CobarDataSource.java b/src/main/java/com/alibaba/druid/pool/ha/cobar/CobarDataSource.java new file mode 100644 index 0000000000..7a296df1c0 --- /dev/null +++ b/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 proxyFilters = new ArrayList(); + 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 getProxyFilters() { + return proxyFilters; + } + + public void setProxyFilters(List 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; + } + +} diff --git a/src/main/java/com/alibaba/druid/pool/ha/config/ConfigLoader.java b/src/main/java/com/alibaba/druid/pool/ha/config/ConfigLoader.java index 6ef218fa0f..cb0cc033ae 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/config/ConfigLoader.java +++ b/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; } diff --git a/src/main/java/com/alibaba/druid/pool/ha/config/URLConnectionConfigLoader.java b/src/main/java/com/alibaba/druid/pool/ha/config/URLConnectionConfigLoader.java index 8c4cb7c068..fabf69d0cd 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/config/URLConnectionConfigLoader.java +++ b/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; }