ha-support

This commit is contained in:
wenshao
2011-12-13 14:35:58 +08:00
parent c31d0921c2
commit c870f42578
8 changed files with 298 additions and 31 deletions
@@ -1,5 +1,6 @@
package com.alibaba.druid.pool;
import javax.management.ObjectName;
import javax.sql.DataSource;
public interface ManagedDataSource extends DataSource {
@@ -7,4 +8,8 @@ public interface ManagedDataSource extends DataSource {
boolean isEnable();
void setEnable(boolean value);
ObjectName getObjectName();
void setObjectName(ObjectName objectName);
}
@@ -10,8 +10,9 @@ import javax.sql.DataSource;
import com.alibaba.druid.logging.Log;
import com.alibaba.druid.logging.LogFactory;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.ManagedDataSource;
public class HADataSource extends MultiDataSource implements HADataSourceMBean, DataSource {
public class HADataSource extends MultiDataSource implements HADataSourceMBean, ManagedDataSource, DataSource {
private final static Log LOG = LogFactory.getLog(HADataSource.class);
@@ -26,6 +27,8 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
}
public void resetStat() {
super.resetStat();
masterConnectCount.set(0);
slaveConnectCount.set(0);
}
@@ -71,7 +74,23 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
master.setEnable(value);
}
public String getMasterUrl() {
if (master == null) {
return null;
}
return master.getUrl();
}
public String getSlaveUrl() {
if (slave == null) {
return null;
}
return slave.getUrl();
}
public boolean isSlaveEnable() {
if (slave == null) {
return false;
@@ -127,5 +146,4 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
LOG.debug("HADataSource closed");
}
}
}
@@ -1,6 +1,6 @@
package com.alibaba.druid.pool.ha;
public interface HADataSourceMBean {
public interface HADataSourceMBean extends MultiDataSourceMBean {
long getMasterConnectCount();
long getSlaveConnectCount();
@@ -12,6 +12,10 @@ public interface HADataSourceMBean {
boolean isSlaveEnable();
void setSlaveEnable(boolean value);
String getMasterUrl();
String getSlaveUrl();
void switchMasterSlave();
@@ -1,7 +1,10 @@
package com.alibaba.druid.pool.ha;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
@@ -9,33 +12,78 @@ import java.util.concurrent.ConcurrentMap;
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.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.management.ObjectName;
import com.alibaba.druid.filter.Filter;
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;
import com.alibaba.druid.proxy.jdbc.DataSourceProxy;
import com.alibaba.druid.util.JdbcUtils;
public abstract class MultiDataSource extends DataSourceAdapter {
public abstract class MultiDataSource extends DataSourceAdapter implements MultiDataSourceMBean, DataSourceProxy {
private Properties properties = new Properties();
private Properties properties = new Properties();
private final AtomicInteger connectionIdSeed = new AtomicInteger();
private final AtomicInteger statementIdSeed = new AtomicInteger();
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 ValidDataSourceChecker validDataSourceChecker = new DefaultValidDataSourceChecker();
private long validDataSourceCheckPeriodMillis = 3000;
private ValidDataSourceChecker validDataSourceChecker = new DefaultValidDataSourceChecker();
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 ConcurrentMap<String, DruidDataSource> dataSources = new ConcurrentHashMap<String, DruidDataSource>();
private boolean inited = false;
private final Lock lock = new ReentrantLock();
private ConcurrentMap<String, DruidDataSource> dataSources = new ConcurrentHashMap<String, DruidDataSource>();
private ObjectName objectName;
private List<Filter> filters = new ArrayList<Filter>();
private boolean enable;
private String name;
public String getName() {
if (name == null) {
return "HADataSource-" + System.identityHashCode(this);
}
return name;
}
public String getNameInternal() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public boolean isEnable() {
return enable;
}
public void setEnable(boolean enable) {
this.enable = enable;
}
public ObjectName getObjectName() {
return objectName;
}
public void setObjectName(ObjectName objectName) {
this.objectName = objectName;
}
public MultiDataSource(){
@@ -53,21 +101,38 @@ public abstract class MultiDataSource extends DataSourceAdapter {
}
scheduler = Executors.newScheduledThreadPool(schedulerThreadCount);
scheduler.scheduleAtFixedRate(new ValidTask(), validDataSourceCheckPeriodMillis,
scheduler.scheduleAtFixedRate(new FailureDetectTask(), validDataSourceCheckPeriodMillis,
validDataSourceCheckPeriodMillis, TimeUnit.MILLISECONDS);
inited = true;
MultiDataSourceStatManager.add(this);
} finally {
lock.unlock();
}
}
public void resetStat() {
}
protected void close() {
scheduler.shutdownNow();
Object[] items = this.getDataSources().values().toArray();
for (Object item : items) {
JdbcUtils.close((DruidDataSource) item);
}
MultiDataSourceStatManager.remove(this);
}
public void failureDetect() {
for (DruidDataSource dataSource : getDataSources().values()) {
boolean isValid = validDataSourceChecker.isValid(dataSource);
if (isValid) {
handleNotAwailableDatasource(dataSource);
}
}
}
protected ScheduledExecutorService getScheduler() {
@@ -90,11 +155,11 @@ public abstract class MultiDataSource extends DataSourceAdapter {
this.validDataSourceChecker = validDataSourceChecker;
}
public int createConnectionId() {
public long createConnectionId() {
return connectionIdSeed.getAndIncrement();
}
public int createStatementId() {
public long createStatementId() {
return statementIdSeed.incrementAndGet();
}
@@ -116,16 +181,46 @@ public abstract class MultiDataSource extends DataSourceAdapter {
public abstract void handleNotAwailableDatasource(DruidDataSource dataSource);
class ValidTask implements Runnable {
@Override
public String getDbType() {
return null;
}
@Override
public Driver getRawDriver() {
return null;
}
@Override
public String getUrl() {
return null;
}
@Override
public String getRawJdbcUrl() {
return null;
}
@Override
public List<Filter> getProxyFilters() {
return filters;
}
@Override
public long createResultSetId() {
return resultSetIdSeed.incrementAndGet();
}
@Override
public long createTransactionId() {
return transactionIdSeed.incrementAndGet();
}
class FailureDetectTask implements Runnable {
@Override
public void run() {
for (DruidDataSource dataSource : getDataSources().values()) {
boolean isValid = validDataSourceChecker.isValid(dataSource);
if (isValid) {
handleNotAwailableDatasource(dataSource);
}
}
failureDetect();
}
}
@@ -34,7 +34,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect
private Connection conn;
private final int id;
private final long id;
private Boolean autoCommit = null;
private Boolean readOnly = null;
@@ -49,7 +49,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect
private Date connectedTime = null;
public MultiDataSourceConnection(MultiDataSource haDataSource, int id){
public MultiDataSourceConnection(MultiDataSource haDataSource, long id){
this.haDataSource = haDataSource;
this.id = id;
}
@@ -572,7 +572,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect
@Override
public DataSourceProxy getDirectDataSource() {
throw new UnsupportedOperationException();
return haDataSource;
}
@Override
@@ -0,0 +1,6 @@
package com.alibaba.druid.pool.ha;
public interface MultiDataSourceMBean {
}
@@ -0,0 +1,133 @@
/*
* Copyright 1999-2011 Alibaba Group Holding Ltd.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.druid.pool.ha;
import java.lang.management.ManagementFactory;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.JMException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.alibaba.druid.util.ConcurrentIdentityHashMap;
public class MultiDataSourceStatManager implements MultiDataSourceStatManagerMBean {
private final static Log LOG = LogFactory.getLog(MultiDataSourceStatManager.class);
private final static MultiDataSourceStatManager instance = new MultiDataSourceStatManager();
private final AtomicLong resetCount = new AtomicLong();
// global instances
private static final ConcurrentIdentityHashMap<MultiDataSource, ObjectName> dataSources = new ConcurrentIdentityHashMap<MultiDataSource, ObjectName>();
private final static String MBEAN_NAME = "com.alibaba.druid.ha:type=HADruidDataSourceStat";
public static MultiDataSourceStatManager getInstance() {
return instance;
}
public synchronized static void add(MultiDataSource dataSource) {
MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
if (dataSources.size() == 0) {
try {
mbeanServer.registerMBean(instance, new ObjectName(MBEAN_NAME));
} catch (JMException ex) {
LOG.error("register mbean error", ex);
}
}
ObjectName objectName = null;
if (dataSource.getNameInternal() != null) {
try {
objectName = new ObjectName("com.alibaba.druid.ha:type=DruidDataSource,id=" + dataSource.getNameInternal());
mbeanServer.registerMBean(dataSource, objectName);
} catch (JMException ex) {
LOG.error("register mbean error", ex);
objectName = null;
}
}
if (objectName == null) {
try {
int id = System.identityHashCode(dataSource);
objectName = new ObjectName("com.alibaba.druid.ha:type=DruidDataSource,id=" + id);
mbeanServer.registerMBean(dataSource, objectName);
} catch (JMException ex) {
LOG.error("register mbean error", ex);
objectName = null;
}
}
dataSources.put(dataSource, objectName);
dataSource.setObjectName(objectName);
}
public synchronized static void remove(MultiDataSource dataSource) {
ObjectName objectName = dataSources.remove(dataSource);
if (objectName == null) {
objectName = dataSource.getObjectName();
}
if (objectName == null) {
LOG.error("unregister mbean failed. ");
return;
}
MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
if (objectName != null) {
try {
mbeanServer.unregisterMBean(objectName);
} catch (JMException ex) {
LOG.error("unregister mbean error", ex);
}
}
if (dataSources.size() == 0) {
try {
mbeanServer.unregisterMBean(new ObjectName(MBEAN_NAME));
} catch (JMException ex) {
LOG.error("unregister mbean error", ex);
}
}
}
public static Set<MultiDataSource> getDruidDataSourceInstances() {
return dataSources.keySet();
}
public void reset() {
final Set<MultiDataSource> dataSources = getDruidDataSourceInstances();
for (MultiDataSource dataSource : dataSources) {
dataSource.resetStat();
}
resetCount.incrementAndGet();
}
public long getResetCount() {
return resetCount.get();
}
}
@@ -0,0 +1,6 @@
package com.alibaba.druid.pool.ha;
public interface MultiDataSourceStatManagerMBean {
}