This commit is contained in:
wenshao
2011-12-22 20:13:05 +08:00
parent 447a709ef7
commit a04b6faa3b
17 changed files with 384 additions and 22 deletions
@@ -28,6 +28,7 @@ import com.alibaba.druid.sql.ast.statement.SQLSelectQueryBlock;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
import com.alibaba.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlBooleanExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.CobarShowStatus;
import com.alibaba.druid.sql.dialect.mysql.parser.MySqlStatementParser;
import com.alibaba.druid.sql.parser.SQLStatementParser;
@@ -48,6 +49,10 @@ public class MySqlMockExecuteHandlerImpl implements MockExecuteHandler {
SQLStatement stmt = stmtList.get(0);
if (stmt instanceof CobarShowStatus) {
return showStatus(statement);
}
if (!(stmt instanceof SQLSelectStatement)) {
throw new SQLException("executeQueryError : " + sql);
}
@@ -80,7 +85,21 @@ public class MySqlMockExecuteHandlerImpl implements MockExecuteHandler {
} else {
throw new SQLException("TODO");
}
}
public ResultSet showStatus(MockStatement statement) throws SQLException {
MockResultSet rs = new MockResultSet(statement);
MockResultSetMetaData metaData = rs.getMockMetaData();
Object[] row = new Object[] { "on" };
ColumnMetaData column = new ColumnMetaData();
column.setColumnType(Types.NVARCHAR);
metaData.getColumns().add(column);
rs.getRows().add(row);
return rs;
}
public ResultSet executeQueryFromDual(MockStatement statement, SQLSelectQueryBlock query) throws SQLException {
@@ -197,6 +197,12 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
private ObjectName objectName;
protected Exception createError = null;
public Exception getCreateError() {
return createError;
}
public boolean isDupCloseLogEnable() {
return dupCloseLogEnable;
}
@@ -1137,11 +1143,14 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
}
dataSource.validateConnection(conn);
dataSource.createError = null;
} catch (SQLException ex) {
dataSource.createErrorCount++;
dataSource.createError = ex;
throw ex;
} catch (RuntimeException ex) {
dataSource.createErrorCount++;
dataSource.createError = ex;
throw ex;
} catch (Error ex) {
dataSource.createErrorCount++;
@@ -735,7 +735,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
continue;
}
throw new GetConnectionTimeoutException();
if (createError != null) {
throw new GetConnectionTimeoutException(createError);
} else {
throw new GetConnectionTimeoutException();
}
}
}
@@ -61,6 +61,7 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
private boolean inited = false;
protected final Lock lock = new ReentrantLock();
protected final Condition notFull = lock.newCondition();
protected final Condition notFail = lock.newCondition();
private ConcurrentMap<String, DataSourceHolder> dataSources = new ConcurrentHashMap<String, DataSourceHolder>();
@@ -81,6 +82,23 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
private int maxPoolSize = 50;
private long activeCount = 0;
private long maxWaitMillis = 0;
public long getMaxWaitMillis() {
return maxWaitMillis;
}
public void setMaxWaitMillis(long maxWaitMillis) {
this.maxWaitMillis = maxWaitMillis;
}
public void setMaxWait(long seconds) {
this.setMaxWaitMillis(1000 * seconds);
}
public long getMaxWait() {
return this.getMaxWaitMillis() / 1000;
}
public int getMaxPoolSize() {
return maxPoolSize;
@@ -162,6 +180,7 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
scheduler = Executors.newScheduledThreadPool(schedulerThreadCount);
startFailureDetectScheduleTask();
startConfigLoadScheduleTask();
inited = true;
MultiDataSourceStatManager.add(this);
@@ -171,7 +190,7 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
}
public boolean startConfigLoadScheduleTask() {
if (configLoadFuture != null) {
if (configLoadFuture == null) {
configLoadFuture = scheduler.scheduleAtFixedRate(new ConfigLoadTask(), configLoadPeriodMillis,
configLoadPeriodMillis, TimeUnit.MILLISECONDS);
return true;
@@ -309,6 +328,13 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
holder.setWeightRegionEnd(totalWeight);
}
this.totalWeight = totalWeight;
lock.lock();
try {
notFail.signalAll();
} finally {
lock.unlock();
}
}
public Connection getConnection() throws SQLException {
@@ -341,13 +367,21 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
lock.unlock();
}
}
public int produceRandomNumber() {
if (totalWeight == 0) {
return 0;
}
return random.nextInt(totalWeight);
}
public MultiConnectionHolder getConnectionInternal(MultiDataSourceConnection multiConn, String sql)
throws SQLException {
public MultiConnectionHolder getRealConnection(MultiDataSourceConnection multiConn, String sql) throws SQLException {
long startNano = -1;
if (maxWaitMillis > 0) {
startNano = System.nanoTime();
}
DataSourceHolder dataSource = null;
@@ -385,12 +419,52 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc
if (dataSource == null) {
dataSource = first;
}
if (dataSource == null && i != MAX_RETRY - 1) {
lock.lock();
try {
if (getEnabledDataSourceCount() == 0) {
try {
if (maxWaitMillis > 0) {
long nano = System.nanoTime() - startNano;
long restNano = maxWaitMillis * 1000 * 1000 - nano;
if (restNano > 0) {
notFail.awaitNanos(restNano);
} else {
break;
}
} else {
notFail.await();
}
continue;
} catch (InterruptedException e) {
throw new SQLException("interrupted", e);
}
}
} finally {
lock.unlock();
}
}
break;
}
if (dataSource == null) {
throw new SQLException("cannot get connection. enabledDataSourceCount " + getEnabledDataSourceCount());
}
return dataSource.getConnection();
}
public int getEnabledDataSourceCount() {
int count = 0;
for (DataSourceHolder item : this.dataSources.values()) {
if (item.isEnable()) {
count++;
}
}
return count;
}
public void handleNotAwailableDatasource(DataSourceHolder dataSourceHolder) {
}
@@ -62,7 +62,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect
public void checkConnection(String sql) throws SQLException {
if (conn == null) {
MultiConnectionHolder connHolder = haDataSource.getConnectionInternal(this, sql);
MultiConnectionHolder connHolder = haDataSource.getRealConnection(this, sql);
conn = connHolder.getConnection();
this.dataSourceHolder = connHolder.getDataSourceHolder();
}
@@ -48,8 +48,8 @@ public class CobarConfigLoader extends URLConnectionConfigLoader implements Conf
{
int pos = ip.indexOf(':');
if (pos != -1) {
ip = ip.substring(0, pos);
port = Integer.parseInt(ip.substring(pos + 1));
ip = ip.substring(0, pos);
}
}
@@ -64,8 +64,10 @@ public class CobarConfigLoader extends URLConnectionConfigLoader implements Conf
}
@Override
protected void handleResponseMessage(String response) throws SQLException {
JSONObject json = JSON.parseObject(response);
protected void handleResponseMessage() throws SQLException {
String responseMessage = this.getResponseMessage();
JSONObject json = JSON.parseObject(responseMessage);
String errorMessage = json.getString("error");
if (errorMessage != null) {
throw new SQLException("load config error, message : " + errorMessage);
@@ -88,7 +90,7 @@ public class CobarConfigLoader extends URLConnectionConfigLoader implements Conf
DataSourceHolder holder = dataSource.getDataSourceHolder(key);
if (holder == null) {
String jdbcUrl = "jdbc:mysql://" + ip + ":" + port + "/" + schema;
String jdbcUrl = createJdbcUrl(ip, port, schema);
holder = dataSource.createDataSourceHolder(jdbcUrl, weight);
dataSource.addDataSource(key, holder);
} else {
@@ -114,4 +116,8 @@ public class CobarConfigLoader extends URLConnectionConfigLoader implements Conf
}
}
protected String createJdbcUrl(String ip, int port, String schema) {
return "jdbc:mysql://" + ip + ":" + port + "/" + schema;
}
}
@@ -35,6 +35,7 @@ public class CobarDataSource extends MultiDataSource {
private List<Filter> proxyFilters = new ArrayList<Filter>();
private String filters;
public CobarDataSource(){
this.setFailureDetector(new CobarFailureDetecter());
this.setFailureDetectPeriodMillis(DEFAULT_FAILURE_DETECT_PERRIOD_MILLIS);
@@ -103,8 +104,10 @@ public class CobarDataSource extends MultiDataSource {
}
if (CobarConfigLoader.isCobar(url)) {
this.setConfigLoader(new CobarConfigLoader(this));
if (this.getConfigLoader() == null) {
this.setConfigLoader(new CobarConfigLoader(this));
}
final int RETRY_COUNT = 3;
for (int i = 0; i < RETRY_COUNT; ++i) {
try {
@@ -11,9 +11,11 @@ import com.alibaba.druid.util.JdbcUtils;
public abstract class URLConnectionConfigLoader implements ConfigLoader {
private URL url;
private int connectTimeout = 1000 * 3;
private int readTimeout;
private URL url;
private int connectTimeout = 1000 * 3;
private int readTimeout;
protected String responseMessage;
public URLConnectionConfigLoader(URL configServerURL){
this.url = configServerURL;
@@ -47,14 +49,16 @@ public abstract class URLConnectionConfigLoader implements ConfigLoader {
this.readTimeout = readTimeout;
}
public String getResponseMessage() {
return responseMessage;
}
@Override
public void load() throws SQLException {
if (url == null) {
throw new IllegalStateException("configServerURL is null");
}
String responseMessage;
{
HttpURLConnection conn = null;
Reader reader = null;
@@ -68,17 +72,17 @@ public abstract class URLConnectionConfigLoader implements ConfigLoader {
reader = new InputStreamReader(conn.getInputStream());
responseMessage = JdbcUtils.read(reader);
handleResponseMessage(responseMessage);
handleResponseMessage();
} catch (Exception e) {
throw new SQLException("load config error, url : " + url.toString());
throw new SQLException("load config error, url : " + url.toString(), e);
} finally {
JdbcUtils.close(reader);
conn.disconnect();
}
}
}
protected abstract void handleResponseMessage(String response) throws SQLException;
protected abstract void handleResponseMessage() throws SQLException;
}
@@ -0,0 +1,14 @@
package com.alibaba.druid.sql.dialect.mysql.ast.statement;
import com.alibaba.druid.sql.dialect.mysql.visitor.MySqlASTVisitor;
public class CobarShowStatus extends MySqlStatementImpl {
private static final long serialVersionUID = 1L;
protected void accept0(MySqlASTVisitor visitor) {
visitor.visit(this);
visitor.endVisit(this);
}
}
@@ -29,6 +29,7 @@ import com.alibaba.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
import com.alibaba.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.druid.sql.ast.statement.SQLUpdateSetItem;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.CobarShowStatus;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCommitStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCreateTableParser;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlDeleteStatement;
@@ -292,6 +293,11 @@ public class MySqlStatementParser extends SQLStatementParser {
return stmt;
}
}
if (identifierEquals("COBAR_STATUS")) {
lexer.nextToken();
return new CobarShowStatus();
}
throw new ParserException("TODO");
}
@@ -22,6 +22,7 @@ import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlBooleanExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlExtractExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlIntervalExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlMatchAgainstExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.CobarShowStatus;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCommitStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlDeleteStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlExecuteStatement;
@@ -143,4 +144,8 @@ public interface MySqlASTVisitor extends SQLASTVisitor {
void endVisit(MySqlShowStatusStatement x);
boolean visit(MySqlShowStatusStatement x);
void endVisit(CobarShowStatus x);
boolean visit(CobarShowStatus x);
}
@@ -7,6 +7,7 @@ import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlBooleanExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlExtractExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlIntervalExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlMatchAgainstExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.CobarShowStatus;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCommitStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlDeleteStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlExecuteStatement;
@@ -300,4 +301,14 @@ public class MySqlASTVisitorAdapter extends SQLASTVisitorAdapter implements MySq
return true;
}
@Override
public void endVisit(CobarShowStatus x) {
}
@Override
public boolean visit(CobarShowStatus x) {
return true;
}
}
@@ -36,6 +36,7 @@ import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlCharExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlExtractExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlIntervalExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlMatchAgainstExpr;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.CobarShowStatus;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCommitStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCreateTableStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlDeleteStatement;
@@ -1190,4 +1191,14 @@ public class MySqlOutputVisitor extends SQLASTOutputVisitor implements MySqlASTV
return false;
}
@Override
public void endVisit(CobarShowStatus x) {
}
@Override
public boolean visit(CobarShowStatus x) {
print("SHOW COBAR_STATUS");
return false;
}
}
@@ -501,6 +501,9 @@ public final class JdbcUtils {
}
public static String toString(java.util.Date date) {
if (date == null) {
return null;
}
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
return format.format(date);
}