This commit is contained in:
wenshao
2011-12-23 16:51:11 +08:00
parent 9d4d45512b
commit 9207dd46a7
4 changed files with 127 additions and 10 deletions
@@ -22,6 +22,16 @@ public class DataSourceHolder implements Closeable {
private transient int weightRegionEnd;
private boolean writable = true;
public boolean isWritable() {
return writable;
}
public void setWritable(boolean writable) {
this.writable = writable;
}
public int getWeightRegionBegin() {
return weightRegionBegin;
}
@@ -93,7 +103,7 @@ public class DataSourceHolder implements Closeable {
connectCount.incrementAndGet();
try {
Connection conn = dataSource.getConnection();
return new MultiConnectionHolder(this, conn);
} catch (SQLException ex) {
connectErrorCount.incrementAndGet();
@@ -24,6 +24,14 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
master.resetState();
slave.resetState();
}
public boolean isMasterWritable() {
return master != null && master.isWritable();
}
public boolean isSlaveWritable() {
return slave != null && slave.isWritable();
}
public long getMasterConnectCount() {
return master.getConnectCount();
@@ -45,8 +53,10 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
this.restartDataSource("slave");
}
public void setMaster(DruidDataSource master) {
this.setMaster(new DataSourceHolder(master));
public DataSourceHolder setMaster(DruidDataSource master) {
DataSourceHolder holder = new DataSourceHolder(master);
this.setMaster(holder);
return holder;
}
public void setMaster(DataSourceHolder master) {
@@ -58,8 +68,10 @@ public class HADataSource extends MultiDataSource implements HADataSourceMBean,
return slave;
}
public void setSlave(DruidDataSource slave) {
this.setSlave(new DataSourceHolder(slave));
public DataSourceHolder setSlave(DruidDataSource slave) {
DataSourceHolder holder = new DataSourceHolder(slave);
this.setSlave(holder);
return holder;
}
public void setSlave(DataSourceHolder slave) {
@@ -0,0 +1,22 @@
package com.alibaba.druid.pool.ha.config;
import java.net.URL;
import java.sql.SQLException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
public class HAConfigLoader extends URLConnectionConfigLoader {
public HAConfigLoader(URL configServerURL){
super(configServerURL);
}
@Override
protected void handleResponseMessage() throws SQLException {
JSONObject json = JSON.parseObject(this.getResponseMessage());
json.get("serverList");
}
}
@@ -4,14 +4,17 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import org.junit.Assert;
import java.util.concurrent.atomic.AtomicBoolean;
import junit.framework.TestCase;
import org.junit.Assert;
import com.alibaba.druid.mock.MockConnection;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.ha.cobar.CobarConfigLoader;
import com.alibaba.druid.pool.ha.cobar.CobarDataSource;
import com.alibaba.druid.pool.ha.cobar.CobarFailureDetecter;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
@@ -40,8 +43,11 @@ public class CobarDataSourceTest extends TestCase {
cobarList.add(cobarB);
String mockUrlA = "jdbc:mock://mock:80/cobarA";
String mockUrlB = "jdbc:mock://mock:81/cobarB";
final String mockUrlA = "jdbc:mock://mock:80/cobarA";
final String mockUrlB = "jdbc:mock://mock:81/cobarB";
final AtomicBoolean statusA = new AtomicBoolean(true);
final AtomicBoolean statusB = new AtomicBoolean(true);
config.put("cobarList", cobarList);
@@ -50,7 +56,8 @@ public class CobarDataSourceTest extends TestCase {
dataSource.setUsername("test");
dataSource.setPassword("");
dataSource.setMaxWait(1);
dataSource.setConfigLoadPeriodMillis(3);
dataSource.setConfigLoadPeriodMillis(10);
dataSource.setFailureDetectPeriodMillis(1);
CobarConfigLoader configLoader = new CobarConfigLoader(dataSource) {
@@ -65,6 +72,23 @@ public class CobarDataSourceTest extends TestCase {
}
};
dataSource.setConfigLoader(configLoader);
CobarFailureDetecter failureDetector = new CobarFailureDetecter() {
public boolean isValidConnection(DruidDataSource dataSource, Connection conn) {
MockConnection mockConn;
try {
mockConn = conn.unwrap(MockConnection.class);
} catch (SQLException e) {
e.printStackTrace();
return false;
}
if (mockConn.getUrl().equals(mockUrlA)) {
return statusA.get();
}
return statusB.get();
}
};
dataSource.setFailureDetector(failureDetector);
{
Connection conn = dataSource.getConnection();
@@ -113,6 +137,55 @@ public class CobarDataSourceTest extends TestCase {
stmt.close();
conn.close();
}
cobarList.add(cobarA);
Assert.assertEquals(2, cobarList.size());
Thread.sleep(dataSource.getConfigLoadPeriodMillis() * 2);
for (int i = 0; i < 100; ++i) {
Connection conn = dataSource.getConnection();
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("select 1");
rs.next();
rs.close();
stmt.close();
conn.close();
}
statusA.set(false);
Thread.sleep(dataSource.getFailureDetectPeriodMillis() * 2);
for (int i = 0; i < 100; ++i) {
Connection conn = dataSource.getConnection();
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("select 1");
MockConnection mockConn = conn.unwrap(MockConnection.class);
Assert.assertNotNull(mockConn);
Assert.assertEquals(mockUrlB, mockConn.getUrl());
rs.next();
rs.close();
stmt.close();
conn.close();
}
statusA.set(true);
statusB.set(false);
Thread.sleep(dataSource.getFailureDetectPeriodMillis() * 2);
for (int i = 0; i < 100; ++i) {
Connection conn = dataSource.getConnection();
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("select 1");
MockConnection mockConn = conn.unwrap(MockConnection.class);
Assert.assertNotNull(mockConn);
Assert.assertEquals(mockUrlA, mockConn.getUrl());
rs.next();
rs.close();
stmt.close();
conn.close();
}
}
}