Enhance druid connection pool manage

# Background
现在的seata的rm在和数据库交互的时候使用到了druid作为数据库连接池，通过druid的DruidDataSource对象可以很方便的知道此时的连接池的各项指标，包括当前活跃连接数，空闲连接数，最大连接数等等。但是seata缺少相应的指标收集和监控机制，使得连接池的实时状态难以知晓。

# Proposal
我的大致思路是通过一个定时执行的线程来收集连接池的指标，缓存到一个LinkedBlockingQueue中，并且通过一个sender批量发送到其他服务做进一步的处理和展示。

我编写了一部分代码加以说明：

```java
public class ConnectionPoolService implements Runnable {

private DruidConnectionPoolManager manager = DruidConnectionPoolManager.getInstance();

private ConnectionPoolMetricsSender sender = ConnectionPoolMetricsSender.getInstance();

public void init() throws Exception {
Executors.newSingleThreadScheduledExecutor(r -> {
Thread thread = new Thread(r);
thread.setDaemon(true);
return thread;
}).scheduleAtFixedRate(this, 0, 1, TimeUnit.SECONDS);
}

@Override
public void run() {
List metricsList = manager.getConnectionPoolMetricsList();

metricsList.forEach(metrics -> sender.offer(metrics));
}
}
```

```java
public class ConnectionPoolMetricsSender {

private final static LinkedBlockingQueue queue = new LinkedBlockingQueue<>(600);

private static ConnectionPoolMetricsSender INSTANCE;

public static ConnectionPoolMetricsSender getInstance() {
if (INSTANCE == null) {
synchronized (ConnectionPoolMetricsSender.class) {
if (INSTANCE == null) {
INSTANCE = new ConnectionPoolMetricsSender();
}
}
}
return INSTANCE;
}

private ConnectionPoolMetricsSender() {

}

public void offer(ConnectionPoolMetrics metrics) {
if (!queue.offer(metrics)) {
queue.poll();
queue.offer(metrics);
}
}

}
```

```java
public class DruidConnectionPoolManager extends AbstractConnectionPoolManager {

private static final Logger LOGGER = LoggerFactory.getLogger(DruidConnectionPoolManager.class);

private final Map druidDataSourceMap = new ConcurrentHashMap<>();

private static DruidConnectionPoolManager INSTANCE;

public static DruidConnectionPoolManager getInstance() {
if (INSTANCE == null) {
synchronized (DruidConnectionPoolManager.class) {
if (INSTANCE == null) {
INSTANCE = new DruidConnectionPoolManager();
}
}
}
return INSTANCE;
}

private DruidConnectionPoolManager() {

}

public void registerDataSource(Resource resource) {
if (resource instanceof DataSourceProxy) {
DataSourceProxy dataSourceProxy = (DataSourceProxy) resource;
druidDataSourceMap.put(dataSourceProxy.getResourceId(), (DruidDataSource) dataSourceProxy.getTargetDataSource());
}
}

public DruidDataSource get(String resourceId) {
return druidDataSourceMap.get(resourceId);
}

public List getConnectionPoolMetricsList() {
List metricsList = new ArrayList<>();

druidDataSourceMap.forEach((resourceId, druidDataSource) -> {
ConnectionPoolMetrics metrics = getConnectionPoolMetrics(resourceId);
metricsList.add(metrics);
});

return metricsList;
}

public ConnectionPoolMetrics getConnectionPoolMetrics(String resourceId) {
DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
if (druidDataSource == null) {
return null;
}

ConnectionPoolMetrics metrics = new ConnectionPoolMetrics();
metrics.setResourceId(resourceId);
metrics.setCurrentTimeMillis(System.currentTimeMillis());
metrics.setActiveCount(druidDataSource.getActiveCount());
metrics.setMaxActive(druidDataSource.getMaxActive());
metrics.setIdleCount(druidDataSource.getPoolingCount());

return metrics;
}

public void adjustPoolConfig(String resourceId, int maxActive, int minIdle) {
DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
if (druidDataSource == null) {
return;
}
if (maxActive >= 0) {
druidDataSource.setMaxActive(maxActive);
} else if (minIdle >= 0) {
druidDataSource.setMinIdle(minIdle);
}
}

public boolean checkPoolIsHealthy(String resourceId) {
DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
if (druidDataSource == null) {
return false;
}
try (Connection conn = druidDataSource.getConnection()) {
try (PreparedStatement stmt = conn.prepareStatement("SELECT 1")) {
return stmt.execute();
}
} catch (SQLException e) {
return false;
}
}

public void logPoolStatus(String resourceId) {
DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);

LOGGER.info("{}-Active Connections: {}", resourceId, druidDataSource.getActiveCount());
LOGGER.info("{}-Idle Connections: {}", resourceId, druidDataSource.getPoolingCount());
if (druidDataSource.getActiveCount() > druidDataSource.getMaxActive() * 0.8) {
LOGGER.warn("{}-Warning: Active connections exceed 80% of max allowed connections!", resourceId);
}
}

}
```

```java

public class ConnectionPoolMetrics {

private String resourceId;

private Long currentTimeMillis;

private Integer activeCount;

private Integer idleCount;

private Integer maxActive;

public ConnectionPoolMetrics(){

}

public String getResourceId() {
return resourceId;
}

public void setResourceId(String resourceId) {
this.resourceId = resourceId;
}

public Integer getActiveCount() {
return activeCount;
}

public void setActiveCount(Integer activeCount) {
this.activeCount = activeCount;
}

public Integer getIdleCount() {
return idleCount;
}

public void setIdleCount(Integer idleCount) {
this.idleCount = idleCount;
}

public Integer getMaxActive() {
return maxActive;
}

public void setMaxActive(Integer maxActive) {
this.maxActive = maxActive;
}

public Long getCurrentTimeMillis() {
return currentTimeMillis;
}

public void setCurrentTimeMillis(Long currentTimeMillis) {
this.currentTimeMillis = currentTimeMillis;
}
}

```

Start by locating DruidConnectionPoolManager, ConnectionPoolService, ConnectionPoolMetricsSender, and ConnectionPoolMetrics in the Seata codebase, then trace how DruidDataSource resources are registered and used. Clarify the integration and delivery design before changing code; done should provide the agreed periodic collection, queueing, and downstream sending of connection-pool metrics.

Written by the indexing model from the issue text.