wyl/hertzbeat
1
0
Fork 0
mirror of https://github.com/dromara/hertzbeat.git synced 2026-09-17 09:40:58 +00:00
Code Issues Projects Releases Wiki Activity

remove sleep, probably busy-waiting (#1456)

Browse Source 
Signed-off-by: tomsun28 <tomsun28@outlook.com>
This commit is contained in:
tomsun28
2023-12-29 17:30:44 +08:00
committed by GitHub
parent fbcdd45fee
commit 2a0b7c19f8
17 changed files with 135 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -4
View File
@@ -413,8 +413,6 @@ HertzBeat is a top project under the [Dromara Open Source Community](https://dro
[Reddit Community](https://www.reddit.com/r/hertzbeat/)
[User Club](https://support.qq.com/products/379369)
[Follow Us Twitter](https://twitter.com/hertzbeat1024)
[Subscribe YouTube](https://www.youtube.com/channel/UCri75zfWX0GHqJFPENEbLow)
@@ -441,8 +439,7 @@ HertzBeat is a top project under the [Dromara Open Source Community](https://dro
##### Sponsor
- Thanks [吉实信息(构建全新的微波+光交易网络)](https://www.flarespeed.com) sponsored server node.
- Thanks [蓝易云(全新智慧上云)](https://www.tsyvps.com/aff/BZBEGYLX) sponsored server node.
- Thanks [吉实信息(构建全新的微波+光交易网络)](https://www.flarespeed.com) sponsored server node.
##### Open-Source Project Build From Open-Source
README_CN.md
+1 -4
View File
@@ -418,8 +418,6 @@ HertzBeat 赫兹跳动是 [Dromara开源社区](https://dromara.org/) 下顶级
[Reddit Community](https://www.reddit.com/r/hertzbeat/)
[User Club](https://support.qq.com/products/379369)
[Follow Us Twitter](https://twitter.com/hertzbeat1024)
[Subscribe YouTube](https://www.youtube.com/channel/UCri75zfWX0GHqJFPENEbLow)
@@ -446,8 +444,7 @@ HertzBeat 赫兹跳动是 [Dromara开源社区](https://dromara.org/) 下顶级
##### 赞助
- 感谢 [吉实信息(构建全新的微波+光交易网络)](https://www.flarespeed.com) 赞助服务器采集节点
- 感谢 [蓝易云(全新智慧上云)](https://www.tsyvps.com/aff/BZBEGYLX) 赞助服务器采集节点
- 感谢 [吉实信息(构建全新的微波+光交易网络)](https://www.flarespeed.com) 赞助服务器采集节点
##### Open-Source Project Build From Open-Source
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/CommonCache.java → collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/ConnectionCommonCache.java
Vendored
+32 -35
View File
@@ -18,6 +18,7 @@
package org.dromara.hertzbeat.collector.collect.common.cache;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
import lombok.extern.slf4j.Slf4j;
@@ -26,20 +27,20 @@ import java.util.Optional;
import java.util.concurrent.*;
/**
* lru common resource cache
* lru common resource cache for client-server connection
*
*
*/
@Slf4j
public class CommonCache {
public class ConnectionCommonCache {
/**
* default cache time 800s
* default cache time 200s
*/
private static final long DEFAULT_CACHE_TIMEOUT = 800 * 1000L;
private static final long DEFAULT_CACHE_TIMEOUT = 200 * 1000L;
/**
* default cache num
* default max cache num
*/
private static final int DEFAULT_MAX_CAPACITY = 10000;
@@ -61,9 +62,9 @@ public class CommonCache {
/**
* the executor who clean cache when timeout
*/
private ThreadPoolExecutor cleanTimeoutExecutor;
private ThreadPoolExecutor timeoutCleanerExecutor;
private CommonCache() {
private ConnectionCommonCache() {
init();
}
@@ -76,23 +77,25 @@ public class CommonCache {
if (value instanceof CacheCloseable) {
((CacheCloseable) value).close();
}
log.info("lru cache discard key: {}, value: {}.", key, value);
log.info("connection common cache discard key: {}, value: {}.", key, value);
}).build();
timeoutMap = new ConcurrentHashMap<>(DEFAULT_MAX_CAPACITY >> 6);
cleanTimeoutExecutor = new ThreadPoolExecutor(1, 1,
1, TimeUnit.SECONDS,
// last-first-coverage algorithm, run the first and last thread, discard mid
timeoutCleanerExecutor = new ThreadPoolExecutor(1, 1, 1, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1),
r -> new Thread(r, "lru-cache-timeout-cleaner"),
r -> new Thread(r, "connection-cache-timeout-cleaner"),
new ThreadPoolExecutor.DiscardOldestPolicy());
// init monitor available detector cyc task
ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1,
r -> new Thread(r, "lru-cache-available-detector"));
scheduledExecutor.scheduleWithFixedDelay(this::detectCacheAvailable,
2, 20, TimeUnit.MINUTES);
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setNameFormat("connection-cache-ava-detector-%d")
.setDaemon(true)
.build();
ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1, threadFactory);
scheduledExecutor.scheduleWithFixedDelay(this::detectCacheAvailable, 2, 20, TimeUnit.MINUTES);
}
/**
* detect all cache available, cleanup not ava object
* detect all cache available, cleanup not ava connection
*/
private void detectCacheAvailable() {
try {
@@ -110,7 +113,7 @@ public class CommonCache {
}
});
} catch (Exception e) {
log.error("detect cache available error: {}.", e.getMessage(), e);
log.error("connection common cache detect cache available error: {}.", e.getMessage(), e);
}
}
@@ -127,17 +130,18 @@ public class CommonCache {
timeoutMap.put(key, new Long[]{currentTime, DEFAULT_CACHE_TIMEOUT});
} else if (cacheTime[0] + cacheTime[1] < currentTime) {
// timeout, remove this object cache
log.warn("[cache] clean the timeout cache, key {}", key);
log.warn("[connection common cache] clean the timeout cache, key {}", key);
timeoutMap.remove(key);
cacheMap.remove(key);
if (value instanceof CacheCloseable) {
log.warn("[cache] close the timeout cache, key {}", key);
log.warn("[connection common cache] close the timeout cache, key {}", key);
((CacheCloseable) value).close();
}
}
});
Thread.sleep(20 * 1000);
} catch (Exception e) {
log.error("[cache] clean timeout cache error: {}.", e.getMessage(), e);
log.error("[connection common cache] clean timeout cache error: {}.", e.getMessage(), e);
}
}
@@ -155,14 +159,7 @@ public class CommonCache {
}
cacheMap.put(key, value);
timeoutMap.put(key, new Long[]{System.currentTimeMillis(), timeDiff});
cleanTimeoutExecutor.execute(() -> {
try {
cleanTimeoutCache();
Thread.sleep(10 * 1000);
} catch (InterruptedException e) {
log.error(e.getMessage(), e);
}
});
timeoutCleanerExecutor.execute(this::cleanTimeoutCache);
}
/**
@@ -185,18 +182,18 @@ public class CommonCache {
public Optional<Object> getCache(Object key, boolean refreshCache) {
Long[] cacheTime = timeoutMap.get(key);
if (cacheTime == null || cacheTime.length != CACHE_TIME_LENGTH) {
log.info("[cache] not hit the cache, key {}.", key);
log.info("[connection common cache] not hit the cache, key {}.", key);
return Optional.empty();
}
if (cacheTime[0] + cacheTime[1] < System.currentTimeMillis()) {
log.warn("[cache] is timeout, remove it, key {}.", key);
log.warn("[connection common cache] is timeout, remove it, key {}.", key);
timeoutMap.remove(key);
cacheMap.remove(key);
return Optional.empty();
}
Object value = cacheMap.get(key);
if (value == null) {
log.error("[cache] value is null, remove it, key {}.", key);
log.error("[connection common cache] value is null, remove it, key {}.", key);
cacheMap.remove(key);
timeoutMap.remove(key);
} else if (refreshCache) {
@@ -222,16 +219,16 @@ public class CommonCache {
/**
* get common cache instance
*
* @return cache
* @return connection common cache
*/
public static CommonCache getInstance() {
public static ConnectionCommonCache getInstance() {
return SingleInstance.INSTANCE;
}
/**
* static instance
* static single instance
*/
private static class SingleInstance {
private static final CommonCache INSTANCE = new CommonCache();
private static final ConnectionCommonCache INSTANCE = new ConnectionCommonCache();
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/JdbcConnect.java
Vendored
+2 -2
View File
@@ -28,7 +28,7 @@ import java.sql.Connection;
@Slf4j
public class JdbcConnect implements CacheCloseable {
private Connection connection;
private final Connection connection;
public JdbcConnect(Connection connection) {
this.connection = connection;
@@ -41,7 +41,7 @@ public class JdbcConnect implements CacheCloseable {
connection.close();
}
} catch (Exception e) {
log.error("close jdbc connect error: {}", e.getMessage());
log.error("[connection common cache] close jdbc connect error: {}", e.getMessage());
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/JmxConnect.java
Vendored
+2 -2
View File
@@ -11,7 +11,7 @@ import javax.management.remote.JMXConnector;
@Slf4j
public class JmxConnect implements CacheCloseable {
private JMXConnector connection;
private final JMXConnector connection;
public JmxConnect(JMXConnector connection) {
this.connection = connection;
@@ -25,7 +25,7 @@ public class JmxConnect implements CacheCloseable {
connection.close();
}
} catch (Exception e) {
log.error("close redis connect error: {}", e.getMessage());
log.error("[connection common cache] close jmx connect error: {}", e.getMessage());
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/MongodbConnect.java
Vendored
+2 -2
View File
@@ -9,7 +9,7 @@ import lombok.extern.slf4j.Slf4j;
*/
@Slf4j
public class MongodbConnect implements CacheCloseable {
private MongoClient mongoClient;
private final MongoClient mongoClient;
public MongodbConnect(MongoClient mongoClient) {
this.mongoClient = mongoClient;
@@ -22,7 +22,7 @@ public class MongodbConnect implements CacheCloseable {
this.mongoClient.close();
}
} catch (Exception e) {
log.error(e.getMessage(), e);
log.error("[connection common cache] close mongodb connect error: {}", e.getMessage());
}
}
public MongoClient getMongoClient() {
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/RedisConnect.java
Vendored
+2 -2
View File
@@ -27,7 +27,7 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class RedisConnect implements CacheCloseable {
private StatefulConnection<String, String> connection;
private final StatefulConnection<String, String> connection;
public RedisConnect(StatefulConnection<String, String> connection) {
this.connection = connection;
@@ -40,7 +40,7 @@ public class RedisConnect implements CacheCloseable {
connection.closeAsync();
}
} catch (Exception e) {
log.error("close redis connect error: {}", e.getMessage());
log.error("[connection common cache] close redis connect error: {}", e.getMessage());
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/cache/SshConnect.java
Vendored
+2 -2
View File
@@ -9,7 +9,7 @@ import org.apache.sshd.client.session.ClientSession;
*/
@Slf4j
public class SshConnect implements CacheCloseable {
private ClientSession clientSession;
private final ClientSession clientSession;
public SshConnect(ClientSession clientSession) {
this.clientSession = clientSession;
@@ -22,7 +22,7 @@ public class SshConnect implements CacheCloseable {
clientSession.close();
}
} catch (Exception e) {
log.error("close ssh connect error: {}", e.getMessage());
log.error("[connection common cache] close ssh connect error: {}", e.getMessage());
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/common/http/CommonHttpClient.java
+20 -23
View File
@@ -17,6 +17,7 @@
package org.dromara.hertzbeat.collector.collect.common.http;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.config.Registry;
@@ -37,6 +38,9 @@ import java.security.cert.CertificateException;
import java.security.cert.CertificateExpiredException;
import java.security.cert.X509Certificate;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
/**
@@ -76,8 +80,7 @@ public class CommonHttpClient {
private static final int SOCKET_TIMEOUT = 60000;
/**
* validated time for idle connection
* 空闲连接免检的有效时间，被重用的空闲连接若超过此时间，需检查此连接的可用性
* validated time for idle connection. if when reuse this connection after this time, we will check it available.
*/
private static final int INACTIVITY_VALIDATED_TIME = 10000;
@@ -91,13 +94,12 @@ public class CommonHttpClient {
SSLContext sslContext = SSLContexts.createDefault();
X509TrustManager x509TrustManager = new X509TrustManager() {
@Override
public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException { }
public void checkClientTrusted(X509Certificate[] x509Certificates, String s) { }
@Override
public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
// check server certificate timeout
// 判断服务器证书有效时间
// check server ssl certificate expired
Date now = new Date();
if (x509Certificates != null && x509Certificates.length > 0) {
if (x509Certificates != null) {
for (X509Certificate certificate : x509Certificates) {
Date deadline = certificate.getNotAfter();
if (deadline != null && now.after(deadline)) {
@@ -130,26 +132,21 @@ public class CommonHttpClient {
httpClient = HttpClients.custom()
.setConnectionManager(connectionManager)
.setDefaultRequestConfig(requestConfig)
// 定期清理不可用过期连接
// clean up unavailable expired connections
.evictExpiredConnections()
// 定期清理可用但空闲的连接
// clean up available but idle connections
.evictIdleConnections(100, TimeUnit.SECONDS)
.build();
Thread connectCleaner = new Thread(() -> {
while (Thread.currentThread().isInterrupted()) {
try {
Thread.sleep(30000);
connectionManager.closeExpiredConnections();
connectionManager.closeIdleConnections(100, TimeUnit.SECONDS);
} catch (InterruptedException e) {
}
}
});
connectCleaner.setName("http-connection-pool-cleaner");
connectCleaner.setDaemon(true);
connectCleaner.start();
} catch (Exception e) {
}
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setNameFormat("http-connection-pool-cleaner-%d")
.setDaemon(true)
.build();
ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1, threadFactory);
scheduledExecutor.scheduleWithFixedDelay(() -> {
connectionManager.closeExpiredConnections();
connectionManager.closeIdleConnections(100, TimeUnit.SECONDS);
}, 40L, 40L, TimeUnit.SECONDS);
} catch (Exception ignored) {}
}
public static CloseableHttpClient getHttpClient() {
collector/src/main/java/org/dromara/hertzbeat/collector/collect/database/JdbcCommonCollect.java
+4 -4
View File
@@ -20,7 +20,7 @@ package org.dromara.hertzbeat.collector.collect.database;
import com.mysql.cj.jdbc.exceptions.CommunicationsException;
import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import org.dromara.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.dromara.hertzbeat.collector.collect.common.cache.CommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.ConnectionCommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.JdbcConnect;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
import org.dromara.hertzbeat.collector.util.CollectUtil;
@@ -138,7 +138,7 @@ public class JdbcCommonCollect extends AbstractCollect {
CacheIdentifier identifier = CacheIdentifier.builder()
.ip(url)
.username(username).password(password).build();
Optional<Object> cacheOption = CommonCache.getInstance().getCache(identifier, true);
Optional<Object> cacheOption = ConnectionCommonCache.getInstance().getCache(identifier, true);
Statement statement = null;
if (cacheOption.isPresent()) {
JdbcConnect jdbcConnect = (JdbcConnect) cacheOption.get();
@@ -161,7 +161,7 @@ public class JdbcCommonCollect extends AbstractCollect {
log.error(e2.getMessage());
}
statement = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
}
if (statement != null) {
@@ -175,7 +175,7 @@ public class JdbcCommonCollect extends AbstractCollect {
statement.setQueryTimeout(timeoutSecond);
statement.setMaxRows(1000);
JdbcConnect jdbcConnect = new JdbcConnect(connection);
CommonCache.getInstance().addCache(identifier, jdbcConnect);
ConnectionCommonCache.getInstance().addCache(identifier, jdbcConnect);
return statement;
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/jmx/JmxCollectImpl.java
+4 -4
View File
@@ -2,7 +2,7 @@ package org.dromara.hertzbeat.collector.collect.jmx;
import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import org.dromara.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.dromara.hertzbeat.collector.collect.common.cache.CommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.ConnectionCommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.JmxConnect;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
import org.dromara.hertzbeat.common.entity.job.Metrics;
@@ -142,7 +142,7 @@ public class JmxCollectImpl extends AbstractCollect {
CacheIdentifier identifier = CacheIdentifier.builder().ip(jmxProtocol.getHost())
.port(jmxProtocol.getPort()).username(jmxProtocol.getUsername())
.password(jmxProtocol.getPassword()).build();
Optional<Object> cacheOption = CommonCache.getInstance().getCache(identifier, true);
Optional<Object> cacheOption = ConnectionCommonCache.getInstance().getCache(identifier, true);
JMXConnector conn = null;
if (cacheOption.isPresent()) {
JmxConnect jmxConnect = (JmxConnect) cacheOption.get();
@@ -151,7 +151,7 @@ public class JmxCollectImpl extends AbstractCollect {
conn.getMBeanServerConnection();
} catch (Exception e) {
conn = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
}
if (conn != null) {
@@ -176,7 +176,7 @@ public class JmxCollectImpl extends AbstractCollect {
}
JMXServiceURL jmxServiceUrl = new JMXServiceURL(url);
conn = JMXConnectorFactory.connect(jmxServiceUrl, environment);
CommonCache.getInstance().addCache(identifier, new JmxConnect(conn));
ConnectionCommonCache.getInstance().addCache(identifier, new JmxConnect(conn));
return conn;
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/mongodb/MongodbSingleCollectImpl.java
+4 -4
View File
@@ -25,7 +25,7 @@ import java.util.Optional;
import com.mongodb.MongoServerUnavailableException;
import com.mongodb.MongoTimeoutException;
import org.dromara.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.dromara.hertzbeat.collector.collect.common.cache.CommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.ConnectionCommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.MongodbConnect;
import org.dromara.hertzbeat.common.util.CommonUtil;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
@@ -176,7 +176,7 @@ public class MongodbSingleCollectImpl extends AbstractCollect {
CacheIdentifier identifier = CacheIdentifier.builder()
.ip(mongodbProtocol.getHost()).port(mongodbProtocol.getPort())
.username(mongodbProtocol.getUsername()).password(mongodbProtocol.getPassword()).build();
Optional<Object> cacheOption = CommonCache.getInstance().getCache(identifier, true);
Optional<Object> cacheOption = ConnectionCommonCache.getInstance().getCache(identifier, true);
MongoClient mongoClient = null;
if (cacheOption.isPresent()) {
MongodbConnect mongodbConnect = (MongodbConnect) cacheOption.get();
@@ -192,7 +192,7 @@ public class MongodbSingleCollectImpl extends AbstractCollect {
log.error(e2.getMessage());
}
mongoClient = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
}
if (mongoClient != null) {
@@ -206,7 +206,7 @@ public class MongodbSingleCollectImpl extends AbstractCollect {
mongodbProtocol.getDatabase(), mongodbProtocol.getAuthenticationDatabase());
mongoClient = MongoClients.create(url);
MongodbConnect mongodbConnect = new MongodbConnect(mongoClient);
CommonCache.getInstance().addCache(identifier, mongodbConnect);
ConnectionCommonCache.getInstance().addCache(identifier, mongodbConnect);
return mongoClient;
}
}
collector/src/main/java/org/dromara/hertzbeat/collector/collect/redis/RedisCommonCollectImpl.java
+5 -5
View File
@@ -2,7 +2,7 @@ package org.dromara.hertzbeat.collector.collect.redis;
import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import org.dromara.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.dromara.hertzbeat.collector.collect.common.cache.CommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.ConnectionCommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.RedisConnect;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
import org.dromara.hertzbeat.collector.util.CollectUtil;
@@ -166,7 +166,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
// reuse connection failed, new one
RedisClient redisClient = buildSingleClient(redisProtocol);
connection = redisClient.connect();
CommonCache.getInstance().addCache(identifier, new RedisConnect(connection));
ConnectionCommonCache.getInstance().addCache(identifier, new RedisConnect(connection));
}
return connection;
}
@@ -204,7 +204,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
// reuse connection failed, new one
RedisClusterClient redisClusterClient = buildClusterClient(redisProtocol);
connection = redisClusterClient.connect();
CommonCache.getInstance().addCache(identifier, new RedisConnect(connection));
ConnectionCommonCache.getInstance().addCache(identifier, new RedisConnect(connection));
}
return connection;
}
@@ -217,7 +217,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
*/
private StatefulConnection<String, String> getStatefulConnection(CacheIdentifier identifier) {
StatefulConnection<String, String> connection = null;
Optional<Object> cacheOption = CommonCache.getInstance().getCache(identifier, true);
Optional<Object> cacheOption = ConnectionCommonCache.getInstance().getCache(identifier, true);
if (cacheOption.isPresent()) {
RedisConnect redisConnect = (RedisConnect) cacheOption.get();
connection = redisConnect.getConnection();
@@ -228,7 +228,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
log.info("The redis connect form cache, close error: {}", e.getMessage());
}
connection = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
}
return connection;
collector/src/main/java/org/dromara/hertzbeat/collector/collect/ssh/SshCollectImpl.java
+6 -6
View File
@@ -23,7 +23,7 @@ import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
import org.apache.sshd.common.util.security.SecurityUtils;
import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import org.dromara.hertzbeat.collector.collect.common.cache.CacheIdentifier;
import org.dromara.hertzbeat.collector.collect.common.cache.CommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.ConnectionCommonCache;
import org.dromara.hertzbeat.collector.collect.common.cache.SshConnect;
import org.dromara.hertzbeat.collector.collect.common.ssh.CommonSshClient;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
@@ -284,7 +284,7 @@ public class SshCollectImpl extends AbstractCollect {
.ip(sshProtocol.getHost()).port(sshProtocol.getPort())
.username(sshProtocol.getUsername()).password(sshProtocol.getPassword())
.build();
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
private ClientSession getConnectSession(SshProtocol sshProtocol, int timeout, boolean reuseConnection)
@@ -295,18 +295,18 @@ public class SshCollectImpl extends AbstractCollect {
.build();
ClientSession clientSession = null;
if (reuseConnection) {
Optional<Object> cacheOption = CommonCache.getInstance().getCache(identifier, true);
Optional<Object> cacheOption = ConnectionCommonCache.getInstance().getCache(identifier, true);
if (cacheOption.isPresent()) {
clientSession = ((SshConnect) cacheOption.get()).getConnection();
try {
if (clientSession == null || clientSession.isClosed() || clientSession.isClosing()) {
clientSession = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
} catch (Exception e) {
log.warn(e.getMessage());
clientSession = null;
CommonCache.getInstance().removeCache(identifier);
ConnectionCommonCache.getInstance().removeCache(identifier);
}
}
if (clientSession != null) {
@@ -331,7 +331,7 @@ public class SshCollectImpl extends AbstractCollect {
}
if (reuseConnection) {
SshConnect sshConnect = new SshConnect(clientSession);
CommonCache.getInstance().addCache(identifier, sshConnect);
ConnectionCommonCache.getInstance().addCache(identifier, sshConnect);
}
return clientSession;
}
collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/CommonDispatcher.java
+45 -73
View File
@@ -17,12 +17,12 @@
package org.dromara.hertzbeat.collector.dispatch;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.dromara.hertzbeat.collector.dispatch.entrance.internal.CollectJobService;
import org.dromara.hertzbeat.collector.dispatch.timer.Timeout;
import org.dromara.hertzbeat.collector.dispatch.timer.TimerDispatch;
@@ -34,15 +34,10 @@ import org.dromara.hertzbeat.common.entity.job.Job;
import org.dromara.hertzbeat.common.entity.job.Metrics;
import org.dromara.hertzbeat.common.entity.message.CollectRep;
import org.dromara.hertzbeat.common.queue.CommonDataQueue;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -52,7 +47,7 @@ import java.util.concurrent.atomic.AtomicReference;
*/
@Component
@Slf4j
public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatch, DisposableBean {
public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatch {
/**
* Collection task timeout value
@@ -86,8 +81,6 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
private final List<UnitConvert> unitConvertList;
private final ThreadPoolExecutor poolExecutor;
private final WorkerPool workerPool;
private final String collectorIdentity;
@@ -105,20 +98,13 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
this.workerPool = workerPool;
this.collectorIdentity = collectJobService.getCollectorIdentity();
this.metricsTimeoutMonitorMap = new ConcurrentHashMap<>(16);
poolExecutor = new ThreadPoolExecutor(2, 2, 1,
TimeUnit.SECONDS,
new SynchronousQueue<>(), r -> {
Thread thread = new Thread(r);
thread.setDaemon(true);
return thread;
});
this.start();
}
public void start() {
try {
// Pull the collection task from the task queue and put it into the thread pool for execution
poolExecutor.execute(() -> {
workerPool.executeJob(() -> {
Thread.currentThread().setName("metrics-task-dispatcher");
while (!Thread.currentThread().isInterrupted()) {
MetricsCollect metricsCollect = null;
@@ -128,17 +114,10 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
workerPool.executeJob(metricsCollect);
}
} catch (RejectedExecutionException rejected) {
log.info("[Dispatcher]-the worker pool is full, reject this metrics task, " +
"sleep and put in queue again.");
try {
Thread.sleep(1000);
if (metricsCollect != null) {
metricsCollect.setRunPriority((byte) (metricsCollect.getRunPriority() + 1));
jobRequestQueue.addJob(metricsCollect);
}
} catch (InterruptedException ignored) {
log.info("[Dispatcher]-metrics-task-dispatcher has been interrupt when sleep to close.");
Thread.currentThread().interrupt();
log.info("[Dispatcher]-the worker pool is full, reject this metrics task，put in queue again.");
if (metricsCollect != null) {
metricsCollect.setRunPriority((byte) (metricsCollect.getRunPriority() + 1));
jobRequestQueue.addJob(metricsCollect);
}
} catch (InterruptedException interruptedException) {
log.info("[Dispatcher]-metrics-task-dispatcher has been interrupt to close.");
@@ -148,47 +127,47 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
}
});
// Monitoring metrics collection task execution time
poolExecutor.execute(() -> {
Thread.currentThread().setName("metrics-task-monitor");
while (!Thread.currentThread().isInterrupted()) {
try {
// Detect whether the collection unit of each metrics has timed out for 4 minutes,
// and if it times out, it will be discarded and an exception will be returned.
long deadline = System.currentTimeMillis() - DURATION_TIME;
for (Map.Entry<String, MetricsTime> entry : metricsTimeoutMonitorMap.entrySet()) {
MetricsTime metricsTime = entry.getValue();
if (metricsTime.getStartTime() < deadline) {
// Metrics collection timeout
WheelTimerTask timerJob = (WheelTimerTask) metricsTime.getTimeout().task();
CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
.setId(timerJob.getJob().getMonitorId())
.setTenantId(timerJob.getJob().getTenantId())
.setApp(timerJob.getJob().getApp())
.setMetrics(metricsTime.getMetrics().getName())
.setPriority(metricsTime.getMetrics().getPriority())
.setTime(System.currentTimeMillis())
.setCode(CollectRep.Code.TIMEOUT).setMsg("collect timeout").build();
log.error("[Collect Timeout]: \n{}", metricsData);
if (metricsData.getPriority() == 0) {
dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData);
}
metricsTimeoutMonitorMap.remove(entry.getKey());
}
}
Thread.sleep(20000);
} catch (InterruptedException interruptedException) {
log.info("[Dispatcher]-metrics-task-monitor has been interrupt to close.");
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error("[Task Monitor]-{}.", e.getMessage(), e);
}
}
});
// monitoring metrics collection task execution timeout
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setNameFormat("metrics-task-timeout-monitor-%d")
.setDaemon(true)
.build();
ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1, threadFactory);
scheduledExecutor.scheduleWithFixedDelay(this::monitorCollectTaskTimeout, 2, 20, TimeUnit.SECONDS);
} catch (Exception e) {
log.error("Common Dispatcher error: {}.", e.getMessage(), e);
}
}
private void monitorCollectTaskTimeout() {
try {
// Detect whether the collection unit of each metrics has timed out for 4 minutes,
// and if it times out, it will be discarded and an exception will be returned.
long deadline = System.currentTimeMillis() - DURATION_TIME;
for (Map.Entry<String, MetricsTime> entry : metricsTimeoutMonitorMap.entrySet()) {
MetricsTime metricsTime = entry.getValue();
if (metricsTime.getStartTime() < deadline) {
// Metrics collection timeout
WheelTimerTask timerJob = (WheelTimerTask) metricsTime.getTimeout().task();
CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
.setId(timerJob.getJob().getMonitorId())
.setTenantId(timerJob.getJob().getTenantId())
.setApp(timerJob.getJob().getApp())
.setMetrics(metricsTime.getMetrics().getName())
.setPriority(metricsTime.getMetrics().getPriority())
.setTime(System.currentTimeMillis())
.setCode(CollectRep.Code.TIMEOUT).setMsg("collect timeout").build();
log.error("[Collect Timeout]: \n{}", metricsData);
if (metricsData.getPriority() == 0) {
dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData);
}
metricsTimeoutMonitorMap.remove(entry.getKey());
}
}
} catch (Exception e) {
log.error("[Task Timeout Monitor]-{}.", e.getMessage(), e);
}
}
@Override
public void dispatchMetricsTask(Timeout timeout) {
@@ -383,13 +362,6 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
return mapList;
}
@Override
public void destroy() throws Exception {
if (poolExecutor != null) {
poolExecutor.shutdownNow();
}
}
@Data
@AllArgsConstructor
private static class MetricsTime {
collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/MetricsCollectorQueue.java
+1 -1
View File
@@ -33,7 +33,7 @@ public class MetricsCollectorQueue {
private final PriorityBlockingQueue<MetricsCollect> jobQueue;
public MetricsCollectorQueue() {
jobQueue = new PriorityBlockingQueue<>(2000);
jobQueue = new PriorityBlockingQueue<>();
}
public void addJob(MetricsCollect job) {
collector/src/test/java/org/dromara/hertzbeat/collector/collect/common/cache/CommonCacheTest.java
Vendored
+2 -2
View File
@@ -5,7 +5,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/**
* Test case for {@link CommonCache}
* Test case for {@link ConnectionCommonCache}
*/
class CommonCacheTest {
@@ -36,4 +36,4 @@ class CommonCacheTest {
@Test
void getInstance() {
}
}
}