+
+
+
includeWriteNode is false
- * @param filterWithSlaveThreshold
- *
- * @return
- */
- private ArrayList getAllActiveRWSources(
- boolean includeWriteNode,
- boolean includeCurWriteNode, boolean filterWithSlaveThreshold) {
- int curActive = activedIndex;
- ArrayList okSources = new ArrayList(
- this.allDs.size());
- for (int i = 0; i < this.writeSources.length; i++) {
- PhysicalDatasource theSource = writeSources[i];
- if (isAlive(theSource)) {// write node is active
- if (includeWriteNode) {
- if (i == curActive && includeCurWriteNode == false) {
- // not include cur active source
- } else if (filterWithSlaveThreshold) {
- if (canSelectAsReadNode(theSource)) {
- okSources.add(theSource);
- } else {
- continue;
- }
- } else {
- okSources.add(theSource);
- }
- }
- if (!readSources.isEmpty()) {
- // check all slave nodes
- PhysicalDatasource[] allSlaves = this.readSources.get(i);
- if (allSlaves != null) {
- for (PhysicalDatasource slave : allSlaves) {
- if (isAlive(slave)) {
- if (filterWithSlaveThreshold) {
- if (canSelectAsReadNode(slave)) {
- okSources.add(slave);
- } else {
- continue;
- }
- } else {
- okSources.add(slave);
- }
- }
- }
- }
- }
-
- } else {
-
- // TODO : add by zhuam
- // 如果写节点不OK, 也要保证临时的读服务正常
- if ( this.dataHostConfig.isTempReadHostAvailable() ) {
-
- if (!readSources.isEmpty()) {
- // check all slave nodes
- PhysicalDatasource[] allSlaves = this.readSources.get(i);
- if (allSlaves != null) {
- for (PhysicalDatasource slave : allSlaves) {
- if (isAlive(slave)) {
-
- if (filterWithSlaveThreshold) {
- if (canSelectAsReadNode(slave)) {
- okSources.add(slave);
- } else {
- continue;
- }
-
- } else {
- okSources.add(slave);
- }
- }
- }
- }
- }
- }
- }
-
- }
- return okSources;
- }
-
- public String[] getSchemas() {
- return schemas;
- }
-
- public void setSchemas(String[] mySchemas) {
- this.schemas = mySchemas;
- }
-
-}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/PhysicalDatasource.java b/src/main/java/io/mycat/backend/PhysicalDatasource.java
deleted file mode 100644
index 6da2469bc..000000000
--- a/src/main/java/io/mycat/backend/PhysicalDatasource.java
+++ /dev/null
@@ -1,429 +0,0 @@
-/*
- * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
- * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
- *
- * This code is free software;Designed and Developed mainly by many Chinese
- * opensource volunteers. you can redistribute it and/or modify it under the
- * terms of the GNU General Public License version 2 only, as published by the
- * Free Software Foundation.
- *
- * This code is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
- * version 2 for more details (a copy is included in the LICENSE file that
- * accompanied this code).
- *
- * You should have received a copy of the GNU General Public License version
- * 2 along with this work; if not, write to the Free Software Foundation,
- * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
- *
- * Any questions about this component can be directed to it's project Web address
- * https://code.google.com/p/opencloudb/.
- *
- */
-package io.mycat.backend;
-
-import io.mycat.backend.heartbeat.DBHeartbeat;
-import io.mycat.net.NetSystem;
-import io.mycat.server.Alarms;
-import io.mycat.server.config.node.DBHostConfig;
-import io.mycat.server.config.node.DataHostConfig;
-import io.mycat.server.executors.ConnectionHeartBeatHandler;
-import io.mycat.server.executors.DelegateResponseHandler;
-import io.mycat.server.executors.NewConnectionRespHandler;
-import io.mycat.server.executors.ResponseHandler;
-import io.mycat.util.TimeUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
-
-public abstract class PhysicalDatasource {
- public static final Logger LOGGER = LoggerFactory
- .getLogger(PhysicalDatasource.class);
-
- private final String name;
- private final int size;
- private final DBHostConfig config;
- private final ConMap conMap = new ConMap();
- private DBHeartbeat heartbeat;
- private final boolean readNode;
- private volatile long heartbeatRecoveryTime;
- private final DataHostConfig hostConfig;
- private final ConnectionHeartBeatHandler conHeartBeatHanler = new ConnectionHeartBeatHandler();
- private PhysicalDBPool dbPool;
-
- public PhysicalDatasource(DBHostConfig config, DataHostConfig hostConfig,
- boolean isReadNode) {
- this.size = config.getMaxCon();
- this.config = config;
- this.name = config.getHostName();
- this.hostConfig = hostConfig;
- heartbeat = this.createHeartBeat();
- this.readNode = isReadNode;
- }
-
- public boolean isMyConnection(BackendConnection con) {
- return (con.getPool() == this);
- }
-
- public DataHostConfig getHostConfig() {
- return hostConfig;
- }
-
- public boolean isReadNode() {
- return readNode;
- }
-
- public int getSize() {
- return size;
- }
-
- public void setDbPool(PhysicalDBPool dbPool) {
- this.dbPool = dbPool;
- }
-
- public PhysicalDBPool getDbPool() {
- return dbPool;
- }
-
- public abstract DBHeartbeat createHeartBeat();
-
- public String getName() {
- return name;
- }
- public int getIndex(){
- int currentIndex = 0;
- for(int i=0;i heartBeatCons, ConQueue queue,
- ConcurrentLinkedQueue checkLis,
- long hearBeatTime, long hearBeatTime2) {
- int maxConsInOneCheck = 10;
- Iterator checkListItor = checkLis.iterator();
- while (checkListItor.hasNext()) {
- BackendConnection con = checkListItor.next();
- if (con.isClosedOrQuit()) {
- checkListItor.remove();
- continue;
- }
- if (validSchema(con.getSchema())) {
- if (con.getLastTime() < hearBeatTime
- && heartBeatCons.size() < maxConsInOneCheck) {
- checkListItor.remove();
- // Heart beat check
- con.setBorrowed(true);
- heartBeatCons.add(con);
- }
- } else if (con.getLastTime() < hearBeatTime2) {
- // not valid schema conntion should close for idle
- // exceed 2*conHeartBeatPeriod
- checkListItor.remove();
- con.close(" heart beate idle ");
- }
-
- }
-
- }
-
- public void heatBeatCheck(long timeout, long conHeartBeatPeriod) {
- int ildeCloseCount = hostConfig.getMinCon() * 3;
- int maxConsInOneCheck = 5;
- LinkedList heartBeatCons = new LinkedList();
-
- long hearBeatTime = TimeUtil.currentTimeMillis() - conHeartBeatPeriod;
- long hearBeatTime2 = TimeUtil.currentTimeMillis() - 2
- * conHeartBeatPeriod;
- for (ConQueue queue : conMap.getAllConQueue()) {
- checkIfNeedHeartBeat(heartBeatCons, queue,
- queue.getAutoCommitCons(), hearBeatTime, hearBeatTime2);
- if (heartBeatCons.size() < maxConsInOneCheck) {
- checkIfNeedHeartBeat(heartBeatCons, queue,
- queue.getManCommitCons(), hearBeatTime, hearBeatTime2);
- } else if (heartBeatCons.size() >= maxConsInOneCheck) {
- break;
- }
- }
-
- if (!heartBeatCons.isEmpty()) {
- for (BackendConnection con : heartBeatCons) {
- conHeartBeatHanler
- .doHeartBeat(con, hostConfig.getHeartbeatSQL());
- }
- }
-
- // check if there has timeouted heatbeat cons
- conHeartBeatHanler.abandTimeOuttedConns();
- int idleCons = getIdleCount();
- int activeCons = this.getActiveCount();
- int createCount = (hostConfig.getMinCon() - idleCons) / 3;
- // create if idle too little
- if ((createCount > 0) && (idleCons + activeCons < size)
- && (idleCons < hostConfig.getMinCon())) {
- createByIdleLitte(idleCons, createCount);
- } else if (idleCons > hostConfig.getMinCon()) {
- closeByIdleMany(idleCons-hostConfig.getMinCon());
- } else {
- int activeCount = this.getActiveCount();
- if (activeCount > size) {
- StringBuilder s = new StringBuilder();
- s.append(Alarms.DEFAULT).append("DATASOURCE EXCEED [name=")
- .append(name).append(",active=");
- s.append(activeCount).append(",size=").append(size).append(']');
- LOGGER.warn(s.toString());
- }
- }
- }
-
- private void closeByIdleMany(int ildeCloseCount) {
- LOGGER.info("too many ilde cons ,close some for datasouce " + name);
- List readyCloseCons = new ArrayList(
- ildeCloseCount);
- for (ConQueue queue : conMap.getAllConQueue()) {
- readyCloseCons.addAll(queue.getIdleConsToClose(ildeCloseCount));
- if (readyCloseCons.size() >= ildeCloseCount) {
- break;
- }
- }
-
- for (BackendConnection idleCon : readyCloseCons) {
- if (idleCon.isBorrowed()) {
- LOGGER.warn("find idle con is using " + idleCon);
- }
- idleCon.close("too many idle con");
- }
- }
-
- private void createByIdleLitte(int idleCons, int createCount) {
- LOGGER.info("create connections ,because idle connection not enough ,cur is "
- + idleCons
- + ", minCon is "
- + hostConfig.getMinCon()
- + " for "
- + name);
- NewConnectionRespHandler simpleHandler = new NewConnectionRespHandler();
-
- final String[] schemas = dbPool.getSchemas();
- for (int i = 0; i < createCount; i++) {
- if (this.getActiveCount() + this.getIdleCount() >= size) {
- break;
- }
- try {
- // creat new connection
- this.createNewConnection(simpleHandler, null, schemas[i
- % schemas.length]);
- } catch (IOException e) {
- LOGGER.warn("create connection err " + e);
- }
-
- }
- }
-
- public int getActiveCount() {
- return this.conMap.getActiveCountForDs(this);
- }
-
- public void clearCons(String reason) {
- this.conMap.clearConnections(reason, this);
- }
-
- public void startHeartbeat() {
- heartbeat.start();
- }
-
- public void stopHeartbeat() {
- heartbeat.stop();
- }
-
- public void doHeartbeat() {
- // 未到预定恢复时间，不执行心跳检测。
- if (TimeUtil.currentTimeMillis() < heartbeatRecoveryTime) {
- return;
- }
- if (!heartbeat.isStop()) {
- try {
- heartbeat.heartbeat();
- } catch (Exception e) {
- LOGGER.error(name + " heartbeat error.", e);
- }
- }
- }
-
- private BackendConnection takeCon(BackendConnection conn,
- final ResponseHandler handler, final Object attachment,
- String schema) {
-
- conn.setBorrowed(true);
- if (!conn.getSchema().equals(schema)) {
- // need do schema syn in before sql send
- conn.setSchema(schema);
- }
- ConQueue queue = conMap.getSchemaConQueue(schema);
- queue.incExecuteCount();
- conn.setAttachment(attachment);
- conn.setLastTime(System.currentTimeMillis()); // 每次取连接的时候，更新下lasttime，防止在前端连接检查的时候，关闭连接，导致sql执行失败
- handler.connectionAcquired(conn);
- return conn;
- }
-
- private void createNewConnection(final ResponseHandler handler,
- final Object attachment, final String schema) throws IOException {
- // aysn create connection
- NetSystem.getInstance().getExecutor().execute(new Runnable() {
- public void run() {
- try {
- createNewConnection(new DelegateResponseHandler(handler) {
- @Override
- public void connectionError(Throwable e,
- BackendConnection conn) {
- handler.connectionError(e, conn);
- }
-
- @Override
- public void connectionAcquired(BackendConnection conn) {
- takeCon(conn, handler, attachment, schema);
- }
- }, schema);
- } catch (IOException e) {
- handler.connectionError(e, null);
- }
- }
- });
- }
-
- public void getConnection(String schema, boolean autocommit,
- final ResponseHandler handler, final Object attachment)
- throws IOException {
- BackendConnection con = this.conMap.tryTakeCon(schema, autocommit);
- if (con != null) {
- takeCon(con, handler, attachment, schema);
- return;
- } else {
- int activeCons = this.getActiveCount();//当前最大活动连接
- if(activeCons+1>size){//下一个连接大于最大连接数
- LOGGER.error("the max activeConnnections size can not be max than maxconnections");
- throw new IOException("the max activeConnnections size can not be max than maxconnections");
- }else{ // create connection
- LOGGER.info("no ilde connection in pool,create new connection for " + this.name
- + " of schema "+schema);
- createNewConnection(handler, attachment, schema);
- }
- }
-
- }
-
- private void returnCon(BackendConnection c) {
- c.setAttachment(null);
- c.setBorrowed(false);
- c.setLastTime(TimeUtil.currentTimeMillis());
- ConQueue queue = this.conMap.getSchemaConQueue(c.getSchema());
-
- boolean ok = false;
- if (c.isAutocommit()) {
- ok = queue.getAutoCommitCons().offer(c);
- } else {
- ok = queue.getManCommitCons().offer(c);
- }
- if (!ok) {
-
- LOGGER.warn("can't return to pool ,so close con " + c);
- c.close("can't return to pool ");
- }
- }
-
- public void releaseChannel(BackendConnection c) {
- returnCon(c);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("release channel " + c);
- }
- }
-
- public void connectionClosed(BackendConnection conn) {
- ConQueue queue = this.conMap.getSchemaConQueue(conn.getSchema());
- if (queue != null) {
- queue.removeCon(conn);
- }
-
- }
-
- public abstract void createNewConnection(ResponseHandler handler,
- String schema) throws IOException;
-
- public long getHeartbeatRecoveryTime() {
- return heartbeatRecoveryTime;
- }
-
- public void setHeartbeatRecoveryTime(long heartbeatRecoveryTime) {
- this.heartbeatRecoveryTime = heartbeatRecoveryTime;
- }
-
- public DBHostConfig getConfig() {
- return config;
- }
-}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/datasource/PhysicalDBNode.java b/src/main/java/io/mycat/backend/datasource/PhysicalDBNode.java
new file mode 100644
index 000000000..71deb7ed2
--- /dev/null
+++ b/src/main/java/io/mycat/backend/datasource/PhysicalDBNode.java
@@ -0,0 +1,160 @@
+/*
+ * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software;Designed and Developed mainly by many Chinese
+ * opensource volunteers. you can redistribute it and/or modify it under the
+ * terms of the GNU General Public License version 2 only, as published by the
+ * Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Any questions about this component can be directed to it's project Web address
+ * https://code.google.com/p/opencloudb/.
+ *
+ */
+package io.mycat.backend.datasource;
+
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.backend.BackendConnection;
+import io.mycat.backend.mysql.nio.handler.ResponseHandler;
+import io.mycat.route.RouteResultsetNode;
+
+public class PhysicalDBNode {
+ protected static final Logger LOGGER = LoggerFactory
+ .getLogger(PhysicalDBNode.class);
+
+ protected final String name;
+ protected final String database;
+ protected final PhysicalDBPool dbPool;
+
+ public PhysicalDBNode(String hostName, String database,
+ PhysicalDBPool dbPool) {
+ this.name = hostName;
+ this.database = database;
+ this.dbPool = dbPool;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public PhysicalDBPool getDbPool() {
+ return dbPool;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ /**
+ * get connection from the same datasource
+ *
+ * @param exitsCon
+ * @throws Exception
+ */
+ public void getConnectionFromSameSource(String schema,boolean autocommit,
+ BackendConnection exitsCon, ResponseHandler handler,
+ Object attachment) throws Exception {
+
+ PhysicalDatasource ds = this.dbPool.findDatasouce(exitsCon);
+ if (ds == null) {
+ throw new RuntimeException(
+ "can't find exits connection,maybe fininshed " + exitsCon);
+ } else {
+ ds.getConnection(schema,autocommit, handler, attachment);
+ }
+
+ }
+
+ private void checkRequest(String schema){
+ if (schema != null
+ && !schema.equals(this.database)) {
+ throw new RuntimeException(
+ "invalid param ,connection request db is :"
+ + schema + " and datanode db is "
+ + this.database);
+ }
+ if (!dbPool.isInitSuccess()) {
+ dbPool.init(dbPool.activedIndex);
+ }
+ }
+
+ public void getConnection(String schema,boolean autoCommit, RouteResultsetNode rrs,
+ ResponseHandler handler, Object attachment) throws Exception {
+ checkRequest(schema);
+ if (dbPool.isInitSuccess()) {
+ LOGGER.debug("rrs.getRunOnSlave() " + rrs.getRunOnSlave());
+ if(rrs.getRunOnSlave() != null){ // 带有 /*db_type=master/slave*/ 注解
+ // 强制走 slave
+ if(rrs.getRunOnSlave()){
+ LOGGER.debug("rrs.isHasBlanceFlag() " + rrs.isHasBlanceFlag());
+ if (rrs.isHasBlanceFlag()) { // 带有 /*balance*/ 注解(目前好像只支持一个注解...)
+ dbPool.getReadBanlanceCon(schema,autoCommit,handler, attachment, this.database);
+ }else{ // 没有 /*balance*/ 注解
+ LOGGER.debug("rrs.isHasBlanceFlag()" + rrs.isHasBlanceFlag());
+ if(!dbPool.getReadCon(schema, autoCommit, handler, attachment, this.database)){
+ LOGGER.warn("Do not have slave connection to use, use master connection instead.");
+ PhysicalDatasource writeSource=dbPool.getSource();
+ //记录写节点写负载值
+ writeSource.setWriteCount();
+ writeSource.getConnection(schema,
+ autoCommit, handler, attachment);
+ rrs.setRunOnSlave(false);
+ rrs.setCanRunInReadDB(false);
+ }
+ }
+ }else{ // 强制走 master
+ // 默认获得的是 writeSource，也就是 走master
+ LOGGER.debug("rrs.getRunOnSlave() " + rrs.getRunOnSlave());
+ PhysicalDatasource writeSource=dbPool.getSource();
+ //记录写节点写负载值
+ writeSource.setReadCount();
+ writeSource.getConnection(schema, autoCommit,
+ handler, attachment);
+ rrs.setCanRunInReadDB(false);
+ }
+ }else{ // 没有 /*db_type=master/slave*/ 注解，按照原来的处理方式
+ LOGGER.debug("rrs.getRunOnSlave() " + rrs.getRunOnSlave()); // null
+ if (rrs.canRunnINReadDB(autoCommit)) {
+ dbPool.getRWBanlanceCon(schema,autoCommit, handler, attachment, this.database);
+ } else {
+ PhysicalDatasource writeSource =dbPool.getSource();
+ //记录写节点写负载值
+ writeSource.setWriteCount();
+ writeSource.getConnection(schema, autoCommit,
+ handler, attachment);
+ }
+ }
+
+ } else {
+ throw new IllegalArgumentException("Invalid DataSource:" + dbPool.getActivedIndex());
+ }
+ }
+
+// public void getConnection(String schema,boolean autoCommit, RouteResultsetNode rrs,
+// ResponseHandler handler, Object attachment) throws Exception {
+// checkRequest(schema);
+// if (dbPool.isInitSuccess()) {
+// if (rrs.canRunnINReadDB(autoCommit)) {
+// dbPool.getRWBanlanceCon(schema,autoCommit, handler, attachment,
+// this.database);
+// } else {
+// dbPool.getSource().getConnection(schema,autoCommit, handler, attachment);
+// }
+//
+// } else {
+// throw new IllegalArgumentException("Invalid DataSource:"
+// + dbPool.getActivedIndex());
+// }
+// }
+}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/datasource/PhysicalDBPool.java b/src/main/java/io/mycat/backend/datasource/PhysicalDBPool.java
new file mode 100644
index 000000000..56968f2c8
--- /dev/null
+++ b/src/main/java/io/mycat/backend/datasource/PhysicalDBPool.java
@@ -0,0 +1,721 @@
+/*
+ * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software;Designed and Developed mainly by many Chinese
+ * opensource volunteers. you can redistribute it and/or modify it under the
+ * terms of the GNU General Public License version 2 only, as published by the
+ * Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Any questions about this component can be directed to it's project Web address
+ * https://code.google.com/p/opencloudb/.
+ *
+ */
+package io.mycat.backend.datasource;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.MycatServer;
+import io.mycat.backend.BackendConnection;
+import io.mycat.backend.heartbeat.DBHeartbeat;
+import io.mycat.backend.mysql.nio.handler.GetConnectionHandler;
+import io.mycat.backend.mysql.nio.handler.ResponseHandler;
+import io.mycat.config.Alarms;
+import io.mycat.config.model.DataHostConfig;
+
+public class PhysicalDBPool {
+
+ protected static final Logger LOGGER = LoggerFactory.getLogger(PhysicalDBPool.class);
+
+ public static final int BALANCE_NONE = 0;
+ public static final int BALANCE_ALL_BACK = 1;
+ public static final int BALANCE_ALL = 2;
+ public static final int BALANCE_ALL_READ = 3;
+
+ public static final int WRITE_ONLYONE_NODE = 0;
+ public static final int WRITE_RANDOM_NODE = 1;
+ public static final int WRITE_ALL_NODE = 2;
+
+ public static final long LONG_TIME = 300000;
+ public static final int WEIGHT = 0;
+
+ private final String hostName;
+
+ protected PhysicalDatasource[] writeSources;
+ protected Map readSources;
+
+ protected volatile int activedIndex;
+ protected volatile boolean initSuccess;
+
+ protected final ReentrantLock switchLock = new ReentrantLock();
+ private final Collection allDs;
+ private final int banlance;
+ private final int writeType;
+ private final Random random = new Random();
+ private final Random wnrandom = new Random();
+ private String[] schemas;
+ private final DataHostConfig dataHostConfig;
+ private String slaveIDs;
+
+ public PhysicalDBPool(String name, DataHostConfig conf,
+ PhysicalDatasource[] writeSources,
+ Map readSources, int balance,
+ int writeType) {
+
+ this.hostName = name;
+ this.dataHostConfig = conf;
+ this.writeSources = writeSources;
+ this.banlance = balance;
+ this.writeType = writeType;
+
+ Iterator> entryItor = readSources.entrySet().iterator();
+ while (entryItor.hasNext()) {
+ PhysicalDatasource[] values = entryItor.next().getValue();
+ if (values.length == 0) {
+ entryItor.remove();
+ }
+ }
+
+ this.readSources = readSources;
+ this.allDs = this.genAllDataSources();
+
+ LOGGER.info("total resouces of dataHost " + this.hostName + " is :" + allDs.size());
+
+ setDataSourceProps();
+ }
+
+ public int getWriteType() {
+ return writeType;
+ }
+
+ private void setDataSourceProps() {
+ for (PhysicalDatasource ds : this.allDs) {
+ ds.setDbPool(this);
+ }
+ }
+
+ public PhysicalDatasource findDatasouce(BackendConnection exitsCon) {
+ for (PhysicalDatasource ds : this.allDs) {
+ if ((ds.isReadNode() == exitsCon.isFromSlaveDB())
+ && ds.isMyConnection(exitsCon)) {
+ return ds;
+ }
+ }
+
+ LOGGER.warn("can't find connection in pool " + this.hostName + " con:" + exitsCon);
+ return null;
+ }
+
+ public String getSlaveIDs() {
+ return slaveIDs;
+ }
+
+ public void setSlaveIDs(String slaveIDs) {
+ this.slaveIDs = slaveIDs;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ /**
+ * all write datanodes
+ * @return
+ */
+ public PhysicalDatasource[] getSources() {
+ return writeSources;
+ }
+
+ public PhysicalDatasource getSource() {
+
+ switch (writeType) {
+ case WRITE_ONLYONE_NODE: {
+ return writeSources[activedIndex];
+ }
+ case WRITE_RANDOM_NODE: {
+
+ int index = Math.abs(wnrandom.nextInt(Integer.MAX_VALUE)) % writeSources.length;
+ PhysicalDatasource result = writeSources[index];
+ if (!this.isAlive(result)) {
+
+ // find all live nodes
+ ArrayList alives = new ArrayList(writeSources.length - 1);
+ for (int i = 0; i < writeSources.length; i++) {
+ if (i != index
+ && this.isAlive(writeSources[i])) {
+ alives.add(i);
+ }
+ }
+
+ if (alives.isEmpty()) {
+ result = writeSources[0];
+ } else {
+ // random select one
+ index = Math.abs(wnrandom.nextInt(Integer.MAX_VALUE)) % alives.size();
+ result = writeSources[alives.get(index)];
+
+ }
+ }
+
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("select write source " + result.getName()
+ + " for dataHost:" + this.getHostName());
+ }
+ return result;
+ }
+ default: {
+ throw new java.lang.IllegalArgumentException("writeType is "
+ + writeType + " ,so can't return one write datasource ");
+ }
+ }
+
+ }
+
+ public int getActivedIndex() {
+ return activedIndex;
+ }
+
+ public boolean isInitSuccess() {
+ return initSuccess;
+ }
+
+ public int next(int i) {
+ if (checkIndex(i)) {
+ return (++i == writeSources.length) ? 0 : i;
+ } else {
+ return 0;
+ }
+ }
+
+ public boolean switchSource(int newIndex, boolean isAlarm, String reason) {
+ if (this.writeType != PhysicalDBPool.WRITE_ONLYONE_NODE || !checkIndex(newIndex)) {
+ return false;
+ }
+
+ final ReentrantLock lock = this.switchLock;
+ lock.lock();
+ try {
+ int current = activedIndex;
+ if (current != newIndex) {
+
+ // switch index
+ activedIndex = newIndex;
+
+ // init again
+ this.init(activedIndex);
+
+ // clear all connections
+ this.getSources()[current].clearCons("switch datasource");
+
+ // write log
+ LOGGER.warn(switchMessage(current, newIndex, false, reason));
+
+ return true;
+ }
+ } finally {
+ lock.unlock();
+ }
+ return false;
+ }
+
+ private String switchMessage(int current, int newIndex, boolean alarm, String reason) {
+ StringBuilder s = new StringBuilder();
+ if (alarm) {
+ s.append(Alarms.DATANODE_SWITCH);
+ }
+ s.append("[Host=").append(hostName).append(",result=[").append(current).append("->");
+ s.append(newIndex).append("],reason=").append(reason).append(']');
+ return s.toString();
+ }
+
+ private int loop(int i) {
+ return i < writeSources.length ? i : (i - writeSources.length);
+ }
+
+ public void init(int index) {
+
+ if (!checkIndex(index)) {
+ index = 0;
+ }
+
+ int active = -1;
+ for (int i = 0; i < writeSources.length; i++) {
+ int j = loop(i + index);
+ if ( initSource(j, writeSources[j]) ) {
+
+ //不切换-1时，如果主写挂了 不允许切换过去
+ boolean isNotSwitchDs = ( dataHostConfig.getSwitchType() == DataHostConfig.NOT_SWITCH_DS );
+ if ( isNotSwitchDs && j > 0 ) {
+ break;
+ }
+
+ active = j;
+ activedIndex = active;
+ initSuccess = true;
+ LOGGER.info(getMessage(active, " init success"));
+
+ if (this.writeType == WRITE_ONLYONE_NODE) {
+ // only init one write datasource
+ MycatServer.getInstance().saveDataHostIndex(hostName, activedIndex);
+ break;
+ }
+ }
+ }
+
+ if (!checkIndex(active)) {
+ initSuccess = false;
+ StringBuilder s = new StringBuilder();
+ s.append(Alarms.DEFAULT).append(hostName).append(" init failure");
+ LOGGER.error(s.toString());
+ }
+ }
+
+ private boolean checkIndex(int i) {
+ return i >= 0 && i < writeSources.length;
+ }
+
+ private String getMessage(int index, String info) {
+ return new StringBuilder().append(hostName).append(" index:").append(index).append(info).toString();
+ }
+
+ private boolean initSource(int index, PhysicalDatasource ds) {
+ int initSize = ds.getConfig().getMinCon();
+
+ LOGGER.info("init backend myqsl source ,create connections total " + initSize + " for " + ds.getName() + " index :" + index);
+
+ CopyOnWriteArrayList list = new CopyOnWriteArrayList();
+ GetConnectionHandler getConHandler = new GetConnectionHandler(list, initSize);
+ // long start = System.currentTimeMillis();
+ // long timeOut = start + 5000 * 1000L;
+
+ for (int i = 0; i < initSize; i++) {
+ try {
+ ds.getConnection(this.schemas[i % schemas.length], true, getConHandler, null);
+ } catch (Exception e) {
+ LOGGER.warn(getMessage(index, " init connection error."), e);
+ }
+ }
+ long timeOut = System.currentTimeMillis() + 60 * 1000;
+
+ // waiting for finish
+ while (!getConHandler.finished() && (System.currentTimeMillis() < timeOut)) {
+ try {
+ Thread.sleep(100);
+
+ } catch (InterruptedException e) {
+ LOGGER.error("initError", e);
+ }
+ }
+ LOGGER.info("init result :" + getConHandler.getStatusInfo());
+// for (BackendConnection c : list) {
+// c.release();
+// }
+ return !list.isEmpty();
+ }
+
+ public void doHeartbeat() {
+
+
+ if (writeSources == null || writeSources.length == 0) {
+ return;
+ }
+
+ for (PhysicalDatasource source : this.allDs) {
+
+ if (source != null) {
+ source.doHeartbeat();
+ } else {
+ StringBuilder s = new StringBuilder();
+ s.append(Alarms.DEFAULT).append(hostName).append(" current dataSource is null!");
+ LOGGER.error(s.toString());
+ }
+ }
+
+ }
+
+ /**
+ * back physical connection heartbeat check
+ */
+ public void heartbeatCheck(long ildCheckPeriod) {
+
+ for (PhysicalDatasource ds : allDs) {
+ // only readnode or all write node or writetype=WRITE_ONLYONE_NODE
+ // and current write node will check
+ if (ds != null
+ && (ds.getHeartbeat().getStatus() == DBHeartbeat.OK_STATUS)
+ && (ds.isReadNode()
+ || (this.writeType != WRITE_ONLYONE_NODE)
+ || (this.writeType == WRITE_ONLYONE_NODE
+ && ds == this.getSource()))) {
+
+ ds.heatBeatCheck(ds.getConfig().getIdleTimeout(), ildCheckPeriod);
+ }
+ }
+ }
+
+ public void startHeartbeat() {
+ for (PhysicalDatasource source : this.allDs) {
+ source.startHeartbeat();
+ }
+ }
+
+ public void stopHeartbeat() {
+ for (PhysicalDatasource source : this.allDs) {
+ source.stopHeartbeat();
+ }
+ }
+
+ /**
+ * 强制清除 dataSources
+ * @param reason
+ */
+ public void clearDataSources(String reason) {
+ LOGGER.info("clear datasours of pool " + this.hostName);
+ for (PhysicalDatasource source : this.allDs) {
+ LOGGER.info("clear datasoure of pool " + this.hostName + " ds:" + source.getConfig());
+ source.clearCons(reason);
+ source.stopHeartbeat();
+ }
+ }
+
+ public Collection genAllDataSources() {
+
+ LinkedList allSources = new LinkedList();
+ for (PhysicalDatasource ds : writeSources) {
+ if (ds != null) {
+ allSources.add(ds);
+ }
+ }
+
+ for (PhysicalDatasource[] dataSources : this.readSources.values()) {
+ for (PhysicalDatasource ds : dataSources) {
+ if (ds != null) {
+ allSources.add(ds);
+ }
+ }
+ }
+ return allSources;
+ }
+
+ public Collection getAllDataSources() {
+ return this.allDs;
+ }
+
+ /**
+ * return connection for read balance
+ *
+ * @param handler
+ * @param attachment
+ * @param database
+ * @throws Exception
+ */
+ public void getRWBanlanceCon(String schema, boolean autocommit,
+ ResponseHandler handler, Object attachment, String database) throws Exception {
+
+ PhysicalDatasource theNode = null;
+ ArrayList okSources = null;
+ switch (banlance) {
+ case BALANCE_ALL_BACK: {
+ // all read nodes and the standard by masters
+ okSources = getAllActiveRWSources(true, false, checkSlaveSynStatus());
+ if (okSources.isEmpty()) {
+ theNode = this.getSource();
+
+ } else {
+ theNode = randomSelect(okSources);
+ }
+ break;
+ }
+ case BALANCE_ALL: {
+ okSources = getAllActiveRWSources(true, true, checkSlaveSynStatus());
+ theNode = randomSelect(okSources);
+ break;
+ }
+ case BALANCE_ALL_READ: {
+ okSources = getAllActiveRWSources(false, false, checkSlaveSynStatus());
+ theNode = randomSelect(okSources);
+ break;
+ }
+ case BALANCE_NONE:
+ default:
+ // return default write data source
+ theNode = this.getSource();
+ }
+
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("select read source " + theNode.getName() + " for dataHost:" + this.getHostName());
+ }
+ //统计节点读操作次数
+ theNode.setReadCount();
+ theNode.getConnection(schema, autocommit, handler, attachment);
+ }
+
+ /**
+ * slave 读负载均衡，也就是 readSource 之间实现负载均衡
+ * @param schema
+ * @param autocommit
+ * @param handler
+ * @param attachment
+ * @param database
+ * @throws Exception
+ */
+ public void getReadBanlanceCon(String schema, boolean autocommit, ResponseHandler handler,
+ Object attachment, String database)throws Exception {
+ PhysicalDatasource theNode = null;
+ ArrayList okSources = null;
+ okSources = getAllActiveRWSources(false, false, checkSlaveSynStatus());
+ theNode = randomSelect(okSources);
+ //统计节点读操作次数
+ theNode.setReadCount();
+ theNode.getConnection(schema, autocommit, handler, attachment);
+ }
+
+ /**
+ * 从 writeHost 下面的 readHost中随机获取一个 connection, 用于slave注解
+ * @param schema
+ * @param autocommit
+ * @param handler
+ * @param attachment
+ * @param database
+ * @return
+ * @throws Exception
+ */
+ public boolean getReadCon(String schema, boolean autocommit, ResponseHandler handler,
+ Object attachment, String database)throws Exception {
+ PhysicalDatasource theNode = null;
+
+ LOGGER.debug("!readSources.isEmpty() " + !readSources.isEmpty());
+ if (!readSources.isEmpty()) {
+ int index = Math.abs(random.nextInt(Integer.MAX_VALUE)) % readSources.size();
+ PhysicalDatasource[] allSlaves = this.readSources.get(index);
+// System.out.println("allSlaves.length " + allSlaves.length);
+ if (allSlaves != null) {
+ index = Math.abs(random.nextInt(Integer.MAX_VALUE)) % readSources.size();
+ PhysicalDatasource slave = allSlaves[index];
+
+ for (int i=0; i okSources) {
+
+ if (okSources.isEmpty()) {
+ return this.getSource();
+
+ } else {
+
+ int length = okSources.size(); // 总个数
+ int totalWeight = 0; // 总权重
+ boolean sameWeight = true; // 权重是否都一样
+ for (int i = 0; i < length; i++) {
+ int weight = okSources.get(i).getConfig().getWeight();
+ totalWeight += weight; // 累计总权重
+ if (sameWeight && i > 0
+ && weight != okSources.get(i-1).getConfig().getWeight() ) { // 计算所有权重是否一样
+ sameWeight = false;
+ }
+ }
+
+ if (totalWeight > 0 && !sameWeight ) {
+
+ // 如果权重不相同且权重大于0则按总权重数随机
+ int offset = random.nextInt(totalWeight);
+
+ // 并确定随机值落在哪个片断上
+ for (int i = 0; i < length; i++) {
+ offset -= okSources.get(i).getConfig().getWeight();
+ if (offset < 0) {
+ return okSources.get(i);
+ }
+ }
+ }
+
+ // 如果权重相同或权重为0则均等随机
+ return okSources.get( random.nextInt(length) );
+
+ //int index = Math.abs(random.nextInt()) % okSources.size();
+ //return okSources.get(index);
+ }
+ }
+
+ //
+ public int getBalance() {
+ return banlance;
+ }
+
+ private boolean isAlive(PhysicalDatasource theSource) {
+ return (theSource.getHeartbeat().getStatus() == DBHeartbeat.OK_STATUS);
+ }
+
+ private boolean canSelectAsReadNode(PhysicalDatasource theSource) {
+
+ Integer slaveBehindMaster = theSource.getHeartbeat().getSlaveBehindMaster();
+ int dbSynStatus = theSource.getHeartbeat().getDbSynStatus();
+
+ if ( slaveBehindMaster == null || dbSynStatus == DBHeartbeat.DB_SYN_ERROR) {
+ return false;
+ }
+ boolean isSync = dbSynStatus == DBHeartbeat.DB_SYN_NORMAL;
+ boolean isNotDelay = slaveBehindMaster < this.dataHostConfig.getSlaveThreshold();
+ return isSync && isNotDelay;
+ }
+
+ /**
+ * return all backup write sources
+ *
+ * @param includeWriteNode if include write nodes
+ * @param includeCurWriteNode if include current active write node. invalid when includeWriteNode is false
+ * @param filterWithSlaveThreshold
+ *
+ * @return
+ */
+ private ArrayList getAllActiveRWSources(
+ boolean includeWriteNode, boolean includeCurWriteNode, boolean filterWithSlaveThreshold) {
+
+ int curActive = activedIndex;
+ ArrayList okSources = new ArrayList(this.allDs.size());
+
+ for (int i = 0; i < this.writeSources.length; i++) {
+ PhysicalDatasource theSource = writeSources[i];
+ if (isAlive(theSource)) {// write node is active
+
+ if (includeWriteNode) {
+ boolean isCurWriteNode = ( i == curActive );
+ if ( isCurWriteNode && includeCurWriteNode == false) {
+ // not include cur active source
+ } else if (filterWithSlaveThreshold && theSource.isSalveOrRead() ) {
+ boolean selected = canSelectAsReadNode(theSource);
+ if ( selected ) {
+ okSources.add(theSource);
+ } else {
+ continue;
+ }
+ } else {
+ okSources.add(theSource);
+ }
+ }
+
+ if (!readSources.isEmpty()) {
+ // check all slave nodes
+ PhysicalDatasource[] allSlaves = this.readSources.get(i);
+ if (allSlaves != null) {
+ for (PhysicalDatasource slave : allSlaves) {
+ if (isAlive(slave)) {
+ if (filterWithSlaveThreshold) {
+ boolean selected = canSelectAsReadNode(slave);
+ if ( selected ) {
+ okSources.add(slave);
+ } else {
+ continue;
+ }
+ } else {
+ okSources.add(slave);
+ }
+ }
+ }
+ }
+ }
+
+ } else {
+
+ // TODO : add by zhuam
+ // 如果写节点不OK, 也要保证临时的读服务正常
+ if ( this.dataHostConfig.isTempReadHostAvailable()
+ && !readSources.isEmpty()) {
+
+ // check all slave nodes
+ PhysicalDatasource[] allSlaves = this.readSources.get(i);
+ if (allSlaves != null) {
+ for (PhysicalDatasource slave : allSlaves) {
+ if (isAlive(slave)) {
+
+ if (filterWithSlaveThreshold) {
+ if (canSelectAsReadNode(slave)) {
+ okSources.add(slave);
+ } else {
+ continue;
+ }
+
+ } else {
+ okSources.add(slave);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ }
+ return okSources;
+ }
+
+ public String[] getSchemas() {
+ return schemas;
+ }
+
+ public void setSchemas(String[] mySchemas) {
+ this.schemas = mySchemas;
+ }
+}
diff --git a/src/main/java/io/mycat/backend/datasource/PhysicalDatasource.java b/src/main/java/io/mycat/backend/datasource/PhysicalDatasource.java
new file mode 100644
index 000000000..6851f0900
--- /dev/null
+++ b/src/main/java/io/mycat/backend/datasource/PhysicalDatasource.java
@@ -0,0 +1,635 @@
+/*
+ * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software;Designed and Developed mainly by many Chinese
+ * opensource volunteers. you can redistribute it and/or modify it under the
+ * terms of the GNU General Public License version 2 only, as published by the
+ * Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Any questions about this component can be directed to it's project Web address
+ * https://code.google.com/p/opencloudb/.
+ *
+ */
+package io.mycat.backend.datasource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.mycat.MycatServer;
+import io.mycat.backend.BackendConnection;
+import io.mycat.backend.ConMap;
+import io.mycat.backend.ConQueue;
+import io.mycat.backend.heartbeat.DBHeartbeat;
+import io.mycat.backend.mysql.nio.MySQLConnection;
+import io.mycat.backend.mysql.nio.handler.ConnectionHeartBeatHandler;
+import io.mycat.backend.mysql.nio.handler.DelegateResponseHandler;
+import io.mycat.backend.mysql.nio.handler.NewConnectionRespHandler;
+import io.mycat.backend.mysql.nio.handler.ResponseHandler;
+import io.mycat.config.Alarms;
+import io.mycat.config.model.DBHostConfig;
+import io.mycat.config.model.DataHostConfig;
+import io.mycat.util.TimeUtil;
+
+
+public abstract class PhysicalDatasource {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PhysicalDatasource.class);
+
+ private final String name;
+ private final int size;
+ private final DBHostConfig config;
+ private final ConMap conMap = new ConMap();
+ private DBHeartbeat heartbeat;
+ private final boolean readNode;
+ private volatile long heartbeatRecoveryTime;
+ private final DataHostConfig hostConfig;
+ private final ConnectionHeartBeatHandler conHeartBeatHanler = new ConnectionHeartBeatHandler();
+ private PhysicalDBPool dbPool;
+
+ // 添加DataSource读计数
+ private AtomicLong readCount = new AtomicLong(0);
+
+ // 添加DataSource写计数
+ private AtomicLong writeCount = new AtomicLong(0);
+
+
+ /**
+ * edit by dingw at 2017.06.08
+ * @see https://github.com/MyCATApache/Mycat-Server/issues/1524
+ *
+ */
+ // 当前活动连接
+ //private volatile AtomicInteger activeCount = new AtomicInteger(0);
+
+ // 当前存活的总连接数,为什么不直接使用activeCount,主要是因为连接的创建是异步完成的
+ //private volatile AtomicInteger totalConnection = new AtomicInteger(0);
+
+ /**
+ * 由于在Mycat中，returnCon被多次调用（与takeCon并没有成对调用）导致activeCount、totalConnection容易出现负数
+ */
+ //private static final String TAKE_CONNECTION_FLAG = "1";
+ //private ConcurrentMap heartBeatCons, ConQueue queue,
+ ConcurrentLinkedQueue checkLis,
+ long hearBeatTime, long hearBeatTime2) {
+ int maxConsInOneCheck = 10;
+ Iterator checkListItor = checkLis.iterator();
+ while (checkListItor.hasNext()) {
+ BackendConnection con = checkListItor.next();
+ if (con.isClosedOrQuit()) {
+ checkListItor.remove();
+ continue;
+ }
+ if (validSchema(con.getSchema())) {
+ if (con.getLastTime() < hearBeatTime
+ && heartBeatCons.size() < maxConsInOneCheck) {
+ if(checkLis.remove(con)) {
+ //如果移除成功，则放入到心跳连接中，如果移除失败，说明该连接已经被其他线程使用，忽略本次心跳检测
+ con.setBorrowed(true);
+ heartBeatCons.add(con);
+ }
+ }
+ } else if (con.getLastTime() < hearBeatTime2) {
+ // not valid schema conntion should close for idle
+ // exceed 2*conHeartBeatPeriod
+ // 同样，这里也需要先移除，避免被业务连接
+ if(checkLis.remove(con)) {
+ con.close(" heart beate idle ");
+ }
+ }
+
+ }
+
+ }
+
+ public int getIndex() {
+ int currentIndex = 0;
+ for (int i = 0; i < dbPool.getSources().length; i++) {
+ PhysicalDatasource writeHostDatasource = dbPool.getSources()[i];
+ if (writeHostDatasource.getName().equals(getName())) {
+ currentIndex = i;
+ break;
+ }
+ }
+ return currentIndex;
+ }
+
+ public boolean isSalveOrRead() {
+ int currentIndex = getIndex();
+ if (currentIndex != dbPool.activedIndex || this.readNode) {
+ return true;
+ }
+ return false;
+ }
+
+ public void heatBeatCheck(long timeout, long conHeartBeatPeriod) {
+// int ildeCloseCount = hostConfig.getMinCon() * 3;
+ int maxConsInOneCheck = 5;
+ LinkedList heartBeatCons = new LinkedList();
+
+ long hearBeatTime = TimeUtil.currentTimeMillis() - conHeartBeatPeriod;
+ long hearBeatTime2 = TimeUtil.currentTimeMillis() - 2
+ * conHeartBeatPeriod;
+ for (ConQueue queue : conMap.getAllConQueue()) {
+ checkIfNeedHeartBeat(heartBeatCons, queue,
+ queue.getAutoCommitCons(), hearBeatTime, hearBeatTime2);
+ if (heartBeatCons.size() < maxConsInOneCheck) {
+ checkIfNeedHeartBeat(heartBeatCons, queue,
+ queue.getManCommitCons(), hearBeatTime, hearBeatTime2);
+ } else if (heartBeatCons.size() >= maxConsInOneCheck) {
+ break;
+ }
+ }
+
+ if (!heartBeatCons.isEmpty()) {
+ for (BackendConnection con : heartBeatCons) {
+ conHeartBeatHanler
+ .doHeartBeat(con, hostConfig.getHearbeatSQL());
+ }
+ }
+
+ // check if there has timeouted heatbeat cons
+ conHeartBeatHanler.abandTimeOuttedConns();
+ int idleCons = getIdleCount();
+ int activeCons = this.getActiveCount();
+ int createCount = (hostConfig.getMinCon() - idleCons) / 3;
+ // create if idle too little
+ if ((createCount > 0) && (idleCons + activeCons < size)
+ && (idleCons < hostConfig.getMinCon())) {
+ createByIdleLitte(idleCons, createCount);
+ } else if (idleCons > hostConfig.getMinCon()) {
+ closeByIdleMany(idleCons - hostConfig.getMinCon());
+ } else {
+ int activeCount = this.getActiveCount();
+ if (activeCount > size) {
+ StringBuilder s = new StringBuilder();
+ s.append(Alarms.DEFAULT).append("DATASOURCE EXCEED [name=")
+ .append(name).append(",active=");
+ s.append(activeCount).append(",size=").append(size).append(']');
+ LOGGER.warn(s.toString());
+ }
+ }
+ }
+
+ /**
+ *
+ * @param ildeCloseCount
+ * 首先，从已创建的连接中选择本次心跳需要关闭的空闲连接数（由当前连接连接数-减去配置的最小连接数。
+ * 然后依次关闭这些连接。由于连接空闲心跳检测与业务是同时并发的，在心跳关闭阶段，可能有连接被使用，导致需要关闭的空闲连接数减少.
+ *
+ * 所以每次关闭新连接时，先判断当前空闲连接数是否大于配置的最少空闲连接，如果为否，则结束本次关闭空闲连接操作。
+ * 该方法修改之前：
+ * 首先从ConnMap中获取 ildeCloseCount 个连接，然后关闭；在关闭中，可能又有连接被使用，导致可能多关闭一些链接，
+ * 导致相对频繁的创建新连接和关闭连接
+ *
+ * 该方法修改之后：
+ * ildeCloseCount 为预期要关闭的连接
+ * 使用循环操作，首先在关闭之前，先再一次判断是否需要关闭连接，然后每次从ConnMap中获取一个空闲连接，然后进行关闭
+ * edit by dingw at 2017.06.16
+ */
+ private void closeByIdleMany(int ildeCloseCount) {
+ LOGGER.info("too many ilde cons ,close some for datasouce " + name);
+ List readyCloseCons = new ArrayList(
+ ildeCloseCount);
+ for (ConQueue queue : conMap.getAllConQueue()) {
+ readyCloseCons.addAll(queue.getIdleConsToClose(ildeCloseCount));
+ if (readyCloseCons.size() >= ildeCloseCount) {
+ break;
+ }
+ }
+
+ for (BackendConnection idleCon : readyCloseCons) {
+ if (idleCon.isBorrowed()) {
+ LOGGER.warn("find idle con is using " + idleCon);
+ }
+ idleCon.close("too many idle con");
+ }
+
+// LOGGER.info("too many ilde cons ,close some for datasouce " + name);
+//
+// Iterator conQueueIt = conMap.getAllConQueue().iterator();
+// ConQueue queue = null;
+// if(conQueueIt.hasNext()) {
+// queue = conQueueIt.next();
+// }
+//
+// for(int i = 0; i < ildeCloseCount; i ++ ) {
+//
+// if(!needCloseIdleConnection() || queue == null) {
+// break; //如果当时空闲连接数没有超过最小配置连接数，则结束本次连接关闭
+// }
+//
+// LOGGER.info("cur conns:" + getTotalConnectionsSafe() );
+//
+// BackendConnection idleCon = queue.takeIdleCon(false);
+//
+// while(idleCon == null && conQueueIt.hasNext()) {
+// queue = conQueueIt.next();
+// idleCon = queue.takeIdleCon(false);
+// }
+//
+// if(idleCon == null) {
+// break;
+// }
+//
+// if (idleCon.isBorrowed() ) {
+// LOGGER.warn("find idle con is using " + idleCon);
+// }
+// idleCon.close("too many idle con");
+//
+// }
+
+ }
+
+ private void createByIdleLitte(int idleCons, int createCount) {
+ LOGGER.info("create connections ,because idle connection not enough ,cur is "
+ + idleCons
+ + ", minCon is "
+ + hostConfig.getMinCon()
+ + " for "
+ + name);
+ NewConnectionRespHandler simpleHandler = new NewConnectionRespHandler();
+
+ final String[] schemas = dbPool.getSchemas();
+ for (int i = 0; i < createCount; i++) {
+ if (this.getActiveCount() + this.getIdleCount() >= size) {
+ break;
+ }
+ try {
+ // creat new connection
+ this.createNewConnection(simpleHandler, null, schemas[i
+ % schemas.length]);
+ } catch (IOException e) {
+ LOGGER.warn("create connection err " + e);
+ }
+
+ }
+ }
+
+ public int getActiveCount() {
+ return this.conMap.getActiveCountForDs(this);
+ }
+
+
+
+ public void clearCons(String reason) {
+ this.conMap.clearConnections(reason, this);
+ }
+
+ public void startHeartbeat() {
+ heartbeat.start();
+ }
+
+ public void stopHeartbeat() {
+ heartbeat.stop();
+ }
+
+ public void doHeartbeat() {
+ // 未到预定恢复时间，不执行心跳检测。
+ if (TimeUtil.currentTimeMillis() < heartbeatRecoveryTime) {
+ return;
+ }
+
+ if (!heartbeat.isStop()) {
+ try {
+ heartbeat.heartbeat();
+ } catch (Exception e) {
+ LOGGER.error(name + " heartbeat error.", e);
+ }
+ }
+ }
+
+ private BackendConnection takeCon(BackendConnection conn,
+ final ResponseHandler handler, final Object attachment,
+ String schema) {
+
+ conn.setBorrowed(true);
+
+// if(takeConnectionContext.putIfAbsent(conn.getId(), TAKE_CONNECTION_FLAG) == null) {
+// incrementActiveCountSafe();
+// }
+
+
+ if (!conn.getSchema().equals(schema)) {
+ // need do schema syn in before sql send
+ conn.setSchema(schema);
+ }
+ ConQueue queue = conMap.getSchemaConQueue(schema);
+ queue.incExecuteCount();
+ conn.setAttachment(attachment);
+ conn.setLastTime(System.currentTimeMillis()); // 每次取连接的时候，更新下lasttime，防止在前端连接检查的时候，关闭连接，导致sql执行失败
+ handler.connectionAcquired(conn);
+ return conn;
+ }
+
+ private void createNewConnection(final ResponseHandler handler,
+ final Object attachment, final String schema) throws IOException {
+ // aysn create connection
+ MycatServer.getInstance().getBusinessExecutor().execute(new Runnable() {
+ public void run() {
+ try {
+ createNewConnection(new DelegateResponseHandler(handler) {
+ @Override
+ public void connectionError(Throwable e, BackendConnection conn) {
+ //decrementTotalConnectionsSafe(); // 如果创建连接失败，将当前连接数减1
+ handler.connectionError(e, conn);
+ }
+
+ @Override
+ public void connectionAcquired(BackendConnection conn) {
+ takeCon(conn, handler, attachment, schema);
+ }
+ }, schema);
+ } catch (IOException e) {
+ handler.connectionError(e, null);
+ }
+ }
+ });
+ }
+
+ public void getConnection(String schema, boolean autocommit,
+ final ResponseHandler handler, final Object attachment)
+ throws IOException {
+
+ // 从当前连接map中拿取已建立好的后端连接
+ BackendConnection con = this.conMap.tryTakeCon(schema, autocommit);
+ if (con != null) {
+ //如果不为空，则绑定对应前端请求的handler
+ takeCon(con, handler, attachment, schema);
+ return;
+
+ } else { // this.getActiveCount并不是线程安全的（严格上说该方法获取数量不准确），
+// int curTotalConnection = this.totalConnection.get();
+// while(curTotalConnection + 1 <= size) {
+//
+// if (this.totalConnection.compareAndSet(curTotalConnection, curTotalConnection + 1)) {
+// LOGGER.info("no ilde connection in pool,create new connection for " + this.name + " of schema " + schema);
+// createNewConnection(handler, attachment, schema);
+// return;
+// }
+//
+// curTotalConnection = this.totalConnection.get(); //CAS更新失败，则重新判断当前连接是否超过最大连接数
+//
+// }
+//
+// // 如果后端连接不足，立即失败,故直接抛出连接数超过最大连接异常
+// LOGGER.error("the max activeConnnections size can not be max than maxconnections:" + curTotalConnection);
+// throw new IOException("the max activeConnnections size can not be max than maxconnections:" + curTotalConnection);
+
+ int activeCons = this.getActiveCount();// 当前最大活动连接
+ if (activeCons + 1 > size) {// 下一个连接大于最大连接数
+ LOGGER.error("the max activeConnnections size can not be max than maxconnections");
+ throw new IOException("the max activeConnnections size can not be max than maxconnections");
+ } else { // create connection
+ LOGGER.info("no ilde connection in pool,create new connection for " + this.name + " of schema " + schema);
+ createNewConnection(handler, attachment, schema);
+ }
+ }
+ }
+
+ /**
+ * 是否超过最大连接数
+ * @return
+ */
+// private boolean exceedMaxConnections() {
+// return this.totalConnection.get() + 1 > size;
+// }
+//
+// public int decrementActiveCountSafe() {
+// return this.activeCount.decrementAndGet();
+// }
+//
+// public int incrementActiveCountSafe() {
+// return this.activeCount.incrementAndGet();
+// }
+//
+// public int getActiveCountSafe() {
+// return this.activeCount.get();
+// }
+//
+// public int getTotalConnectionsSafe() {
+// return this.totalConnection.get();
+// }
+//
+// public int decrementTotalConnectionsSafe() {
+// return this.totalConnection.decrementAndGet();
+// }
+//
+// public int incrementTotalConnectionSafe() {
+// return this.totalConnection.incrementAndGet();
+// }
+
+ private void returnCon(BackendConnection c) {
+
+ c.setAttachment(null);
+ c.setBorrowed(false);
+ c.setLastTime(TimeUtil.currentTimeMillis());
+ ConQueue queue = this.conMap.getSchemaConQueue(c.getSchema());
+
+ boolean ok = false;
+ if (c.isAutocommit()) {
+ ok = queue.getAutoCommitCons().offer(c);
+ } else {
+ ok = queue.getManCommitCons().offer(c);
+ }
+
+// if(c.getId() > 0 && takeConnectionContext.remove(c.getId(), TAKE_CONNECTION_FLAG) ) {
+// decrementActiveCountSafe();
+// }
+
+ if(!ok) {
+ LOGGER.warn("can't return to pool ,so close con " + c);
+ c.close("can't return to pool ");
+
+ }
+
+ }
+
+ public void releaseChannel(BackendConnection c) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("release channel " + c);
+ }
+ // release connection
+ returnCon(c);
+ }
+
+ public void connectionClosed(BackendConnection conn) {
+ ConQueue queue = this.conMap.getSchemaConQueue(conn.getSchema());
+ if (queue != null ) {
+ queue.removeCon(conn);
+ }
+
+// decrementTotalConnectionsSafe();
+ }
+
+ /**
+ * 创建新连接
+ */
+ public abstract void createNewConnection(ResponseHandler handler, String schema) throws IOException;
+
+ /**
+ * 测试连接，用于初始化及热更新配置检测
+ */
+ public abstract boolean testConnection(String schema) throws IOException;
+
+ public long getHeartbeatRecoveryTime() {
+ return heartbeatRecoveryTime;
+ }
+
+ public void setHeartbeatRecoveryTime(long heartbeatRecoveryTime) {
+ this.heartbeatRecoveryTime = heartbeatRecoveryTime;
+ }
+
+ public DBHostConfig getConfig() {
+ return config;
+ }
+
+ public boolean isAlive() {
+ return getHeartbeat().getStatus() == DBHeartbeat.OK_STATUS;
+ }
+}
diff --git a/src/main/java/io/mycat/backend/heartbeat/DBHeartbeat.java b/src/main/java/io/mycat/backend/heartbeat/DBHeartbeat.java
index cb265c2a1..285517438 100644
--- a/src/main/java/io/mycat/backend/heartbeat/DBHeartbeat.java
+++ b/src/main/java/io/mycat/backend/heartbeat/DBHeartbeat.java
@@ -23,10 +23,11 @@
*/
package io.mycat.backend.heartbeat;
-import io.mycat.backend.HeartbeatRecorder;
-
import java.util.concurrent.atomic.AtomicBoolean;
+import io.mycat.statistic.DataSourceSyncRecorder;
+import io.mycat.statistic.HeartbeatRecorder;
+
public abstract class DBHeartbeat {
public static final int DB_SYN_ERROR = -1;
public static final int DB_SYN_NORMAL = 1;
@@ -46,6 +47,7 @@ public abstract class DBHeartbeat {
protected int errorCount;
protected volatile int status;
protected final HeartbeatRecorder recorder = new HeartbeatRecorder();
+ protected final DataSourceSyncRecorder asynRecorder = new DataSourceSyncRecorder();
private volatile Integer slaveBehindMaster;
private volatile int dbSynStatus = DB_SYN_NORMAL;
@@ -124,4 +126,8 @@ public boolean isNeedHeartbeat() {
return heartbeatSQL != null;
}
+ public DataSourceSyncRecorder getAsynRecorder() {
+ return this.asynRecorder;
+ }
+
}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyChecker.java b/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyChecker.java
index 2756fa1ab..ed8699041 100644
--- a/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyChecker.java
+++ b/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyChecker.java
@@ -23,7 +23,7 @@
*/
package io.mycat.backend.heartbeat;
-import io.mycat.backend.MySQLDataSource;
+import io.mycat.backend.mysql.nio.MySQLDataSource;
import io.mycat.server.interceptor.impl.GlobalTableUtil;
import io.mycat.sqlengine.OneRawSQLQueryResultHandler;
import io.mycat.sqlengine.SQLJob;
@@ -54,9 +54,14 @@ public class MySQLConsistencyChecker{
private String maxSQL;
private String tableName; // global table name
private long beginTime;
- private String columnExistSQL = "select count(*) as "+GlobalTableUtil.INNER_COLUMN
- + " from information_schema.columns where column_name='"
- + GlobalTableUtil.GLOBAL_TABLE_MYCAT_COLUMN + "' and table_name='";
+// private String columnExistSQL = "select count(*) as "+GlobalTableUtil.INNER_COLUMN
+// + " from information_schema.columns where column_name='"
+// + GlobalTableUtil.GLOBAL_TABLE_MYCAT_COLUMN + "' and table_name='";
+
+ // 此处用到了 mysql 多行转一行 group_concat 的用法，主要是为了简化对结果的处理
+ // 得到的结果类似于：id,name,_mycat_op_time
+ private String columnExistSQL = "select group_concat(COLUMN_NAME separator ',') as "
+ + GlobalTableUtil.INNER_COLUMN +" from information_schema.columns where TABLE_NAME='"; //user' and TABLE_SCHEMA='db1';
private List>> list = new ArrayList<>();
@@ -74,34 +79,44 @@ public MySQLConsistencyChecker(MySQLDataSource source, String tableName) {
public void checkRecordCout() {
// ["db3","db2","db1"]
- this.jobCount.set(0);
- beginTime = new Date().getTime();
- String[] physicalSchemas = source.getDbPool().getSchemas();
- for(String dbName : physicalSchemas){
- MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null);
- OneRawSQLQueryResultHandler resultHandler =
- new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.COUNT_COLUMN}, detector);
- SQLJob sqlJob = new SQLJob(this.getCountSQL(), dbName, resultHandler, source);
- detector.setSqlJob(sqlJob);
- sqlJob.run();
- this.jobCount.incrementAndGet();
- }
+ lock.lock();
+ try{
+ this.jobCount.set(0);
+ beginTime = new Date().getTime();
+ String[] physicalSchemas = source.getDbPool().getSchemas();
+ for(String dbName : physicalSchemas){
+ MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null);
+ OneRawSQLQueryResultHandler resultHandler =
+ new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.COUNT_COLUMN}, detector);
+ SQLJob sqlJob = new SQLJob(this.getCountSQL(), dbName, resultHandler, source);
+ detector.setSqlJob(sqlJob);
+ sqlJob.run();
+ this.jobCount.incrementAndGet();
+ }
+ }finally{
+ lock.unlock();
+ }
}
public void checkMaxTimeStamp() {
// ["db3","db2","db1"]
- this.jobCount.set(0);
- beginTime = new Date().getTime();
- String[] physicalSchemas = source.getDbPool().getSchemas();
- for(String dbName : physicalSchemas){
- MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null);
- OneRawSQLQueryResultHandler resultHandler =
- new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.MAX_COLUMN}, detector);
- SQLJob sqlJob = new SQLJob(this.getMaxSQL(), dbName, resultHandler, source);
- detector.setSqlJob(sqlJob);
- sqlJob.run();
- this.jobCount.incrementAndGet();
- }
+ lock.lock();
+ try{
+ this.jobCount.set(0);
+ beginTime = new Date().getTime();
+ String[] physicalSchemas = source.getDbPool().getSchemas();
+ for(String dbName : physicalSchemas){
+ MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null);
+ OneRawSQLQueryResultHandler resultHandler =
+ new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.MAX_COLUMN}, detector);
+ SQLJob sqlJob = new SQLJob(this.getMaxSQL(), dbName, resultHandler, source);
+ detector.setSqlJob(sqlJob);
+ sqlJob.run();
+ this.jobCount.incrementAndGet();
+ }
+ }finally{
+ lock.unlock();
+ }
}
/**
@@ -109,20 +124,25 @@ public void checkMaxTimeStamp() {
*/
public void checkInnerColumnExist() {
// ["db3","db2","db1"]
- this.jobCount.set(0);
- beginTime = new Date().getTime();
- String[] physicalSchemas = source.getDbPool().getSchemas();
- for(String dbName : physicalSchemas){
- MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null, 1);
- OneRawSQLQueryResultHandler resultHandler =
- new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.INNER_COLUMN}, detector);
- String db = " and table_schema='" + dbName + "'";
- SQLJob sqlJob = new SQLJob(this.columnExistSQL + db , dbName, resultHandler, source);
- detector.setSqlJob(sqlJob);//table_schema='db1'
- LOGGER.debug(sqlJob.toString());
- sqlJob.run();
- this.jobCount.incrementAndGet();
- }
+ lock.lock();
+ try{
+ this.jobCount.set(0);
+ beginTime = new Date().getTime();
+ String[] physicalSchemas = source.getDbPool().getSchemas();
+ for(String dbName : physicalSchemas){
+ MySQLConsistencyHelper detector = new MySQLConsistencyHelper(this, null, 1);
+ OneRawSQLQueryResultHandler resultHandler =
+ new OneRawSQLQueryResultHandler(new String[] {GlobalTableUtil.INNER_COLUMN}, detector);
+ String db = " and table_schema='" + dbName + "'";
+ SQLJob sqlJob = new SQLJob(this.columnExistSQL + db , dbName, resultHandler, source);
+ detector.setSqlJob(sqlJob);//table_schema='db1'
+ LOGGER.debug(sqlJob.toString());
+ sqlJob.run();
+ this.jobCount.incrementAndGet();
+ }
+ }finally{
+ lock.unlock();
+ }
}
public void setResult(SQLQueryResult> result) {
diff --git a/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyHelper.java b/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyHelper.java
index 21488f9f3..67b6400aa 100644
--- a/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyHelper.java
+++ b/src/main/java/io/mycat/backend/heartbeat/MySQLConsistencyHelper.java
@@ -129,14 +129,4 @@ public void setSqlJob(SQLJob sqlJob) {
this.sqlJob = sqlJob;
}
-// if(count == null){
-// LOGGER.warn(heartbeat.getCountSQL() + " execute failed in db: "
-// + result.getDataNode() +" during global table consistency heartbeat.");
-// }
-// if(maxTimestamp == null){
-// LOGGER.warn(heartbeat.getMaxSQL() + " execute failed in db: "
-// + result.getDataNode() +" during global table consistency heartbeat.");
-// }
-// return;
-// }
}
diff --git a/src/main/java/io/mycat/backend/heartbeat/MySQLDetector.java b/src/main/java/io/mycat/backend/heartbeat/MySQLDetector.java
index 0306907a0..691ae2ad7 100644
--- a/src/main/java/io/mycat/backend/heartbeat/MySQLDetector.java
+++ b/src/main/java/io/mycat/backend/heartbeat/MySQLDetector.java
@@ -2,8 +2,8 @@
* Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
- * This code is free software;Designed and Developed mainly by many Chinese
- * opensource volunteers. you can redistribute it and/or modify it under the
+ * This code is free software;Designed and Developed mainly by many Chinese
+ * opensource volunteers. you can redistribute it and/or modify it under the
* terms of the GNU General Public License version 2 only, as published by the
* Free Software Foundation.
*
@@ -16,145 +16,201 @@
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
- *
- * Any questions about this component can be directed to it's project Web address
+ *
+ * Any questions about this component can be directed to it's project Web address
* https://code.google.com/p/opencloudb/.
*
*/
package io.mycat.backend.heartbeat;
-import io.mycat.backend.MySQLDataSource;
-import io.mycat.backend.PhysicalDBPool;
-import io.mycat.backend.PhysicalDatasource;
-import io.mycat.server.config.node.DataHostConfig;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.mycat.backend.datasource.PhysicalDBPool;
+import io.mycat.backend.datasource.PhysicalDatasource;
+import io.mycat.backend.mysql.nio.MySQLDataSource;
+import io.mycat.config.model.DataHostConfig;
import io.mycat.sqlengine.OneRawSQLQueryResultHandler;
import io.mycat.sqlengine.SQLJob;
import io.mycat.sqlengine.SQLQueryResult;
import io.mycat.sqlengine.SQLQueryResultListener;
import io.mycat.util.TimeUtil;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-
/**
* @author mycat
*/
-public class MySQLDetector implements
- SQLQueryResultListener>> {
- private MySQLHeartbeat heartbeat;
- private long heartbeatTimeout;
- private final AtomicBoolean isQuit;
- private volatile long lastSendQryTime;
- private volatile long lasstReveivedQryTime;
- private volatile SQLJob sqlJob;
- private static final String[] MYSQL_SLAVE_STAUTS_COLMS = new String[] {
- "Seconds_Behind_Master", "Slave_IO_Running", "Slave_SQL_Running" };
-
- public MySQLDetector(MySQLHeartbeat heartbeat) {
- this.heartbeat = heartbeat;
- this.isQuit = new AtomicBoolean(false);
- }
-
- public MySQLHeartbeat getHeartbeat() {
- return heartbeat;
- }
-
- public long getHeartbeatTimeout() {
- return heartbeatTimeout;
- }
-
- public void setHeartbeatTimeout(long heartbeatTimeout) {
- this.heartbeatTimeout = heartbeatTimeout;
- }
-
- public boolean isHeartbeatTimeout() {
- return TimeUtil.currentTimeMillis() > Math.max(lastSendQryTime,
- lasstReveivedQryTime) + heartbeatTimeout;
- }
-
- public long getLastSendQryTime() {
- return lastSendQryTime;
- }
-
- public long getLasstReveivedQryTime() {
- return lasstReveivedQryTime;
- }
-
- public void heartbeat() {
- lastSendQryTime = System.currentTimeMillis();
- MySQLDataSource ds = heartbeat.getSource();
- String databaseName = ds.getDbPool().getSchemas()[0];
- String[] fetchColms={};
- if (heartbeat.getSource().getHostConfig().isShowSlaveSql() ) {
- fetchColms=MYSQL_SLAVE_STAUTS_COLMS;
- }
- OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler(
- fetchColms, this);
- sqlJob = new SQLJob(heartbeat.getHeartbeatSQL(), databaseName,
- resultHandler, ds);
- sqlJob.run();
- }
-
- public void quit() {
- if (isQuit.compareAndSet(false, true)) {
- close("heart beat quit");
- }
-
- }
-
- public boolean isQuit() {
- return isQuit.get();
- }
-
- @Override
- public void onResult(SQLQueryResult> result) {
- if (result.isSuccess()) {
- int balance = heartbeat.getSource().getDbPool().getBalance();
- PhysicalDatasource source = heartbeat.getSource();
+public class MySQLDetector implements SQLQueryResultListener>> {
+
+ private MySQLHeartbeat heartbeat;
+
+ private long heartbeatTimeout;
+ private final AtomicBoolean isQuit;
+ private volatile long lastSendQryTime;
+ private volatile long lasstReveivedQryTime;
+ private volatile SQLJob sqlJob;
+
+ private static final String[] MYSQL_SLAVE_STAUTS_COLMS = new String[] {
+ "Seconds_Behind_Master",
+ "Slave_IO_Running",
+ "Slave_SQL_Running",
+ "Slave_IO_State",
+ "Master_Host",
+ "Master_User",
+ "Master_Port",
+ "Connect_Retry",
+ "Last_IO_Error"};
+
+ private static final String[] MYSQL_CLUSTER_STAUTS_COLMS = new String[] {
+ "Variable_name",
+ "Value"};
+
+ public MySQLDetector(MySQLHeartbeat heartbeat) {
+ this.heartbeat = heartbeat;
+ this.isQuit = new AtomicBoolean(false);
+ }
+
+ public MySQLHeartbeat getHeartbeat() {
+ return heartbeat;
+ }
+
+ public long getHeartbeatTimeout() {
+ return heartbeatTimeout;
+ }
+
+ public void setHeartbeatTimeout(long heartbeatTimeout) {
+ this.heartbeatTimeout = heartbeatTimeout;
+ }
+
+ public boolean isHeartbeatTimeout() {
+ return TimeUtil.currentTimeMillis() > Math.max(lastSendQryTime,
+ lasstReveivedQryTime) + heartbeatTimeout;
+ }
+
+ public long getLastSendQryTime() {
+ return lastSendQryTime;
+ }
+
+ public long getLasstReveivedQryTime() {
+ return lasstReveivedQryTime;
+ }
+
+ public void heartbeat() {
+ lastSendQryTime = System.currentTimeMillis();
+ MySQLDataSource ds = heartbeat.getSource();
+ String databaseName = ds.getDbPool().getSchemas()[0];
+ String[] fetchColms={};
+ if (heartbeat.getSource().getHostConfig().isShowSlaveSql() ) {
+ fetchColms=MYSQL_SLAVE_STAUTS_COLMS;
+ }
+ if (heartbeat.getSource().getHostConfig().isShowClusterSql() ) {
+ fetchColms=MYSQL_CLUSTER_STAUTS_COLMS;
+ }
+ OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler( fetchColms, this);
+ sqlJob = new SQLJob(heartbeat.getHeartbeatSQL(), databaseName, resultHandler, ds);
+ sqlJob.run();
+ }
+
+ public void quit() {
+ if (isQuit.compareAndSet(false, true)) {
+ close("heart beat quit");
+ }
+
+ }
+
+ public boolean isQuit() {
+ return isQuit.get();
+ }
+
+ @Override
+ public void onResult(SQLQueryResult> result) {
+
+ if (result.isSuccess()) {
+
+ int balance = heartbeat.getSource().getDbPool().getBalance();
+
+ PhysicalDatasource source = heartbeat.getSource();
+ int switchType = source.getHostConfig().getSwitchType();
Map resultResult = result.getResult();
- if (source.getHostConfig().isShowSlaveSql()
- &&(source.getHostConfig().getSwitchType() == DataHostConfig.SYN_STATUS_SWITCH_DS ||
- PhysicalDBPool.BALANCE_NONE!=balance )
- )
- {
-
- String Slave_IO_Running =resultResult!=null? resultResult.get(
- "Slave_IO_Running"):null;
- String Slave_SQL_Running = resultResult!=null?resultResult.get(
- "Slave_SQL_Running"):null;
- if (Slave_IO_Running != null
- && Slave_IO_Running.equals(Slave_SQL_Running)
- && Slave_SQL_Running.equals("Yes")) {
- heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_NORMAL);
- String Seconds_Behind_Master = resultResult.get(
- "Seconds_Behind_Master");
- if (null != Seconds_Behind_Master
- && !"".equals(Seconds_Behind_Master)) {
- heartbeat.setSlaveBehindMaster(Integer
- .valueOf(Seconds_Behind_Master));
- }
- } else if(source.isSalveOrRead())
- {
- MySQLHeartbeat.LOGGER
- .warn("found MySQL master/slave Replication err !!! "
- + heartbeat.getSource().getConfig());
- heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_ERROR);
- }
-
- }
- heartbeat.setResult(MySQLHeartbeat.OK_STATUS, this, null);
- } else {
- heartbeat.setResult(MySQLHeartbeat.ERROR_STATUS, this, null);
- }
- lasstReveivedQryTime = System.currentTimeMillis();
- }
-
- public void close(String msg) {
- SQLJob curJob = sqlJob;
- if (curJob != null && !curJob.isFinished()) {
- curJob.teminate(msg);
- sqlJob = null;
- }
- }
-
-}
\ No newline at end of file
+
+ if ( resultResult!=null&& !resultResult.isEmpty() &&switchType == DataHostConfig.SYN_STATUS_SWITCH_DS
+ && source.getHostConfig().isShowSlaveSql()) {
+
+ String Slave_IO_Running = resultResult != null ? resultResult.get("Slave_IO_Running") : null;
+ String Slave_SQL_Running = resultResult != null ? resultResult.get("Slave_SQL_Running") : null;
+
+ if (Slave_IO_Running != null
+ && Slave_IO_Running.equals(Slave_SQL_Running)
+ && Slave_SQL_Running.equals("Yes")) {
+
+ heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_NORMAL);
+ String Seconds_Behind_Master = resultResult.get( "Seconds_Behind_Master");
+ if (null != Seconds_Behind_Master && !"".equals(Seconds_Behind_Master)) {
+
+ int Behind_Master = Integer.parseInt(Seconds_Behind_Master);
+ if ( Behind_Master > source.getHostConfig().getSlaveThreshold() ) {
+ MySQLHeartbeat.LOGGER.warn("found MySQL master/slave Replication delay !!! "
+ + heartbeat.getSource().getConfig() + ", binlog sync time delay: " + Behind_Master + "s" );
+ }
+ heartbeat.setSlaveBehindMaster( Behind_Master );
+ }
+
+ } else if( source.isSalveOrRead() ) {
+ //String Last_IO_Error = resultResult != null ? resultResult.get("Last_IO_Error") : null;
+ MySQLHeartbeat.LOGGER.warn("found MySQL master/slave Replication err !!! "
+ + heartbeat.getSource().getConfig() + ", " + resultResult);
+ heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_ERROR);
+ }
+
+ heartbeat.getAsynRecorder().set(resultResult, switchType);
+ heartbeat.setResult(MySQLHeartbeat.OK_STATUS, this, null);
+
+ } else if ( resultResult!=null&& !resultResult.isEmpty() && switchType==DataHostConfig.CLUSTER_STATUS_SWITCH_DS
+ && source.getHostConfig().isShowClusterSql() ) {
+
+ //String Variable_name = resultResult != null ? resultResult.get("Variable_name") : null;
+ String wsrep_cluster_status = resultResult != null ? resultResult.get("wsrep_cluster_status") : null;// Primary
+ String wsrep_connected = resultResult != null ? resultResult.get("wsrep_connected") : null;// ON
+ String wsrep_ready = resultResult != null ? resultResult.get("wsrep_ready") : null;// ON
+
+ if ("ON".equals(wsrep_connected)
+ && "ON".equals(wsrep_ready)
+ && "Primary".equals(wsrep_cluster_status)) {
+
+ heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_NORMAL);
+ heartbeat.setResult(MySQLHeartbeat.OK_STATUS, this, null);
+
+ } else {
+ MySQLHeartbeat.LOGGER.warn("found MySQL cluster status err !!! "
+ + heartbeat.getSource().getConfig()
+ + " wsrep_cluster_status: "+ wsrep_cluster_status
+ + " wsrep_connected: "+ wsrep_connected
+ + " wsrep_ready: "+ wsrep_ready
+ );
+
+ heartbeat.setDbSynStatus(DBHeartbeat.DB_SYN_ERROR);
+ heartbeat.setResult(MySQLHeartbeat.ERROR_STATUS, this, null);
+ }
+ heartbeat.getAsynRecorder().set(resultResult, switchType);
+
+ } else {
+ heartbeat.setResult(MySQLHeartbeat.OK_STATUS, this, null);
+ }
+ //监测数据库同步状态，在 switchType=-1或者1的情况下，也需要收集主从同步状态
+ heartbeat.getAsynRecorder().set(resultResult, switchType);
+
+ } else {
+ heartbeat.setResult(MySQLHeartbeat.ERROR_STATUS, this, null);
+ }
+
+ lasstReveivedQryTime = System.currentTimeMillis();
+ heartbeat.getRecorder().set((lasstReveivedQryTime - lastSendQryTime));
+ }
+
+ public void close(String msg) {
+ SQLJob curJob = sqlJob;
+ if (curJob != null && !curJob.isFinished()) {
+ curJob.teminate(msg);
+ sqlJob = null;
+ }
+ }
+}
diff --git a/src/main/java/io/mycat/backend/heartbeat/MySQLHeartbeat.java b/src/main/java/io/mycat/backend/heartbeat/MySQLHeartbeat.java
index bb11de2d2..1bb66a674 100644
--- a/src/main/java/io/mycat/backend/heartbeat/MySQLHeartbeat.java
+++ b/src/main/java/io/mycat/backend/heartbeat/MySQLHeartbeat.java
@@ -23,25 +23,24 @@
*/
package io.mycat.backend.heartbeat;
-import io.mycat.backend.MySQLDataSource;
-import io.mycat.backend.PhysicalDBPool;
-import io.mycat.backend.PhysicalDatasource;
-import io.mycat.server.config.node.DataHostConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.backend.datasource.PhysicalDBPool;
+import io.mycat.backend.datasource.PhysicalDatasource;
+import io.mycat.backend.mysql.nio.MySQLDataSource;
+import io.mycat.config.model.DataHostConfig;
+
/**
* @author mycat
*/
public class MySQLHeartbeat extends DBHeartbeat {
private static final int MAX_RETRY_COUNT = 5;
- public static final Logger LOGGER = LoggerFactory
- .getLogger(MySQLHeartbeat.class);
+ public static final Logger LOGGER = LoggerFactory.getLogger(MySQLHeartbeat.class);
private final MySQLDataSource source;
@@ -55,7 +54,7 @@ public MySQLHeartbeat(MySQLDataSource source) {
this.lock = new ReentrantLock(false);
this.maxRetryCount = MAX_RETRY_COUNT;
this.status = INIT_STATUS;
- this.heartbeatSQL = source.getHostConfig().getHeartbeatSQL();
+ this.heartbeatSQL = source.getHostConfig().getHearbeatSQL();
}
public MySQLDataSource getSource() {
@@ -171,9 +170,6 @@ public void setResult(int result, MySQLDetector detector, String msg) {
}
private void setOk(MySQLDetector detector) {
-
- recorder.set(detector.getLasstReveivedQryTime()
- - detector.getLastSendQryTime());
switch (status) {
case DBHeartbeat.TIMEOUT_STATUS:
this.status = DBHeartbeat.INIT_STATUS;
@@ -199,19 +195,17 @@ private void setError(MySQLDetector detector) {
// should continues check error status
if (++errorCount < maxRetryCount) {
- if (detector != null && !detector.isQuit()) {
- heartbeat(); // error count not enough, heart beat again
- }
- //return;
- } else
+ if (detector != null && !detector.isQuit()) {
+ heartbeat(); // error count not enough, heart beat again
+ }
+
+ }else
{
if (detector != null ) {
detector.quit();
}
-
this.status = ERROR_STATUS;
this.errorCount = 0;
-
}
}
@@ -243,8 +237,7 @@ private void switchSourceIfNeed(String reason) {
synchronized (pool) {
// try to see if need switch datasource
curDatasourceHB = pool.getSource().getHeartbeat().getStatus();
- if (curDatasourceHB != DBHeartbeat.INIT_STATUS
- && curDatasourceHB != DBHeartbeat.OK_STATUS) {
+ if (curDatasourceHB != DBHeartbeat.INIT_STATUS && curDatasourceHB != DBHeartbeat.OK_STATUS) {
int curIndex = pool.getActivedIndex();
int nextId = pool.next(curIndex);
PhysicalDatasource[] allWriteNodes = pool.getSources();
@@ -257,24 +250,20 @@ private void switchSourceIfNeed(String reason) {
int theSourceHBStatus = theSourceHB.getStatus();
if (theSourceHBStatus == DBHeartbeat.OK_STATUS) {
if (switchType == DataHostConfig.SYN_STATUS_SWITCH_DS) {
- if (Integer.valueOf(0).equals(
- theSourceHB.getSlaveBehindMaster())) {
- LOGGER.info("try to switch datasource ,slave is synchronized to master "
- + theSource.getConfig());
+ if (Integer.valueOf(0).equals( theSourceHB.getSlaveBehindMaster())) {
+ LOGGER.info("try to switch datasource ,slave is synchronized to master " + theSource.getConfig());
pool.switchSource(nextId, true, reason);
break;
} else {
LOGGER.warn("ignored datasource ,slave is not synchronized to master , slave behind master :"
- + theSourceHB
- .getSlaveBehindMaster()
+ + theSourceHB.getSlaveBehindMaster()
+ " " + theSource.getConfig());
}
} else {
// normal switch
- LOGGER.info("try to switch datasource ,not checked slave synchronize status "
- + theSource.getConfig());
+ LOGGER.info("try to switch datasource ,not checked slave synchronize status " + theSource.getConfig());
pool.switchSource(nextId, true, reason);
- break;
+ break;
}
}
diff --git a/src/main/java/io/mycat/backend/jdbc/JDBCConnection.java b/src/main/java/io/mycat/backend/jdbc/JDBCConnection.java
index d2972be7c..11ca234e4 100644
--- a/src/main/java/io/mycat/backend/jdbc/JDBCConnection.java
+++ b/src/main/java/io/mycat/backend/jdbc/JDBCConnection.java
@@ -1,575 +1,877 @@
-package io.mycat.backend.jdbc;
-
-import io.mycat.backend.BackendConnection;
-import io.mycat.net.BufferArray;
-import io.mycat.net.NetSystem;
-import io.mycat.route.RouteResultsetNode;
-import io.mycat.server.ErrorCode;
-import io.mycat.server.Isolations;
-import io.mycat.server.MySQLFrontConnection;
-import io.mycat.server.executors.ConnectionHeartBeatHandler;
-import io.mycat.server.executors.ResponseHandler;
-import io.mycat.server.packet.*;
-import io.mycat.server.parser.ServerParse;
-import io.mycat.server.response.ShowVariables;
-import io.mycat.util.ResultSetUtil;
-import io.mycat.util.StringUtil;
-import io.mycat.util.TimeUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-
-public class JDBCConnection implements BackendConnection {
- protected static final Logger LOGGER = LoggerFactory
- .getLogger(JDBCConnection.class);
- private JDBCDatasource pool;
- private volatile String schema;
- private volatile String dbType;
- private volatile String oldSchema;
- private byte packetId;
- private int txIsolation;
- private volatile boolean running = false;
- private volatile boolean borrowed;
- private long id = 0;
- private String host;
- private int port;
- private Connection con;
- private ResponseHandler respHandler;
- private volatile Object attachement;
-
- boolean headerOutputed = false;
- private volatile boolean modifiedSQLExecuted;
- private final long startTime;
- private long lastTime;
- private boolean isSpark = false;
-
-
- public JDBCConnection() {
- startTime = System.currentTimeMillis();
- }
-
- public Connection getCon() {
- return con;
- }
-
- public void setCon(Connection con) {
- this.con = con;
-
- }
-
- @Override
- public void close(String reason) {
- try {
- con.close();
-
- } catch (SQLException e) {
- }
-
- }
-
- public void setId(long id) {
- this.id = id;
- }
-
- public JDBCDatasource getPool() {
- return pool;
- }
-
- public void setPool(JDBCDatasource pool) {
- this.pool = pool;
- }
-
- public void setHost(String host) {
- this.host = host;
- }
-
- public void setPort(int port) {
- this.port = port;
- }
-
- @Override
- public boolean isClosed() {
- try {
- return con == null || con.isClosed();
- } catch (SQLException e) {
- return true;
- }
- }
-
- @Override
- public void idleCheck() {
- if (TimeUtil.currentTimeMillis() > lastTime
- + pool.getConfig().getIdleTimeout()) {
- close(" idle check");
- }
- }
-
- @Override
- public long getStartupTime() {
- return startTime;
- }
-
- public String getHost() {
- return this.host;
- }
-
- public int getPort() {
- return this.port;
- }
-
- public int getLocalPort() {
- return 0;
- }
-
- public long getNetInBytes() {
-
- return 0;
- }
-
- public long getNetOutBytes() {
- return 0;
- }
-
- @Override
- public boolean isModifiedSQLExecuted() {
- return modifiedSQLExecuted;
- }
-
- @Override
- public boolean isFromSlaveDB() {
- return false;
- }
-
- public String getDbType() {
- return this.dbType;
- }
-
- public void setDbType(String newDbType) {
- this.dbType = newDbType.toUpperCase();
- this.isSpark = dbType.equals("SPARK");
-
- }
-
- @Override
- public String getSchema() {
- return this.schema;
- }
-
- @Override
- public void setSchema(String newSchema) {
- this.oldSchema = this.schema;
- this.schema = newSchema;
-
- }
-
- @Override
- public long getLastTime() {
-
- return lastTime;
- }
-
- @Override
- public boolean isClosedOrQuit() {
- return this.isClosed();
- }
-
- @Override
- public void setAttachment(Object attachment) {
- this.attachement = attachment;
-
- }
-
- @Override
- public void quit() {
- this.close("client quit");
-
- }
-
- @Override
- public void setLastTime(long currentTimeMillis) {
- this.lastTime = currentTimeMillis;
-
- }
-
- @Override
- public void release() {
- modifiedSQLExecuted = false;
- setResponseHandler(null);
- pool.releaseChannel(this);
- }
-
- public void setRunning(boolean running) {
- this.running = running;
-
- }
-
- @Override
- public void setResponseHandler(ResponseHandler commandHandler) {
- respHandler = commandHandler;
- }
-
- @Override
- public void commit() {
- try {
- con.commit();
-
- this.respHandler.okResponse(OkPacket.OK, this);
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
-
- private int convertNativeIsolationToJDBC(int nativeIsolation) {
- if (nativeIsolation == Isolations.REPEATED_READ) {
- return Connection.TRANSACTION_REPEATABLE_READ;
- } else if (nativeIsolation == Isolations.SERIALIZABLE) {
- return Connection.TRANSACTION_SERIALIZABLE;
- } else {
- return nativeIsolation;
- }
- }
-
- private void syncIsolation(int nativeIsolation) {
- int jdbcIsolation = convertNativeIsolationToJDBC(nativeIsolation);
- int srcJdbcIsolation = getTxIsolation();
- if (jdbcIsolation == srcJdbcIsolation)
- return;
- if ("oracle".equalsIgnoreCase(getDbType())
- && jdbcIsolation != Connection.TRANSACTION_READ_COMMITTED
- && jdbcIsolation != Connection.TRANSACTION_SERIALIZABLE) {
- // oracle 只支持2个级别 ,且只能更改一次隔离级别，否则会报 ORA-01453
- return;
- }
- try {
- con.setTransactionIsolation(jdbcIsolation);
- } catch (SQLException e) {
- LOGGER.warn("set txisolation error:", e);
- }
- }
-
- private void executeSQL(RouteResultsetNode rrn, MySQLFrontConnection sc,
- boolean autocommit) throws IOException {
- String orgin = rrn.getStatement();
- // String sql = rrn.getStatement().toLowerCase();
- // LOGGER.info("JDBC SQL:"+orgin+"|"+sc.toString());
- if (!modifiedSQLExecuted && rrn.isModifySQL()) {
- modifiedSQLExecuted = true;
- }
-
- try {
-
- syncIsolation(sc.getTxIsolation());
- if (!this.schema.equals(this.oldSchema)) {
- con.setCatalog(schema);
- this.oldSchema = schema;
- }
- if (!this.isSpark) {
- con.setAutoCommit(autocommit);
- }
- int sqlType = rrn.getSqlType();
-
- if (sqlType == ServerParse.SELECT || sqlType == ServerParse.SHOW) {
- if ((sqlType == ServerParse.SHOW) && (!dbType.equals("MYSQL"))) {
- // showCMD(sc, orgin);
- // ShowVariables.execute(sc, orgin);
- ShowVariables.execute(sc);
-// } else if ("SELECT CONNECTION_ID()".equalsIgnoreCase(orgin)) {
-// // ShowVariables.justReturnValue(sc,String.valueOf(sc.getId()));
-// ShowVariables.justReturnValue(sc,
-// String.valueOf(sc.getId()), this);
- } else
- {
- ouputResultSet(sc, orgin);
- }
- } else {
- executeddl(sc, orgin);
- }
-
- } catch (SQLException e) {
-
- String msg = e.getMessage();
- ErrorPacket error = new ErrorPacket();
- error.packetId = ++packetId;
- error.errno = e.getErrorCode();
- error.message = msg.getBytes();
- this.respHandler.errorResponse(error.writeToBytes(), this);
- } catch (Exception e) {
- String msg = e.getMessage();
- ErrorPacket error = new ErrorPacket();
- error.packetId = ++packetId;
- error.errno = ErrorCode.ER_UNKNOWN_ERROR;
- error.message = msg.getBytes();
- this.respHandler.errorResponse(error.writeToBytes(), this);
- } finally {
- this.running = false;
- }
-
- }
-
-
-
- private void executeddl(MySQLFrontConnection sc, String sql)
- throws SQLException {
- Statement stmt = null;
- try {
- stmt = con.createStatement();
- int count = stmt.executeUpdate(sql);
- OkPacket okPck = new OkPacket();
- okPck.affectedRows = count;
- okPck.insertId = 0;
- okPck.packetId = ++packetId;
- okPck.message = " OK!".getBytes();
- this.respHandler.okResponse(okPck.writeToBytes(), this);
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
-
- }
- }
- }
- }
-
- private void ouputResultSet(MySQLFrontConnection sc, String sql)
- throws SQLException {
- ResultSet rs = null;
- Statement stmt = null;
-
- try {
- stmt = con.createStatement();
- rs = stmt.executeQuery(sql);
-
- List fieldPks = new LinkedList();
- ResultSetUtil.resultSetToFieldPacket(sc.getCharset(), fieldPks, rs,
- this.isSpark);
- int colunmCount = fieldPks.size();
- BufferArray bufferArray = NetSystem.getInstance().getBufferPool()
- .allocateArray();
- ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();
- headerPkg.fieldCount = fieldPks.size();
- headerPkg.packetId = ++packetId;
-
- headerPkg.write(bufferArray);
-
- byte[] header =bufferArray.writeToByteArrayAndRecycle();
-
- List fields = new ArrayList(fieldPks.size());
- Iterator itor = fieldPks.iterator();
- while (itor.hasNext()) {
- bufferArray = NetSystem.getInstance().getBufferPool()
- .allocateArray();
- FieldPacket curField = itor.next();
- curField.packetId = ++packetId;
- curField.write(bufferArray);
- byte[] field = bufferArray.writeToByteArrayAndRecycle();
- fields.add(field);
- itor.remove();
- }
-
- bufferArray = NetSystem.getInstance().getBufferPool()
- .allocateArray();
- EOFPacket eofPckg = new EOFPacket();
- eofPckg.packetId = ++packetId;
- eofPckg.write(bufferArray);
- byte[] eof = bufferArray.writeToByteArrayAndRecycle();
- this.respHandler.fieldEofResponse(header, fields, eof, this);
-
- // output row
- while (rs.next()) {
- bufferArray = NetSystem.getInstance().getBufferPool()
- .allocateArray();
- RowDataPacket curRow = new RowDataPacket(colunmCount);
- for (int i = 0; i < colunmCount; i++) {
- int j = i + 1;
- curRow.add(StringUtil.encode(rs.getString(j),
- sc.getCharset()));
- }
- curRow.packetId = ++packetId;
- curRow.write(bufferArray);
- byte[] row =bufferArray.writeToByteArrayAndRecycle();
- this.respHandler.rowResponse(row, this);
- }
-
- // end row
- bufferArray = NetSystem.getInstance().getBufferPool()
- .allocateArray();
- eofPckg = new EOFPacket();
- eofPckg.packetId = ++packetId;
- eofPckg.write(bufferArray);
- eof = bufferArray.writeToByteArrayAndRecycle();
- this.respHandler.rowEofResponse(eof, this);
- } finally {
- if (rs != null) {
- try {
- rs.close();
- } catch (SQLException e) {
-
- }
- }
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
-
- }
- }
- }
- }
-
- @Override
- public void query(final String sql) throws UnsupportedEncodingException {
- if (respHandler instanceof ConnectionHeartBeatHandler) {
- justForHeartbeat(sql);
- } else {
- throw new UnsupportedEncodingException("unsupported yet ");
- }
- }
-
- private void justForHeartbeat(String sql) {
-
- Statement stmt = null;
-
- try {
- stmt = con.createStatement();
- stmt.execute(sql);
- if (!isAutocommit()) { // 如果在写库上，如果是事务方式的连接，需要进行手动commit
- con.commit();
- }
- this.respHandler.okResponse(OkPacket.OK, this);
-
- } catch (Exception e) {
- String msg = e.getMessage();
- ErrorPacket error = new ErrorPacket();
- error.packetId = ++packetId;
- error.errno = ErrorCode.ER_UNKNOWN_ERROR;
- error.message = msg.getBytes();
- this.respHandler.errorResponse(error.writeToBytes(), this);
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
-
- }
- }
- }
- }
-
- @Override
- public Object getAttachment() {
- return this.attachement;
- }
-
- @Override
- public String getCharset() {
- return null;
- }
-
- @Override
- public void execute(final RouteResultsetNode node,
- final MySQLFrontConnection source, final boolean autocommit)
- throws IOException {
- Runnable runnable = new Runnable() {
- @Override
- public void run() {
- try {
- executeSQL(node, source, autocommit);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
-
- NetSystem.getInstance().getExecutor().execute(runnable);
- }
-
- @Override
- public boolean syncAndExcute() {
- return true;
- }
-
- @Override
- public void rollback() {
- try {
- con.rollback();
-
- this.respHandler.okResponse(OkPacket.OK, this);
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
-
- public boolean isRunning() {
- return this.running;
- }
-
- @Override
- public boolean isBorrowed() {
- return this.borrowed;
- }
-
- @Override
- public void setBorrowed(boolean borrowed) {
- this.borrowed = borrowed;
-
- }
-
- @Override
- public int getTxIsolation() {
- if (con != null) {
- try {
- return con.getTransactionIsolation();
- } catch (SQLException e) {
- return 0;
- }
- } else {
- return -1;
- }
- }
-
- @Override
- public boolean isAutocommit() {
- if (con == null) {
- return true;
- } else {
- try {
- return con.getAutoCommit();
- } catch (SQLException e) {
-
- }
- }
- return true;
- }
-
- @Override
- public long getId() {
- return id;
- }
-
- @Override
- public String toString() {
- return "JDBCConnection [id=" + id + ",autocommit="
- + this.isAutocommit() + ",pool=" + pool + ", schema=" + schema
- + ", dbType=" + dbType + ", oldSchema=" + oldSchema
- + ", packetId=" + packetId + ", txIsolation=" + txIsolation
- + ", running=" + running + ", borrowed=" + borrowed + ", host="
- + host + ", port=" + port + ", con=" + con + ", respHandler="
- + respHandler + ", attachement=" + attachement
- + ", headerOutputed=" + headerOutputed
- + ", modifiedSQLExecuted=" + modifiedSQLExecuted
- + ", startTime=" + startTime + ", lastTime=" + lastTime
- + ", isSpark=" + isSpark+"]";
- }
-
-}
+package io.mycat.backend.jdbc;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.*;
+import java.util.*;
+
+import io.mycat.backend.mysql.PacketUtil;
+import io.mycat.route.Procedure;
+import io.mycat.route.ProcedureParameter;
+import io.mycat.util.*;
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.MycatServer;
+import io.mycat.backend.BackendConnection;
+import io.mycat.backend.mysql.nio.handler.ConnectionHeartBeatHandler;
+import io.mycat.backend.mysql.nio.handler.ResponseHandler;
+import io.mycat.config.ErrorCode;
+import io.mycat.config.Isolations;
+import io.mycat.net.NIOProcessor;
+import io.mycat.net.mysql.EOFPacket;
+import io.mycat.net.mysql.ErrorPacket;
+import io.mycat.net.mysql.FieldPacket;
+import io.mycat.net.mysql.OkPacket;
+import io.mycat.net.mysql.ResultSetHeaderPacket;
+import io.mycat.net.mysql.RowDataPacket;
+import io.mycat.route.RouteResultsetNode;
+import io.mycat.server.ServerConnection;
+import io.mycat.server.parser.ServerParse;
+
+public class JDBCConnection implements BackendConnection {
+ protected static final Logger LOGGER = LoggerFactory
+ .getLogger(JDBCConnection.class);
+ private JDBCDatasource pool;
+ private volatile String schema;
+ private volatile String dbType;
+ private volatile String oldSchema;
+ private byte packetId;
+ private int txIsolation;
+ private volatile boolean running = false;
+ private volatile boolean borrowed;
+ private long id = 0;
+ private String host;
+ private int port;
+ private Connection con;
+ private ResponseHandler respHandler;
+ private volatile Object attachement;
+
+ boolean headerOutputed = false;
+ private volatile boolean modifiedSQLExecuted;
+ private final long startTime;
+ private long lastTime;
+ private boolean isSpark = false;
+
+ private NIOProcessor processor;
+
+
+
+ public NIOProcessor getProcessor() {
+ return processor;
+ }
+
+ public void setProcessor(NIOProcessor processor) {
+ this.processor = processor;
+ }
+
+ public JDBCConnection() {
+ startTime = System.currentTimeMillis();
+ }
+
+ public Connection getCon() {
+ return con;
+ }
+
+ public void setCon(Connection con) {
+ this.con = con;
+
+ }
+
+ @Override
+ public void close(String reason) {
+ try {
+ con.close();
+ if(processor!=null){
+ processor.removeConnection(this);
+ }
+
+ } catch (SQLException e) {
+ }
+
+ }
+
+ public void setId(long id) {
+ this.id = id;
+ }
+
+ public JDBCDatasource getPool() {
+ return pool;
+ }
+
+ public void setPool(JDBCDatasource pool) {
+ this.pool = pool;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ @Override
+ public boolean isClosed() {
+ try {
+ return con == null || con.isClosed();
+ } catch (SQLException e) {
+ return true;
+ }
+ }
+
+ @Override
+ public void idleCheck() {
+ if(TimeUtil.currentTimeMillis() > lastTime + pool.getConfig().getIdleTimeout()){
+ close(" idle check");
+ }
+ }
+
+ @Override
+ public long getStartupTime() {
+ return startTime;
+ }
+
+ @Override
+ public String getHost() {
+ return this.host;
+ }
+
+ @Override
+ public int getPort() {
+ return this.port;
+ }
+
+ @Override
+ public int getLocalPort() {
+ return 0;
+ }
+
+ @Override
+ public long getNetInBytes() {
+
+ return 0;
+ }
+
+ @Override
+ public long getNetOutBytes() {
+ return 0;
+ }
+
+ @Override
+ public boolean isModifiedSQLExecuted() {
+ return modifiedSQLExecuted;
+ }
+
+ @Override
+ public boolean isFromSlaveDB() {
+ return false;
+ }
+
+ public String getDbType() {
+ return this.dbType;
+ }
+
+ public void setDbType(String newDbType) {
+ this.dbType = newDbType.toUpperCase();
+ this.isSpark = dbType.equals("SPARK");
+
+ }
+
+ @Override
+ public String getSchema() {
+ return this.schema;
+ }
+
+ @Override
+ public void setSchema(String newSchema) {
+ this.oldSchema = this.schema;
+ this.schema = newSchema;
+
+ }
+
+ @Override
+ public long getLastTime() {
+
+ return lastTime;
+ }
+
+ @Override
+ public boolean isClosedOrQuit() {
+ return this.isClosed();
+ }
+
+ @Override
+ public void setAttachment(Object attachment) {
+ this.attachement = attachment;
+
+ }
+
+ @Override
+ public void quit() {
+ this.close("client quit");
+
+ }
+
+ @Override
+ public void setLastTime(long currentTimeMillis) {
+ this.lastTime = currentTimeMillis;
+
+ }
+
+ @Override
+ public void release() {
+ modifiedSQLExecuted = false;
+ setResponseHandler(null);
+ pool.releaseChannel(this);
+ }
+
+ public void setRunning(boolean running) {
+ this.running = running;
+
+ }
+
+ @Override
+ public boolean setResponseHandler(ResponseHandler commandHandler) {
+ respHandler = commandHandler;
+ return false;
+ }
+
+ @Override
+ public void commit() {
+ try {
+ con.commit();
+
+ this.respHandler.okResponse(OkPacket.OK, this);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ private int convertNativeIsolationToJDBC(int nativeIsolation)
+ {
+ if(nativeIsolation== Isolations.REPEATED_READ)
+ {
+ return Connection.TRANSACTION_REPEATABLE_READ;
+ }else
+ if(nativeIsolation== Isolations.SERIALIZABLE)
+ {
+ return Connection.TRANSACTION_SERIALIZABLE;
+ } else
+ {
+ return nativeIsolation;
+ }
+ }
+
+
+
+ private void syncIsolation(int nativeIsolation)
+ {
+ int jdbcIsolation=convertNativeIsolationToJDBC(nativeIsolation);
+ int srcJdbcIsolation= getTxIsolation();
+ if (jdbcIsolation == srcJdbcIsolation || "oracle".equalsIgnoreCase(getDbType())
+ && jdbcIsolation != Connection.TRANSACTION_READ_COMMITTED
+ && jdbcIsolation != Connection.TRANSACTION_SERIALIZABLE) {
+ return;
+ }
+ try
+ {
+ con.setTransactionIsolation(jdbcIsolation);
+ } catch (SQLException e)
+ {
+ LOGGER.warn("set txisolation error:",e);
+ }
+ }
+ private void executeSQL(RouteResultsetNode rrn, ServerConnection sc,
+ boolean autocommit) throws IOException {
+ String orgin = rrn.getStatement();
+ // String sql = rrn.getStatement().toLowerCase();
+ // LOGGER.info("JDBC SQL:"+orgin+"|"+sc.toString());
+ if (!modifiedSQLExecuted && rrn.isModifySQL()) {
+ modifiedSQLExecuted = true;
+ }
+
+ try {
+ syncIsolation(sc.getTxIsolation()) ;
+ if (!this.schema.equals(this.oldSchema)) {
+ con.setCatalog(schema);
+ this.oldSchema = schema;
+ }
+ if (!this.isSpark) {
+ con.setAutoCommit(autocommit);
+ }
+ int sqlType = rrn.getSqlType();
+ if(rrn.isCallStatement()&&"oracle".equalsIgnoreCase(getDbType()))
+ {
+ //存储过程暂时只支持oracle
+ ouputCallStatement(rrn,sc,orgin);
+ } else
+ if (sqlType == ServerParse.SELECT || sqlType == ServerParse.SHOW) {
+ if ((sqlType == ServerParse.SHOW) && (!dbType.equals("MYSQL"))) {
+ // showCMD(sc, orgin);
+ //ShowVariables.execute(sc, orgin);
+ ShowVariables.execute(sc, orgin,this);
+ } else if ("SELECT CONNECTION_ID()".equalsIgnoreCase(orgin)) {
+ //ShowVariables.justReturnValue(sc,String.valueOf(sc.getId()));
+ ShowVariables.justReturnValue(sc,String.valueOf(sc.getId()),this);
+ } else {
+ ouputResultSet(sc, orgin);
+ }
+ } else {
+ executeddl(sc, orgin);
+ }
+
+ } catch (SQLException e) {
+
+ String msg = e.getMessage();
+ ErrorPacket error = new ErrorPacket();
+ error.packetId = ++packetId;
+ error.errno = e.getErrorCode();
+ error.message = msg.getBytes();
+ this.respHandler.errorResponse(error.writeToBytes(sc), this);
+ }
+ catch (Exception e) {
+ String msg = e.getMessage();
+ ErrorPacket error = new ErrorPacket();
+ error.packetId = ++packetId;
+ error.errno = ErrorCode.ER_UNKNOWN_ERROR;
+ error.message = ((msg == null) ? e.toString().getBytes() : msg.getBytes());
+ String err = null;
+ if(error.message!=null){
+ err = new String(error.message);
+ }
+ LOGGER.error("sql execute error, "+ err , e);
+ this.respHandler.errorResponse(error.writeToBytes(sc), this);
+ }
+ finally {
+ this.running = false;
+ }
+
+ }
+
+ private FieldPacket getNewFieldPacket(String charset, String fieldName) {
+ FieldPacket fieldPacket = new FieldPacket();
+ fieldPacket.orgName = StringUtil.encode(fieldName, charset);
+ fieldPacket.name = StringUtil.encode(fieldName, charset);
+ fieldPacket.length = 20;
+ fieldPacket.flags = 0;
+ fieldPacket.decimals = 0;
+ int javaType = 12;
+ fieldPacket.type = (byte) (MysqlDefs.javaTypeMysql(javaType) & 0xff);
+ return fieldPacket;
+ }
+
+ private void executeddl(ServerConnection sc, String sql)
+ throws SQLException {
+ Statement stmt = null;
+ try {
+ stmt = con.createStatement();
+ int count = stmt.executeUpdate(sql);
+ OkPacket okPck = new OkPacket();
+ okPck.affectedRows = count;
+ okPck.insertId = 0;
+ okPck.packetId = ++packetId;
+ okPck.message = " OK!".getBytes();
+ this.respHandler.okResponse(okPck.writeToBytes(sc), this);
+ } finally {
+ if (stmt != null) {
+ try {
+ stmt.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ }
+ }
+
+
+ private static int oracleCURSORTypeValue=-10;
+ static
+ {
+ Object cursor = ObjectUtil.getStaticFieldValue("oracle.jdbc.OracleTypes", "CURSOR");
+ if(cursor!=null) {
+ oracleCURSORTypeValue = (int) cursor;
+ }
+ }
+ private void ouputCallStatement(RouteResultsetNode rrn,ServerConnection sc, String sql)
+ throws SQLException {
+
+ CallableStatement stmt = null;
+ ResultSet rs = null;
+ try {
+ Procedure procedure = rrn.getProcedure();
+ Collection paramters= procedure.getParamterMap().values();
+ String callSql = procedure.toPreCallSql(null);
+ stmt = con.prepareCall(callSql);
+
+ for (ProcedureParameter paramter : paramters)
+ {
+ if((ProcedureParameter.IN.equalsIgnoreCase(paramter.getParameterType())
+ ||ProcedureParameter.INOUT.equalsIgnoreCase(paramter.getParameterType())))
+ {
+ Object value= paramter.getValue()!=null ?paramter.getValue():paramter.getName();
+ stmt.setObject(paramter.getIndex(),value);
+ }
+
+ if(ProcedureParameter.OUT.equalsIgnoreCase(paramter.getParameterType())
+ ||ProcedureParameter.INOUT.equalsIgnoreCase(paramter.getParameterType()) )
+ {
+ int jdbcType ="oracle".equalsIgnoreCase(getDbType())&& procedure.getListFields().contains(paramter.getName())?oracleCURSORTypeValue: paramter.getJdbcType();
+ stmt.registerOutParameter(paramter.getIndex(), jdbcType);
+ }
+ }
+
+ boolean hadResults= stmt.execute();
+
+ ByteBuffer byteBuf = sc.allocate();
+ if(procedure.getSelectColumns().size()>0)
+ {
+ List fieldPks = new LinkedList();
+ for (ProcedureParameter paramter : paramters)
+ {
+ if (!procedure.getListFields().contains(paramter.getName())&&(ProcedureParameter.OUT.equalsIgnoreCase(paramter.getParameterType())
+ || ProcedureParameter.INOUT.equalsIgnoreCase(paramter.getParameterType())) )
+ {
+ FieldPacket packet = PacketUtil.getField(paramter.getName(), MysqlDefs.javaTypeMysql(paramter.getJdbcType()));
+ fieldPks.add(packet);
+ }
+ }
+ int colunmCount = fieldPks.size();
+
+ ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();
+ headerPkg.fieldCount = fieldPks.size();
+ headerPkg.packetId = ++packetId;
+
+ byteBuf = headerPkg.write(byteBuf, sc, true);
+ byteBuf.flip();
+ byte[] header = new byte[byteBuf.limit()];
+ byteBuf.get(header);
+ byteBuf.clear();
+
+
+ List fields = new ArrayList(fieldPks.size());
+ Iterator itor = fieldPks.iterator();
+ while (itor.hasNext()) {
+ FieldPacket curField = itor.next();
+ curField.packetId = ++packetId;
+ byteBuf = curField.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] field = new byte[byteBuf.limit()];
+ byteBuf.get(field);
+ byteBuf.clear();
+ fields.add(field);
+ itor.remove();
+ }
+ EOFPacket eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ byteBuf.clear();
+ this.respHandler.fieldEofResponse(header, fields, eof, this);
+ RowDataPacket curRow = new RowDataPacket(colunmCount);
+ for (String name : procedure.getSelectColumns())
+ {
+ ProcedureParameter procedureParameter= procedure.getParamterMap().get(name);
+ curRow.add(StringUtil.encode(String.valueOf(stmt.getObject(procedureParameter.getIndex())),
+ sc.getCharset()));
+ }
+
+ curRow.packetId = ++packetId;
+ byteBuf = curRow.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] row = new byte[byteBuf.limit()];
+ byteBuf.get(row);
+ byteBuf.clear();
+ this.respHandler.rowResponse(row, this);
+
+ eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ if(procedure.isResultList())
+ {
+ eofPckg.status = 42;
+ }
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ byteBuf.clear();
+ this.respHandler.rowEofResponse(eof, this);
+ }
+
+
+ if(procedure.isResultList())
+ {
+ List fieldPks = new LinkedList();
+ int listSize=procedure.getListFields().size();
+ for (ProcedureParameter paramter : paramters)
+ {
+ if (procedure.getListFields().contains(paramter.getName())&&(ProcedureParameter.OUT.equalsIgnoreCase(paramter.getParameterType())
+ || ProcedureParameter.INOUT.equalsIgnoreCase(paramter.getParameterType())) )
+ {
+ listSize--;
+
+ Object object = stmt.getObject(paramter.getIndex());
+ rs= (ResultSet) object;
+ if(rs==null) {
+ continue;
+ }
+ ResultSetUtil.resultSetToFieldPacket(sc.getCharset(), fieldPks, rs,
+ this.isSpark);
+
+ int colunmCount = fieldPks.size();
+ ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();
+ headerPkg.fieldCount = fieldPks.size();
+ headerPkg.packetId = ++packetId;
+
+ byteBuf = headerPkg.write(byteBuf, sc, true);
+ byteBuf.flip();
+ byte[] header = new byte[byteBuf.limit()];
+ byteBuf.get(header);
+ byteBuf.clear();
+
+
+ List fields = new ArrayList(fieldPks.size());
+ Iterator itor = fieldPks.iterator();
+ while (itor.hasNext()) {
+ FieldPacket curField = itor.next();
+ curField.packetId = ++packetId;
+ byteBuf = curField.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] field = new byte[byteBuf.limit()];
+ byteBuf.get(field);
+ byteBuf.clear();
+ fields.add(field);
+ itor.remove();
+ }
+ EOFPacket eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ byteBuf.clear();
+ this.respHandler.fieldEofResponse(header, fields, eof, this);
+
+ // output row
+ while (rs.next()) {
+ RowDataPacket curRow = new RowDataPacket(colunmCount);
+ for (int i = 0; i < colunmCount; i++) {
+ int j = i + 1;
+ curRow.add(StringUtil.encode(rs.getString(j),
+ sc.getCharset()));
+ }
+ curRow.packetId = ++packetId;
+ byteBuf = curRow.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] row = new byte[byteBuf.limit()];
+ byteBuf.get(row);
+ byteBuf.clear();
+ this.respHandler.rowResponse(row, this);
+ }
+ eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ if(listSize!=0)
+ {
+ eofPckg.status = 42;
+ }
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ byteBuf.clear();
+ this.respHandler.rowEofResponse(eof, this);
+ }
+ }
+
+ }
+
+
+
+ if(!procedure.isResultSimpleValue())
+ {
+ byte[] OK = new byte[] { 7, 0, 0, 1, 0, 0, 0, 2, 0, 0,
+ 0 };
+ OK[3]=++packetId;
+ this.respHandler.okResponse(OK,this);
+ }
+ sc.recycle(byteBuf);
+ } finally {
+ if (rs != null) {
+ try {
+ rs.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ if (stmt != null) {
+ try {
+ stmt.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ }
+ }
+
+
+ private void ouputResultSet(ServerConnection sc, String sql)
+ throws SQLException {
+ ResultSet rs = null;
+ Statement stmt = null;
+
+ try {
+ stmt = con.createStatement();
+ rs = stmt.executeQuery(sql);
+
+ List fieldPks = new LinkedList();
+ ResultSetUtil.resultSetToFieldPacket(sc.getCharset(), fieldPks, rs,
+ this.isSpark);
+ int colunmCount = fieldPks.size();
+ ByteBuffer byteBuf = sc.allocate();
+ ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();
+ headerPkg.fieldCount = fieldPks.size();
+ headerPkg.packetId = ++packetId;
+
+ byteBuf = headerPkg.write(byteBuf, sc, true);
+ byteBuf.flip();
+ byte[] header = new byte[byteBuf.limit()];
+ byteBuf.get(header);
+ byteBuf.clear();
+ List fields = new ArrayList(fieldPks.size());
+ Iterator itor = fieldPks.iterator();
+ while (itor.hasNext()) {
+ FieldPacket curField = itor.next();
+ curField.packetId = ++packetId;
+ byteBuf = curField.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] field = new byte[byteBuf.limit()];
+ byteBuf.get(field);
+ byteBuf.clear();
+ fields.add(field);
+ }
+ EOFPacket eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ byteBuf.clear();
+ this.respHandler.fieldEofResponse(header, fields, eof, this);
+
+ // output row
+ while (rs.next()) {
+ RowDataPacket curRow = new RowDataPacket(colunmCount);
+ for (int i = 0; i < colunmCount; i++) {
+ int j = i + 1;
+ if(MysqlDefs.isBianry((byte) fieldPks.get(i).type)) {
+ curRow.add(rs.getBytes(j));
+ } else if(fieldPks.get(i).type == MysqlDefs.FIELD_TYPE_DECIMAL ||
+ fieldPks.get(i).type == (MysqlDefs.FIELD_TYPE_NEW_DECIMAL - 256)) { // field type is unsigned byte
+ // ensure that do not use scientific notation format
+ BigDecimal val = rs.getBigDecimal(j);
+ curRow.add(StringUtil.encode(val != null ? val.toPlainString() : null,
+ sc.getCharset()));
+ } else {
+ curRow.add(StringUtil.encode(rs.getString(j),
+ sc.getCharset()));
+ }
+
+ }
+ curRow.packetId = ++packetId;
+ byteBuf = curRow.write(byteBuf, sc, false);
+ byteBuf.flip();
+ byte[] row = new byte[byteBuf.limit()];
+ byteBuf.get(row);
+ byteBuf.clear();
+ this.respHandler.rowResponse(row, this);
+ }
+
+ fieldPks.clear();
+
+ // end row
+ eofPckg = new EOFPacket();
+ eofPckg.packetId = ++packetId;
+ byteBuf = eofPckg.write(byteBuf, sc, false);
+ byteBuf.flip();
+ eof = new byte[byteBuf.limit()];
+ byteBuf.get(eof);
+ sc.recycle(byteBuf);
+ this.respHandler.rowEofResponse(eof, this);
+ } finally {
+ if (rs != null) {
+ try {
+ rs.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ if (stmt != null) {
+ try {
+ stmt.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ }
+ }
+
+ @Override
+ public void query(final String sql) throws UnsupportedEncodingException {
+ if(respHandler instanceof ConnectionHeartBeatHandler)
+ {
+ justForHeartbeat(sql);
+ } else
+ {
+ throw new UnsupportedEncodingException("unsupported yet ");
+ }
+ }
+ private void justForHeartbeat(String sql)
+ {
+
+ Statement stmt = null;
+
+ try {
+ stmt = con.createStatement();
+ stmt.execute(sql);
+ if(!isAutocommit()){ //如果在写库上，如果是事务方式的连接，需要进行手动commit
+ con.commit();
+ }
+ this.respHandler.okResponse(OkPacket.OK, this);
+
+ }
+ catch (Exception e)
+ {
+ String msg = e.getMessage();
+ ErrorPacket error = new ErrorPacket();
+ error.packetId = ++packetId;
+ error.errno = ErrorCode.ER_UNKNOWN_ERROR;
+ error.message = msg.getBytes();
+ this.respHandler.errorResponse(error.writeToBytes(), this);
+ }
+ finally {
+ if (stmt != null) {
+ try {
+ stmt.close();
+ } catch (SQLException e) {
+
+ }
+ }
+ }
+ }
+ @Override
+ public Object getAttachment() {
+ return this.attachement;
+ }
+
+ @Override
+ public String getCharset() {
+ return null;
+ }
+
+ @Override
+ public void execute(final RouteResultsetNode node,
+ final ServerConnection source, final boolean autocommit)
+ throws IOException {
+ Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ executeSQL(node, source, autocommit);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ MycatServer.getInstance().getBusinessExecutor().execute(runnable);
+ }
+
+ @Override
+ public void recordSql(String host, String schema, String statement) {
+
+ }
+
+ @Override
+ public boolean syncAndExcute() {
+ return true;
+ }
+
+ @Override
+ public void rollback() {
+ try {
+ con.rollback();
+
+ this.respHandler.okResponse(OkPacket.OK, this);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public boolean isRunning() {
+ return this.running;
+ }
+
+ @Override
+ public boolean isBorrowed() {
+ return this.borrowed;
+ }
+
+ @Override
+ public void setBorrowed(boolean borrowed) {
+ this.borrowed = borrowed;
+
+ }
+
+ @Override
+ public int getTxIsolation() {
+ if (con != null) {
+ try {
+ return con.getTransactionIsolation();
+ } catch (SQLException e) {
+ return 0;
+ }
+ } else {
+ return -1;
+ }
+ }
+
+ @Override
+ public boolean isAutocommit() {
+ if (con == null) {
+ return true;
+ } else {
+ try {
+ return con.getAutoCommit();
+ } catch (SQLException e) {
+
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public long getId() {
+ return id;
+ }
+
+ @Override
+ public String toString() {
+ return "JDBCConnection [id=" + id +",autocommit="+this.isAutocommit()+",pool=" + pool + ", schema=" + schema + ", dbType=" + dbType + ", oldSchema="
+ + oldSchema + ", packetId=" + packetId + ", txIsolation=" + txIsolation + ", running=" + running
+ + ", borrowed=" + borrowed + ", host=" + host + ", port=" + port + ", con=" + con
+ + ", respHandler=" + respHandler + ", attachement=" + attachement + ", headerOutputed="
+ + headerOutputed + ", modifiedSQLExecuted=" + modifiedSQLExecuted + ", startTime=" + startTime
+ + ", lastTime=" + lastTime + ", isSpark=" + isSpark + ", processor=" + processor + "]";
+ }
+
+ @Override
+ public void discardClose(String reason) {
+ // TODO Auto-generated method stub
+
+ }
+
+
+
+}
diff --git a/src/main/java/io/mycat/backend/jdbc/JDBCDatasource.java b/src/main/java/io/mycat/backend/jdbc/JDBCDatasource.java
index 4a8ccd91b..597c50ec3 100644
--- a/src/main/java/io/mycat/backend/jdbc/JDBCDatasource.java
+++ b/src/main/java/io/mycat/backend/jdbc/JDBCDatasource.java
@@ -1,54 +1,48 @@
package io.mycat.backend.jdbc;
-import io.mycat.backend.PhysicalDatasource;
-import io.mycat.backend.heartbeat.DBHeartbeat;
-import io.mycat.net.ConnectIdGenerator;
-import io.mycat.server.config.loader.LocalLoader;
-import io.mycat.server.config.node.DBHostConfig;
-import io.mycat.server.config.node.DataHostConfig;
-import io.mycat.server.config.node.JdbcDriver;
-import io.mycat.server.executors.ResponseHandler;
-
import java.io.IOException;
import java.sql.Connection;
-import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
-import java.util.Enumeration;
-import java.util.Map;
+import java.util.List;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.google.common.collect.Lists;
+
+import io.mycat.MycatServer;
+import io.mycat.backend.datasource.PhysicalDatasource;
+import io.mycat.backend.heartbeat.DBHeartbeat;
+import io.mycat.backend.mysql.nio.handler.ResponseHandler;
+import io.mycat.config.model.DBHostConfig;
+import io.mycat.config.model.DataHostConfig;
+import io.mycat.net.NIOConnector;
+import io.mycat.net.NIOProcessor;
public class JDBCDatasource extends PhysicalDatasource {
- public static final Logger logger = LoggerFactory.getLogger(JDBCDatasource.class);
- private static Map jdbcDriverConfig = null;
- static { // 最多也就3,4个数据库，一次性加载驱动类
- jdbcDriverConfig = LocalLoader.loadJdbcDriverConfig();
- if(jdbcDriverConfig != null && jdbcDriverConfig.size() > 0){
- for(String key : jdbcDriverConfig.keySet()){
- JdbcDriver driver = jdbcDriverConfig.get(key);
- if(driver != null && StringUtils.isNotBlank(driver.getClassName())){
- try {
- Class.forName(driver.getClassName());
- } catch (ClassNotFoundException e) {
- logger.error("Class.forName load jdbcDriver for "+key+" error: " + e.getMessage());
- }
- }else{
- logger.error(" driver for " + key + " is not exist or className has no value,"
- + " please check jdbcDriver-config element in mycat.xml.");
- }
+ static {
+ // 加载可能的驱动
+ List drivers = Lists.newArrayList(
+ "com.mysql.jdbc.Driver",
+ "io.mycat.backend.jdbc.mongodb.MongoDriver",
+ "io.mycat.backend.jdbc.sequoiadb.SequoiaDriver",
+ "oracle.jdbc.OracleDriver",
+ "com.microsoft.sqlserver.jdbc.SQLServerDriver",
+ "net.sourceforge.jtds.jdbc.Driver",
+ "org.apache.hive.jdbc.HiveDriver",
+ "com.ibm.db2.jcc.DB2Driver",
+ "org.postgresql.Driver");
+
+ for (String driver : drivers) {
+ try {
+ Class.forName(driver);
+ } catch (ClassNotFoundException ignored) {
}
}
}
-
- public JDBCDatasource(DBHostConfig config, DataHostConfig hostConfig,
- boolean isReadNode) {
+
+ public JDBCDatasource(DBHostConfig config, DataHostConfig hostConfig, boolean isReadNode) {
super(config, hostConfig, isReadNode);
-
}
@Override
@@ -57,69 +51,75 @@ public DBHeartbeat createHeartBeat() {
}
@Override
- public void createNewConnection(ResponseHandler handler, String schema)
- throws IOException {
+ public void createNewConnection(ResponseHandler handler,String schema) throws IOException {
DBHostConfig cfg = getConfig();
+ JDBCConnection c = new JDBCConnection();
+ c.setHost(cfg.getIp());
+ c.setPort(cfg.getPort());
+ c.setPool(this);
+ c.setSchema(schema);
+ c.setDbType(cfg.getDbType());
- JDBCConnection c = null;
+ NIOProcessor processor = (NIOProcessor) MycatServer.getInstance().nextProcessor();
+ c.setProcessor(processor);
+ c.setId(NIOConnector.ID_GENERATOR.getId()); //复用mysql的Backend的ID，需要在process中存储
+
+ processor.addBackend(c);
try {
- // TODO: 这里需要实现连继池
Connection con = getConnection();
- c = new JDBCConnection();
- c.setHost(cfg.getIp());
- c.setPort(cfg.getPort());
- c.setPool(this);
- c.setSchema(schema);
- c.setDbType(cfg.getDbType());
- c.setId(ConnectIdGenerator.getINSTNCE().getId()); // 复用mysql的Backend的ID，需要在process中存储
-
// c.setIdleTimeout(pool.getConfig().getIdleTimeout());
c.setCon(con);
// notify handler
handler.connectionAcquired(c);
-
} catch (Exception e) {
handler.connectionError(e, c);
}
+ }
+
+ @Override
+ public boolean testConnection(String schema) throws IOException {
+ boolean isConnected = false;
+
+ Connection connection = null;
+ Statement statement = null;
+ try {
+ DBHostConfig cfg = getConfig();
+ connection = DriverManager.getConnection(cfg.getUrl(), cfg.getUser(), cfg.getPassword());
+ statement = connection.createStatement();
+ if (connection != null && statement != null) {
+ isConnected = true;
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ } finally {
+ if (statement != null) {
+ try { statement.close(); } catch (SQLException e) {}
+ }
+
+ if (connection != null) {
+ try { connection.close(); } catch (SQLException e) {}
+ }
+ }
+ return isConnected;
}
- Connection getConnection() throws SQLException {
- DBHostConfig cfg = getConfig();
- Enumeration drivers = DriverManager.getDrivers();
- Driver d = drivers.nextElement();
- d.getClass().getName();
- Connection connection = DriverManager.getConnection(cfg.getUrl(),
- cfg.getUser(), cfg.getPassword());
- String initSql = getHostConfig().getConnectionInitSql();
- if (StringUtils.isNotBlank(initSql)) {
- try (Statement statement = connection.createStatement()){
+ Connection getConnection() throws SQLException {
+ DBHostConfig cfg = getConfig();
+ Connection connection = DriverManager.getConnection(cfg.getUrl(), cfg.getUser(), cfg.getPassword());
+ String initSql=getHostConfig().getConnectionInitSql();
+ if (initSql != null && !"".equals(initSql)) {
+ Statement statement = null;
+ try {
+ statement = connection.createStatement();
statement.execute(initSql);
- } catch(SQLException e) {
- logger.warn(" getConnection error: " + e.getMessage());
+ } finally {
+ if (statement != null) {
+ statement.close();
+ }
}
}
return connection;
- }
-
- /**
- * 根据 dbType 获取 JdbcDriver
- * @param dbType mysql
- * @return JdbcDriver: {'mysql':'com.mysql.jdbc.Driver'}
- */
- public static JdbcDriver getJdbcDriverBydbType(String dbType){
- if(StringUtils.isNotBlank(dbType)){
- return jdbcDriverConfig.get(dbType.toLowerCase()); // 获取对应 dbType 的 JdbcDriver
- }
- return null;
- }
-
- public static Map getJdbcDriverConfig() {
- return jdbcDriverConfig;
- }
-
- public static void setJdbcDriverConfig(Map jdbcDriverConfig) {
- JDBCDatasource.jdbcDriverConfig = jdbcDriverConfig;
- }
-
+ }
+
}
diff --git a/src/main/java/io/mycat/backend/jdbc/JDBCHeartbeat.java b/src/main/java/io/mycat/backend/jdbc/JDBCHeartbeat.java
index 4194c983f..85b2b2524 100644
--- a/src/main/java/io/mycat/backend/jdbc/JDBCHeartbeat.java
+++ b/src/main/java/io/mycat/backend/jdbc/JDBCHeartbeat.java
@@ -1,33 +1,33 @@
package io.mycat.backend.jdbc;
-import io.mycat.backend.HeartbeatRecorder;
-import io.mycat.backend.heartbeat.DBHeartbeat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.sql.Connection;
+import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.backend.heartbeat.DBHeartbeat;
+import io.mycat.statistic.HeartbeatRecorder;
+
public class JDBCHeartbeat extends DBHeartbeat{
private final ReentrantLock lock;
private final JDBCDatasource source;
private final boolean heartbeatnull;
private Long lastSendTime = System.currentTimeMillis();
private Long lastReciveTime = System.currentTimeMillis();
-
-
- private static final Logger logger = LoggerFactory
- .getLogger(JDBCHeartbeat.class);
+
+
+ private Logger logger = LoggerFactory.getLogger(this.getClass());
public JDBCHeartbeat(JDBCDatasource source)
{
this.source = source;
lock = new ReentrantLock(false);
this.status = INIT_STATUS;
- this.heartbeatSQL = source.getHostConfig().getHeartbeatSQL().trim();
+ this.heartbeatSQL = source.getHostConfig().getHearbeatSQL().trim();
this.heartbeatnull= heartbeatSQL.length()==0;
}
@@ -88,8 +88,9 @@ public HeartbeatRecorder getRecorder() {
public void heartbeat()
{
- if (isStop.get())
+ if (isStop.get()) {
return;
+ }
lastSendTime = System.currentTimeMillis();
lock.lock();
try
diff --git a/src/main/java/io/mycat/backend/jdbc/ShowVariables.java b/src/main/java/io/mycat/backend/jdbc/ShowVariables.java
new file mode 100644
index 000000000..77130c483
--- /dev/null
+++ b/src/main/java/io/mycat/backend/jdbc/ShowVariables.java
@@ -0,0 +1,198 @@
+/*
+ * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software;Designed and Developed mainly by many Chinese
+ * opensource volunteers. you can redistribute it and/or modify it under the
+ * terms of the GNU General Public License version 2 only, as published by the
+ * Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Any questions about this component can be directed to it's project Web address
+ * https://code.google.com/p/opencloudb/.
+ *
+ */
+package io.mycat.backend.jdbc;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.slf4j.Logger; import org.slf4j.LoggerFactory;
+
+import io.mycat.backend.BackendConnection;
+import io.mycat.backend.mysql.PacketUtil;
+import io.mycat.config.Fields;
+import io.mycat.net.mysql.EOFPacket;
+import io.mycat.net.mysql.FieldPacket;
+import io.mycat.net.mysql.ResultSetHeaderPacket;
+import io.mycat.net.mysql.RowDataPacket;
+import io.mycat.server.NonBlockingSession;
+import io.mycat.server.ServerConnection;
+import io.mycat.util.StringUtil;
+
+/**
+ * @author mycat
+ */
+public final class ShowVariables
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ShowVariables.class);
+ private static final int FIELD_COUNT = 2;
+ private static final ResultSetHeaderPacket header = PacketUtil.getHeader(FIELD_COUNT);
+ private static final FieldPacket[] fields = new FieldPacket[FIELD_COUNT];
+ private static final EOFPacket eof = new EOFPacket();
+ private static final Pattern pattern = Pattern.compile("(?:like|=)\\s*'([^']*(?:\\w+)+[^']*)+'",Pattern.CASE_INSENSITIVE);
+ static {
+ int i = 0;
+ byte packetId = 0;
+ header.packetId = ++packetId;
+
+ fields[i] = PacketUtil.getField("VARIABLE_NAME", Fields.FIELD_TYPE_VAR_STRING);
+ fields[i++].packetId = ++packetId;
+
+ fields[i] = PacketUtil.getField("VALUE", Fields.FIELD_TYPE_VAR_STRING);
+ fields[i++].packetId = ++packetId;
+
+ eof.packetId = ++packetId;
+ }
+ private static List parseVariable(String sql)
+ {
+ List variableList=new ArrayList<>();
+ Matcher matcher = pattern.matcher(sql);
+ while (matcher.find())
+ {
+ variableList.add(matcher.group(1));
+ }
+ return variableList;
+ }
+ public static void execute(ServerConnection c, String sql) {
+ ByteBuffer buffer = c.allocate();
+
+ // write header
+ buffer = header.write(buffer, c,true);
+
+ // write fields
+ for (FieldPacket field : fields) {
+ buffer = field.write(buffer, c,true);
+ }
+
+ // write eof
+ buffer = eof.write(buffer, c,true);
+
+ // write rows
+ byte packetId = eof.packetId;
+
+ List variableList= parseVariable(sql);
+ for (String key : variableList)
+ {
+ String value= variables.get(key) ;
+ if(value!=null)
+ {
+ RowDataPacket row = getRow(key, value, c.getCharset());
+ row.packetId = ++packetId;
+ buffer = row.write(buffer, c,true);
+ }
+ }
+
+
+
+ // write lastEof
+ EOFPacket lastEof = new EOFPacket();
+ lastEof.packetId = ++packetId;
+ buffer = lastEof.write(buffer, c,true);
+
+ // write buffer
+ c.write(buffer);
+ }
+
+ public static void justReturnValue(ServerConnection c, String value) {
+ ByteBuffer buffer = c.allocate();
+
+ // write header
+ buffer = header.write(buffer, c,true);
+
+ // write fields
+ for (FieldPacket field : fields) {
+ buffer = field.write(buffer, c,true);
+ }
+
+ // write eof
+ buffer = eof.write(buffer, c,true);
+
+ // write rows
+ byte packetId = eof.packetId;
+
+
+
+ if(value!=null)
+ {
+
+ RowDataPacket row = new RowDataPacket(1);
+ row.add(StringUtil.encode(value, c.getCharset()));
+ row.packetId = ++packetId;
+ buffer = row.write(buffer, c,true);
+ }
+
+
+
+ // write lastEof
+ EOFPacket lastEof = new EOFPacket();
+ lastEof.packetId = ++packetId;
+ buffer = lastEof.write(buffer, c,true);
+
+ // write buffer
+ c.write(buffer);
+ }
+
+ private static RowDataPacket getRow(String name, String value, String charset) {
+ RowDataPacket row = new RowDataPacket(FIELD_COUNT);
+ row.add(StringUtil.encode(name, charset));
+ row.add(StringUtil.encode(value, charset));
+ return row;
+ }
+
+ private static final Map variables = new HashMap();
+ static {
+ variables.put("character_set_client", "utf8");
+ variables.put("character_set_connection", "utf8");
+ variables.put("character_set_results", "utf8");
+ variables.put("character_set_server", "utf8");
+ variables.put("init_connect", "");
+ variables.put("interactive_timeout", "172800");
+ variables.put("lower_case_table_names", "1");
+ variables.put("max_allowed_packet", "16777216");
+ variables.put("net_buffer_length", "16384");
+ variables.put("net_write_timeout", "60");
+ variables.put("query_cache_size", "0");
+ variables.put("query_cache_type", "OFF");
+ variables.put("sql_mode", "STRICT_TRANS_TABLES");
+ variables.put("system_time_zone", "CST");
+ variables.put("time_zone", "SYSTEM");
+ variables.put("tx_isolation", "REPEATABLE-READ");
+ variables.put("wait_timeout", "172800");
+ }
+
+ public static void execute(ServerConnection sc, String orgin, BackendConnection jdbcConnection) {
+ execute(sc, orgin);
+ NonBlockingSession session = sc.getSession2();
+ session.releaseConnectionIfSafe(jdbcConnection, LOGGER.isDebugEnabled(), false);
+ }
+ public static void justReturnValue(ServerConnection sc, String orgin, BackendConnection jdbcConnection) {
+ justReturnValue(sc, orgin);
+ NonBlockingSession session = sc.getSession2();
+ session.releaseConnectionIfSafe(jdbcConnection, LOGGER.isDebugEnabled(), false);
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/jdbc/mongodb/DriverPropertyInfoHelper.java b/src/main/java/io/mycat/backend/jdbc/mongodb/DriverPropertyInfoHelper.java
index 483e0ef99..43dcb529f 100644
--- a/src/main/java/io/mycat/backend/jdbc/mongodb/DriverPropertyInfoHelper.java
+++ b/src/main/java/io/mycat/backend/jdbc/mongodb/DriverPropertyInfoHelper.java
@@ -1,70 +1,70 @@
-package io.mycat.backend.jdbc.mongodb;
-
-import java.sql.DriverPropertyInfo;
-import java.util.ArrayList;
-
-
-public class DriverPropertyInfoHelper{
-
- public static final String AUTO_CONNECT_RETRY = "autoConnectRetry";
-
- public static final String CONNECTIONS_PER_HOST = "connecionsPerHost";
-
- public static final String CONNECT_TIMEOUT = "connectTimeout";
-
- public static final String CURSOR_FINALIZER_ENABLED = "cursorFinalizerEnabled";
-
- public static final String MAX_AUTO_CONNECT_RETRY_TIME = "maxAutoConnectRetryTime";
-
- public static final String READ_PREFERENCE = "readPreference";
-
- public static final String SOCKET_TIMEOUT = "socketTimeout";
-
- public DriverPropertyInfo[] getPropertyInfo()
- {
- ArrayList propInfos = new ArrayList();
-
- addPropInfo(
- propInfos,
- AUTO_CONNECT_RETRY,
- "false",
- "If true, the driver will keep trying to connect to the same server in case that the socket "
- + "cannot be established. There is maximum amount of time to keep retrying, which is 15s by "
- + "default.", null);
-
- addPropInfo(propInfos, CONNECTIONS_PER_HOST, "10", "The maximum number of connections allowed per "
- + "host for this Mongo instance. Those connections will be kept in a pool when idle.", null);
-
- addPropInfo(propInfos, CONNECT_TIMEOUT, "10000", "The connection timeout in milliseconds. ", null);
-
- addPropInfo(propInfos, CURSOR_FINALIZER_ENABLED, "true", "Sets whether there is a a finalize "
- + "method created that cleans up instances of DBCursor that the client does not close.",
- null);
-
- addPropInfo(propInfos, MAX_AUTO_CONNECT_RETRY_TIME, "0",
- "The maximum amount of time in MS to spend retrying to open connection to the same server."
- + "Default is 0, which means to use the default 15s if autoConnectRetry is on.", null);
-
- addPropInfo(propInfos, READ_PREFERENCE, "primary",
- "represents preferred replica set members to which a query or command can be sent", new String[] {
- "primary", "primary preferred", "secondary", "secondary preferred", "nearest" });
-
- addPropInfo(propInfos, SOCKET_TIMEOUT, "0", "The socket timeout in milliseconds It is used for "
- + "I/O socket read and write operations "
- + "Socket.setSoTimeout(int) Default is 0 and means no timeout.", null);
-
- return propInfos.toArray(new DriverPropertyInfo[propInfos.size()]);
- }
-
- private void addPropInfo(final ArrayList propInfos, final String propName,
- final String defaultVal, final String description, final String[] choices)
- {
- DriverPropertyInfo newProp = new DriverPropertyInfo(propName, defaultVal);
- newProp.description = description;
- if (choices != null)
- {
- newProp.choices = choices;
- }
- propInfos.add(newProp);
- }
+package io.mycat.backend.jdbc.mongodb;
+
+import java.sql.DriverPropertyInfo;
+import java.util.ArrayList;
+
+
+public class DriverPropertyInfoHelper{
+
+ public static final String AUTO_CONNECT_RETRY = "autoConnectRetry";
+
+ public static final String CONNECTIONS_PER_HOST = "connecionsPerHost";
+
+ public static final String CONNECT_TIMEOUT = "connectTimeout";
+
+ public static final String CURSOR_FINALIZER_ENABLED = "cursorFinalizerEnabled";
+
+ public static final String MAX_AUTO_CONNECT_RETRY_TIME = "maxAutoConnectRetryTime";
+
+ public static final String READ_PREFERENCE = "readPreference";
+
+ public static final String SOCKET_TIMEOUT = "socketTimeout";
+
+ public DriverPropertyInfo[] getPropertyInfo()
+ {
+ ArrayList propInfos = new ArrayList();
+
+ addPropInfo(
+ propInfos,
+ AUTO_CONNECT_RETRY,
+ "false",
+ "If true, the driver will keep trying to connect to the same server in case that the socket "
+ + "cannot be established. There is maximum amount of time to keep retrying, which is 15s by "
+ + "default.", null);
+
+ addPropInfo(propInfos, CONNECTIONS_PER_HOST, "10", "The maximum number of connections allowed per "
+ + "host for this Mongo instance. Those connections will be kept in a pool when idle.", null);
+
+ addPropInfo(propInfos, CONNECT_TIMEOUT, "10000", "The connection timeout in milliseconds. ", null);
+
+ addPropInfo(propInfos, CURSOR_FINALIZER_ENABLED, "true", "Sets whether there is a a finalize "
+ + "method created that cleans up instances of DBCursor that the client does not close.",
+ null);
+
+ addPropInfo(propInfos, MAX_AUTO_CONNECT_RETRY_TIME, "0",
+ "The maximum amount of time in MS to spend retrying to open connection to the same server."
+ + "Default is 0, which means to use the default 15s if autoConnectRetry is on.", null);
+
+ addPropInfo(propInfos, READ_PREFERENCE, "primary",
+ "represents preferred replica set members to which a query or command can be sent", new String[] {
+ "primary", "primary preferred", "secondary", "secondary preferred", "nearest" });
+
+ addPropInfo(propInfos, SOCKET_TIMEOUT, "0", "The socket timeout in milliseconds It is used for "
+ + "I/O socket read and write operations "
+ + "Socket.setSoTimeout(int) Default is 0 and means no timeout.", null);
+
+ return propInfos.toArray(new DriverPropertyInfo[propInfos.size()]);
+ }
+
+ private void addPropInfo(final ArrayList propInfos, final String propName,
+ final String defaultVal, final String description, final String[] choices)
+ {
+ DriverPropertyInfo newProp = new DriverPropertyInfo(propName, defaultVal);
+ newProp.description = description;
+ if (choices != null)
+ {
+ newProp.choices = choices;
+ }
+ propInfos.add(newProp);
+ }
}
\ No newline at end of file
diff --git a/src/main/java/io/mycat/backend/jdbc/mongodb/MongoClientPropertyHelper.java b/src/main/java/io/mycat/backend/jdbc/mongodb/MongoClientPropertyHelper.java
new file mode 100644
index 000000000..bea00ae65
--- /dev/null
+++ b/src/main/java/io/mycat/backend/jdbc/mongodb/MongoClientPropertyHelper.java
@@ -0,0 +1,35 @@
+package io.mycat.backend.jdbc.mongodb;
+
+
+import com.google.common.base.Joiner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * @author liuxinsi
+ * @mail akalxs@gmail.com
+ */
+public class MongoClientPropertyHelper {
+ /**
+ * 格式化pro中的属性为{@link com.mongodb.MongoClientURI}中要求的格式。
+ *
+ * @param pro 配置参数
+ * @return 格式化后的字符串
+ */
+ public static String formatProperties(Properties pro) {
+ if (pro == null || pro.isEmpty()) {
+ return null;
+ }
+
+ Set