20260828173144

code/fms/fms-api/src/main/java/cn/g3soft/fmsapi/database/OrgDataSourceFactory.java

@@ -28,9 +28,10 @@ public class OrgDataSourceFactory {
         dataSource.setTestWhileIdle(poolProperties.isTestWhileIdle());
         dataSource.setTestOnBorrow(poolProperties.isTestOnBorrow());
         dataSource.setTestOnReturn(poolProperties.isTestOnReturn());
-        dataSource.setConnectionErrorRetryAttempts(3);
-        dataSource.setBreakAfterAcquireFailure(false);
-        dataSource.setFailFast(false);
+        // 连接失败快速失败,不做重试:重试没有意义,等下次请求再触发建连
+        dataSource.setConnectionErrorRetryAttempts(0);
+        dataSource.setBreakAfterAcquireFailure(true);
+        dataSource.setFailFast(true);
         // 定期 ping 空闲连接,对抗云 NAT/防火墙回收空闲 TCP 会话(默认 120s)
         dataSource.setKeepAlive(true);
         dataSource.setKeepAliveBetweenTimeMillis(120_000);

code/fms/fms-api/src/main/java/cn/g3soft/fmsapi/database/OrgDataSourceManager.java

@@ -1,19 +1,38 @@
 package cn.g3soft.fmsapi.database;
 
+import cn.g3soft.fmsapi.exception.AuthenticationException;
 import com.alibaba.druid.pool.DruidDataSource;
 import jakarta.annotation.PreDestroy;
 import jakarta.annotation.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
+/**
+ * 机构连接池注册表:按机构码懒加载并缓存连接池,定期检测配置文件变更。
+ * 配置(url/密码等)变更后无需重启,到期自动重建连接池并延迟关闭旧池,
+ * 保证在途请求有足够时间归还旧连接。
+ */
 @Component
 public class OrgDataSourceManager {
 
-    private final ConcurrentMap<String, DruidDataSource> dataSources = new ConcurrentHashMap<>();
+    private static final Logger log = LoggerFactory.getLogger(OrgDataSourceManager.class);
+
+    private final ConcurrentMap<String, PoolEntry> dataSources = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService retiredPoolCloser = Executors.newSingleThreadScheduledExecutor(task -> {
+        Thread thread = new Thread(task, "org-datasource-retired-close");
+        thread.setDaemon(true);
+        return thread;
+    });
 
     @Resource
     private OrgDatabaseConfigLoader configLoader;
@@ -21,9 +40,16 @@ public class OrgDataSourceManager {
     @Resource
     private OrgDataSourceFactory dataSourceFactory;
 
+    @Resource
+    private OrgDatabaseProperties properties;
+
     public DruidDataSource getDataSource(String orgId) {
         String normalizedOrgId = configLoader.normalizeOrgId(orgId);
-        return dataSources.computeIfAbsent(normalizedOrgId, this::createDataSource);
+        PoolEntry entry = dataSources.computeIfAbsent(normalizedOrgId, this::createEntry);
+        if (tryClaimRefreshCheck(entry)) {
+            return refreshIfConfigChanged(normalizedOrgId, entry).dataSource();
+        }
+        return entry.dataSource();
     }
 
     public Connection getConnection(String orgId) throws SQLException {
@@ -36,11 +62,11 @@ public class OrgDataSourceManager {
 
     public boolean closeDataSource(String orgId) {
         String normalizedOrgId = configLoader.normalizeOrgId(orgId);
-        DruidDataSource dataSource = dataSources.remove(normalizedOrgId);
-        if (dataSource == null) {
+        PoolEntry entry = dataSources.remove(normalizedOrgId);
+        if (entry == null) {
             return false;
         }
-        dataSource.close();
+        entry.dataSource().close();
         return true;
     }
 
@@ -50,12 +76,123 @@ public class OrgDataSourceManager {
 
     @PreDestroy
     public void closeAll() {
-        dataSources.values().forEach(DruidDataSource::close);
+        retiredPoolCloser.shutdownNow();
+        dataSources.values().forEach(entry -> entry.dataSource().close());
         dataSources.clear();
     }
 
-    private DruidDataSource createDataSource(String orgId) {
+    /**
+     * 配置文件变更事件回调:立即核对并切换连接池(仅对已缓存的机构生效)。
+     * 由配置监听线程调用,配置读取与建池的 I/O 不占用请求线程。
+     */
+    public void onConfigChanged(String orgId) {
+        String normalizedOrgId = configLoader.normalizeOrgId(orgId);
+        PoolEntry entry = dataSources.get(normalizedOrgId);
+        if (entry == null) {
+            return;
+        }
+        try {
+            refreshIfConfigChanged(normalizedOrgId, entry);
+        } catch (AuthenticationException exception) {
+            // 配置文件已删除,旧池已在 refreshIfConfigChanged 中淘汰
+        }
+    }
+
+    /**
+     * TTL 到期后由一个线程认领检测任务(CAS 防止并发重复 stat 文件)。
+     */
+    private boolean tryClaimRefreshCheck(PoolEntry entry) {
+        long intervalNanos = properties.getRefreshCheckInterval().toNanos();
+        if (intervalNanos <= 0) {
+            return false;
+        }
+        long last = entry.checkedAtNanos.get();
+        long now = System.nanoTime();
+        if (now - last < intervalNanos) {
+            return false;
+        }
+        return entry.checkedAtNanos.compareAndSet(last, now);
+    }
+
+    private PoolEntry refreshIfConfigChanged(String orgId, PoolEntry entry) {
+        long lastModified = configLoader.lastModified(orgId);
+        if (lastModified == entry.configLastModified()) {
+            return entry;
+        }
+        PoolEntry fresh;
+        try {
+            fresh = createEntry(orgId);
+        } catch (AuthenticationException exception) {
+            // 配置文件已不存在:机构已下线,淘汰旧池,按"机构码错误"处理
+            if (dataSources.remove(orgId, entry)) {
+                scheduleClose(entry.dataSource());
+            }
+            throw exception;
+        } catch (RuntimeException exception) {
+            // 配置读取或建池瞬时失败:保留旧池,下个检测周期重试
+            log.warn("刷新机构数据库连接池失败,继续使用旧配置: {}", orgId, exception);
+            return entry;
+        }
+        if (fresh.config().equals(entry.config())) {
+            // 内容未变(仅时间戳被 touch):避免无谓重建
+            entry.configLastModified(fresh.configLastModified());
+            fresh.dataSource().close();
+            return entry;
+        }
+        if (dataSources.replace(orgId, entry, fresh)) {
+            scheduleClose(entry.dataSource());
+            log.info("机构数据库配置已变更,已切换到新连接池: {}", orgId);
+            return fresh;
+        }
+        // 并发刷新:其他线程已完成切换或已淘汰旧条目,丢弃本次多余的池
+        fresh.dataSource().close();
+        PoolEntry current = dataSources.get(orgId);
+        if (current != null) {
+            return current;
+        }
+        // 旧条目被并发淘汰且无新条目(如配置文件删除后又被重建):退回懒加载语义
+        return dataSources.computeIfAbsent(orgId, this::createEntry);
+    }
+
+    private PoolEntry createEntry(String orgId) {
+        // 先取时间戳再加载:若加载期间文件又被修改,记录的是旧时间戳,下个周期能检测到
+        long lastModified = configLoader.lastModified(orgId);
         OrgDatabaseConfig config = configLoader.load(orgId);
-        return dataSourceFactory.create(orgId, config);
+        return new PoolEntry(dataSourceFactory.create(orgId, config), config, lastModified);
+    }
+
+    private void scheduleClose(DruidDataSource dataSource) {
+        long delayMillis = properties.getRetiredPoolCloseDelay().toMillis();
+        retiredPoolCloser.schedule(dataSource::close, delayMillis, TimeUnit.MILLISECONDS);
+    }
+
+    private static final class PoolEntry {
+
+        private final DruidDataSource dataSource;
+        private final OrgDatabaseConfig config;
+        private volatile long configLastModified;
+        private final AtomicLong checkedAtNanos = new AtomicLong(System.nanoTime());
+
+        private PoolEntry(DruidDataSource dataSource, OrgDatabaseConfig config, long configLastModified) {
+            this.dataSource = dataSource;
+            this.config = config;
+            this.configLastModified = configLastModified;
+        }
+
+        private DruidDataSource dataSource() {
+            return dataSource;
+        }
+
+        private OrgDatabaseConfig config() {
+            return config;
+        }
+
+        private long configLastModified() {
+            return configLastModified;
+        }
+
+        private void configLastModified(long configLastModified) {
+            this.configLastModified = configLastModified;
+        }
     }
 }

code/fms/fms-api/src/main/java/cn/g3soft/fmsapi/database/OrgDatabaseConfigLoader.java

@@ -35,10 +35,7 @@ public class OrgDatabaseConfigLoader {
 
     public OrgDatabaseConfig load(String orgId) {
         String normalizedOrgId = normalizeOrgId(orgId);
-        Path configFile = configDir.resolve(normalizedOrgId + ".properties").normalize();
-        if (!configFile.startsWith(configDir)) {
-            throw new IllegalArgumentException("机构配置文件路径不合法");
-        }
+        Path configFile = resolveConfigFile(normalizedOrgId);
         if (!Files.isRegularFile(configFile)) {
             throw new AuthenticationException("机构码错误");
         }
@@ -80,6 +77,26 @@ public class OrgDatabaseConfigLoader {
         return configDir;
     }
 
+    /**
+     * 配置文件修改时间戳,用于检测配置变更。文件不存在时返回 -1。
+     */
+    public long lastModified(String orgId) {
+        Path configFile = resolveConfigFile(normalizeOrgId(orgId));
+        try {
+            return Files.getLastModifiedTime(configFile).toMillis();
+        } catch (IOException exception) {
+            return -1L;
+        }
+    }
+
+    private Path resolveConfigFile(String normalizedOrgId) {
+        Path configFile = configDir.resolve(normalizedOrgId + ".properties").normalize();
+        if (!configFile.startsWith(configDir)) {
+            throw new IllegalArgumentException("机构配置文件路径不合法");
+        }
+        return configFile;
+    }
+
     private String required(Properties properties, String key, Path configFile) {
         String value = properties.getProperty(key);
         if (value == null || value.isBlank()) {

code/fms/fms-api/src/main/java/cn/g3soft/fmsapi/database/OrgDatabaseConfigWatcher.java

@@ -0,0 +1,141 @@
+package cn.g3soft.fmsapi.database;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.annotation.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.nio.file.ClosedWatchServiceException;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+import java.nio.file.StandardWatchEventKinds;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * 监听机构配置目录,配置文件变更时立即通知 {@link OrgDataSourceManager} 重建连接池,
+ * 使配置变更毫秒级生效,配置读取与建池的 I/O 全部发生在监听线程,不占用请求线程。
+ * 监听注册失败、事件溢出或目录失效时自动降级为定期轮询兜底,不影响服务。
+ */
+@Component
+public class OrgDatabaseConfigWatcher {
+
+    private static final Logger log = LoggerFactory.getLogger(OrgDatabaseConfigWatcher.class);
+    private static final Pattern CONFIG_FILE_PATTERN =
+            Pattern.compile("([A-Za-z0-9_-]{1,32})\\.properties");
+    private static final long DEBOUNCE_MILLIS = 500;
+
+    @Resource
+    private OrgDatabaseConfigLoader configLoader;
+
+    @Resource
+    private OrgDataSourceManager dataSourceManager;
+
+    private WatchService watchService;
+    private volatile Thread watcherThread;
+
+    @PostConstruct
+    public void start() {
+        Path configDir = configLoader.getConfigDir();
+        try {
+            watchService = FileSystems.getDefault().newWatchService();
+            configDir.register(
+                    watchService,
+                    StandardWatchEventKinds.ENTRY_CREATE,
+                    StandardWatchEventKinds.ENTRY_MODIFY,
+                    StandardWatchEventKinds.ENTRY_DELETE
+            );
+        } catch (IOException exception) {
+            log.warn("注册机构配置目录监听失败,配置变更检测降级为定期轮询: {}", configDir, exception);
+            closeWatchService();
+            return;
+        }
+        Thread thread = new Thread(this::watch, "org-dbconfig-watcher");
+        thread.setDaemon(true);
+        this.watcherThread = thread;
+        thread.start();
+        log.info("已启动机构配置目录监听: {}", configDir);
+    }
+
+    @PreDestroy
+    public void stop() {
+        Thread thread = watcherThread;
+        if (thread != null) {
+            thread.interrupt();
+        }
+        closeWatchService();
+    }
+
+    private void watch() {
+        try {
+            while (true) {
+                WatchKey key = watchService.take();
+                Set<String> orgIds = new HashSet<>();
+                collect(key, orgIds);
+                // 合并安静期内的后续事件(编辑器保存常触发多次写入),避免重复重建连接池
+                long deadline = System.currentTimeMillis() + DEBOUNCE_MILLIS;
+                while (System.currentTimeMillis() < deadline) {
+                    WatchKey more = watchService.poll(
+                            deadline - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+                    if (more == null) {
+                        break;
+                    }
+                    collect(more, orgIds);
+                }
+                orgIds.forEach(this::notifyChanged);
+            }
+        } catch (ClosedWatchServiceException | InterruptedException exception) {
+            if (exception instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
+            // 正常退出:服务关闭或线程中断,轮询兜底仍然生效
+            log.debug("机构配置目录监听线程退出");
+        }
+    }
+
+    private void collect(WatchKey key, Set<String> orgIds) {
+        for (WatchEvent<?> event : key.pollEvents()) {
+            if (event.kind() == StandardWatchEventKinds.OVERFLOW) {
+                continue;
+            }
+            if (event.context() instanceof Path path) {
+                Matcher matcher = CONFIG_FILE_PATTERN.matcher(path.getFileName().toString());
+                if (matcher.matches()) {
+                    orgIds.add(matcher.group(1).toUpperCase(Locale.ROOT));
+                }
+            }
+        }
+        if (!key.reset()) {
+            log.warn("机构配置目录已不可用,文件监听停止,配置变更检测降级为定期轮询");
+            closeWatchService();
+        }
+    }
+
+    private void notifyChanged(String orgId) {
+        try {
+            dataSourceManager.onConfigChanged(orgId);
+        } catch (RuntimeException exception) {
+            log.warn("处理机构配置变更事件失败: {}", orgId, exception);
+        }
+    }
+
+    private void closeWatchService() {
+        if (watchService != null) {
+            try {
+                watchService.close();
+            } catch (IOException ignored) {
+                // 关闭失败无需处理
+            }
+        }
+    }
+}

code/fms/fms-api/src/main/java/cn/g3soft/fmsapi/database/OrgDatabaseProperties.java

@@ -9,6 +9,8 @@ import java.time.Duration;
 public class OrgDatabaseProperties {
 
     private Path configDir = Path.of("./config/dbconfigs");
+    private Duration refreshCheckInterval = Duration.ofSeconds(30);
+    private Duration retiredPoolCloseDelay = Duration.ofSeconds(60);
     private final Pool pool = new Pool();
 
     public Path getConfigDir() {
@@ -19,6 +21,22 @@ public class OrgDatabaseProperties {
         this.configDir = configDir;
     }
 
+    public Duration getRefreshCheckInterval() {
+        return refreshCheckInterval;
+    }
+
+    public void setRefreshCheckInterval(Duration refreshCheckInterval) {
+        this.refreshCheckInterval = refreshCheckInterval;
+    }
+
+    public Duration getRetiredPoolCloseDelay() {
+        return retiredPoolCloseDelay;
+    }
+
+    public void setRetiredPoolCloseDelay(Duration retiredPoolCloseDelay) {
+        this.retiredPoolCloseDelay = retiredPoolCloseDelay;
+    }
+
     public Pool getPool() {
         return pool;
     }

code/fms/fms-api/src/main/resources/application.yaml

@@ -13,8 +13,13 @@ fms:
     jwt-expiration: 24h
   database:
     config-dir: ./config/dbconfigs
+    # 机构配置文件变更检测周期(文件监听的兜底,正常由监听毫秒级触发;设为 0s 关闭轮询)
+    refresh-check-interval: 30s
+    # 配置变更切换后,旧连接池延迟关闭时间,等待在途请求归还连接
+    retired-pool-close-delay: 60s
     pool:
-      initial-size: 0
+      # 建池时先建 1 条物理连接:配置指向不可达的库时快速失败,保留旧池不切换
+      initial-size: 1
       min-idle: 0
       max-active: 10
       max-wait: 5000