lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DataJdbcSource优化

Browse Source
This commit is contained in:
redkale
2023-05-06 08:21:27 +08:00
parent 1b1bfbb3ac
commit 18932bf48b
2 changed files with 25 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
src/main/java/org/redkale/source/DataJdbcSource.java
View File

@@ -2696,9 +2696,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
protected String url; protected String url;
protected int urlVersion; protected final AtomicInteger urlVersion = new AtomicInteger();
protected Properties clientInfo = new Properties();
public ConnectionPool(Properties prop) { public ConnectionPool(Properties prop) {
this.connectTimeoutSeconds = Integer.decode(prop.getProperty(DATA_SOURCE_CONNECTTIMEOUT_SECONDS, "30")); this.connectTimeoutSeconds = Integer.decode(prop.getProperty(DATA_SOURCE_CONNECTTIMEOUT_SECONDS, "30"));
@@ -2706,7 +2704,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
if (workExecutor instanceof ThreadPoolExecutor) { if (workExecutor instanceof ThreadPoolExecutor) {
defMaxConns = ((ThreadPoolExecutor) workExecutor).getCorePoolSize(); defMaxConns = ((ThreadPoolExecutor) workExecutor).getCorePoolSize();
} else if (workExecutor != null) { //maybe virtual thread pool } else if (workExecutor != null) { //maybe virtual thread pool
defMaxConns = Math.min(1024, Utility.cpus() * 100); defMaxConns = Math.min(1000, Utility.cpus() * 100);
} }
this.maxConns = Math.max(1, Integer.decode(prop.getProperty(DATA_SOURCE_MAXCONNS, "" + defMaxConns))); this.maxConns = Math.max(1, Integer.decode(prop.getProperty(DATA_SOURCE_MAXCONNS, "" + defMaxConns)));
this.canNewSemaphore = new Semaphore(this.maxConns); this.canNewSemaphore = new Semaphore(this.maxConns);
@@ -2726,15 +2724,14 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} catch (SQLException e) { } catch (SQLException e) {
throw new SourceException(e); throw new SourceException(e);
} }
clientInfo.put("version", String.valueOf(urlVersion));
resetMaxConnection(); resetMaxConnection();
} }
@ResourceListener @ResourceListener
public void onResourceChange(ResourceEvent[] events) { public void onResourceChange(ResourceEvent[] events) {
String newUrl = this.url;
int newConnectTimeoutSeconds = this.connectTimeoutSeconds; int newConnectTimeoutSeconds = this.connectTimeoutSeconds;
int newMaxconns = this.maxConns; int newMaxconns = this.maxConns;
String newUrl = this.url;
String newUser = this.connectAttrs.getProperty("user"); String newUser = this.connectAttrs.getProperty("user");
String newPassword = this.connectAttrs.getProperty("password"); String newPassword = this.connectAttrs.getProperty("password");
for (ResourceEvent event : events) { for (ResourceEvent event : events) {
@@ -2752,10 +2749,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
if (!Objects.equals(newUser, this.connectAttrs.get("user")) if (!Objects.equals(newUser, this.connectAttrs.get("user"))
|| !Objects.equals(newPassword, this.connectAttrs.get("password")) || !Objects.equals(newUrl, url)) { || !Objects.equals(newPassword, this.connectAttrs.get("password")) || !Objects.equals(newUrl, url)) {
this.urlVersion++; this.urlVersion.incrementAndGet();
Properties newClientInfo = new Properties();
newClientInfo.put("version", String.valueOf(urlVersion));
this.clientInfo = newClientInfo;
} }
this.url = newUrl; this.url = newUrl;
this.connectTimeoutSeconds = newConnectTimeoutSeconds; this.connectTimeoutSeconds = newConnectTimeoutSeconds;
@@ -2802,12 +2796,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
this.canNewSemaphore = new Semaphore(this.maxConns); this.canNewSemaphore = new Semaphore(this.maxConns);
SourceConnection c; SourceConnection c;
while ((c = oldQueue.poll()) != null) { while ((c = oldQueue.poll()) != null) {
try { c.version = -1;
if (c.getClientInfo() != null) {
c.getClientInfo().put("version", "-1");
}
} catch (SQLException e) {
}
offerConnection(c, oldSemaphore); offerConnection(c, oldSemaphore);
} }
} }
@@ -2833,12 +2822,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
SourceConnection conn = null; SourceConnection conn = null;
if (semaphore.tryAcquire()) { if (semaphore.tryAcquire()) {
try { try {
conn = new SourceConnection(driver.connect(url, connectAttrs)); conn = new SourceConnection(driver.connect(url, connectAttrs), this.urlVersion.get());
if (conn.getClientInfo() != null) {
conn.getClientInfo().put("version", clientInfo.getProperty("version"));
} else {
conn.setClientInfo(clientInfo);
}
} catch (SQLException ex) { } catch (SQLException ex) {
throw new SourceException(ex); throw new SourceException(ex);
} }
@@ -2864,34 +2848,25 @@ public class DataJdbcSource extends AbstractDataSqlSource {
private <C> void offerConnection(final C connection, Semaphore semaphore) { private <C> void offerConnection(final C connection, Semaphore semaphore) {
SourceConnection conn = (SourceConnection) connection; SourceConnection conn = (SourceConnection) connection;
if (conn == null) { if (conn != null) {
return; try {
} if (checkValid(conn) && queue.offer(conn)) {
try { usingCounter.decrement();
if (checkValid(conn) && queue.offer(conn)) { } else {
usingCounter.decrement(); usingCounter.decrement();
} else { closeCounter.increment();
usingCounter.decrement(); semaphore.release();
closeCounter.increment(); conn.close();
semaphore.release(); }
conn.close(); } catch (Exception e) {
logger.log(Level.WARNING, "closeSQLConnection abort", e);
} }
} catch (Exception e) {
logger.log(Level.WARNING, "closeSQLConnection abort", e);
} }
} }
protected boolean checkValid(SourceConnection conn) { protected boolean checkValid(SourceConnection conn) {
try { try {
boolean rs = !conn.conn.isClosed() && conn.conn.isValid(1); return !conn.conn.isClosed() && conn.conn.isValid(1) && conn.version == this.urlVersion.get();
if (!rs) {
return rs;
}
Properties prop = conn.getClientInfo();
if (prop == null) {
return false;
}
return prop == clientInfo || Objects.equals(prop.getProperty("version"), clientInfo.getProperty("version"));
} catch (SQLException ex) { } catch (SQLException ex) {
if (!"08S01".equals(ex.getSQLState())) {//MySQL特性， 长时间连接没使用会抛出com.mysql.jdbc.exceptions.jdbc4.CommunicationsException if (!"08S01".equals(ex.getSQLState())) {//MySQL特性， 长时间连接没使用会抛出com.mysql.jdbc.exceptions.jdbc4.CommunicationsException
logger.log(Level.FINER, "result.getConnection from pooled connection abort [" + ex.getSQLState() + "]", ex); logger.log(Level.FINER, "result.getConnection from pooled connection abort [" + ex.getSQLState() + "]", ex);
@@ -2913,21 +2888,14 @@ public class DataJdbcSource extends AbstractDataSqlSource {
protected class SourceConnection { protected class SourceConnection {
public int version;
public final Connection conn; public final Connection conn;
private final Map<String, PreparedStatement> prestms = new ConcurrentHashMap<>(); public SourceConnection(Connection conn, int version) {
public SourceConnection(Connection conn) {
Objects.requireNonNull(conn); Objects.requireNonNull(conn);
this.conn = conn; this.conn = conn;
} this.version = version;
public Properties getClientInfo() throws SQLException {
return conn.getClientInfo();
}
public void setClientInfo(Properties clientInfo) throws SQLException {
conn.setClientInfo(clientInfo);
} }
public Statement createStatement() throws SQLException { public Statement createStatement() throws SQLException {
@@ -2935,16 +2903,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
public PreparedStatement prepareStatement(String sql) throws SQLException { public PreparedStatement prepareStatement(String sql) throws SQLException {
PreparedStatement rs = prestms.computeIfAbsent(sql, s -> { return conn.prepareStatement(sql);
try {
return conn.prepareStatement(sql);
} catch (SQLException e) {
throw new RedkaleException(e);
}
});
rs.clearParameters();
rs.clearBatch();
return rs;
} }
public void setAutoCommit(boolean autoCommit) throws SQLException { public void setAutoCommit(boolean autoCommit) throws SQLException {

2
src/main/java/org/redkale/source/DataResultSet.java
View File

@@ -136,7 +136,7 @@ public interface DataResultSet extends EntityInfo.DataResultSetRow {
if (o instanceof byte[]) { if (o instanceof byte[]) {
o = new BigDecimal(new String((byte[]) o)); o = new BigDecimal(new String((byte[]) o));
} else { } else {
o = new BigInteger(o.toString()); o = new BigDecimal(o.toString());
} }
} }
} else if (t == String.class) { } else if (t == String.class) {