fix(database): 完善数据库与 Redis 资源生命周期

module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/AlkaidRedis.kt

@@ -16,8 +16,50 @@
 package taboolib.expansion
 
 import taboolib.common.Inject
+import taboolib.common.LifeCycle
 import taboolib.common.env.RuntimeDependencies
 import taboolib.common.env.RuntimeDependency
+import taboolib.common.platform.Awake
+import java.io.Closeable
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicBoolean
+
+internal class RedisConnectionRegistry {
+
+    private val closed = AtomicBoolean(false)
+    private val connections = ConcurrentHashMap.newKeySet<Closeable>()
+
+    fun <T : Closeable> register(connection: T): T {
+        if (closed.get()) {
+            runCatching { connection.close() }
+            return connection
+        }
+        connections += connection
+        if (closed.get() && connections.remove(connection)) {
+            runCatching { connection.close() }
+        }
+        return connection
+    }
+
+    fun unregister(connection: Closeable) {
+        connections.remove(connection)
+    }
+
+    fun closeAll() {
+        if (!closed.compareAndSet(false, true)) {
+            return
+        }
+        connections.toList().forEach { connection ->
+            if (connections.remove(connection)) {
+                runCatching { connection.close() }
+            }
+        }
+    }
+
+    internal fun size(): Int {
+        return connections.size
+    }
+}
 
 @Inject
 @RuntimeDependencies(
@@ -52,6 +94,21 @@ import taboolib.common.env.RuntimeDependency
 )
 object AlkaidRedis {
 
+    private val connections = RedisConnectionRegistry()
+
+    internal fun <T : Closeable> register(connection: T): T {
+        return connections.register(connection)
+    }
+
+    internal fun unregister(connection: Closeable) {
+        connections.unregister(connection)
+    }
+
+    @Awake(LifeCycle.DISABLE)
+    internal fun stop() {
+        connections.closeAll()
+    }
+
     /**
      * 创建 Redis 连接器
      *

module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnection.kt

@@ -16,30 +16,23 @@
 package taboolib.expansion
 
 import redis.clients.jedis.JedisPubSub
-import taboolib.common.Inject
-import taboolib.common.LifeCycle
-import taboolib.common.platform.Awake
 import taboolib.module.configuration.Configuration
 import taboolib.module.configuration.Type
 import java.io.Closeable
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 
 class ClusterRedisConnection(val connector: ClusterRedisConnector) : Closeable, IRedisConnection {
 
+    private val closed = AtomicBoolean(false)
+    private val subscriptions = CopyOnWriteArrayList<Closeable>()
     private val service: ExecutorService = Executors.newCachedThreadPool()
 
-    @Inject
-    internal companion object {
-
-        val resources = CopyOnWriteArrayList<Closeable>()
-
-        @Awake(LifeCycle.DISABLE)
-        private fun onDisable() {
-            resources.forEach { runCatching { it.close() } }
-        }
+    init {
+        AlkaidRedis.register(this)
     }
 
     override fun eval(script: String, keys: List<String>, args: List<String>): Any? {
@@ -51,9 +44,14 @@ class ClusterRedisConnection(val connector: ClusterRedisConnector) : Closeable,
     }
 
     override fun close() {
-        connector.close()
-        service.shutdown()
-        service.awaitTermination(30, TimeUnit.SECONDS)
+        if (!closed.compareAndSet(false, true)) {
+            return
+        }
+        subscriptions.forEach { runCatching { it.close() } }
+        subscriptions.clear()
+        service.shutdownNow()
+        runCatching { connector.close() }
+        AlkaidRedis.unregister(this)
     }
 
     override fun set(key: String, value: String?) {
@@ -123,7 +121,7 @@ class ClusterRedisConnection(val connector: ClusterRedisConnector) : Closeable,
         return object : JedisPubSub() {
 
             init {
-                resources.add(Closeable {
+                subscriptions.add(Closeable {
                     if (patternMode) {
                         punsubscribe()
                     } else {

module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnector.kt

@@ -32,25 +32,37 @@ class ClusterRedisConnector : Closeable {
     var clientName: String = "default"
 
     lateinit var cluster: JedisCluster
+    private var active = false
     val nodes: LinkedHashSet<HostAndPort> = linkedSetOf()
     val genericObjectPoolConfig = GenericObjectPoolConfig<Connection>()
 
 
+    @Synchronized
     fun build(): ClusterRedisConnector {
+        if (active) {
+            cluster.close()
+        }
         genericObjectPoolConfig.maxTotal = connect
         cluster = if (auth != null && pass != null) {
             JedisCluster(nodes, timeout, timeout, maxAttempts, auth, pass, clientName, genericObjectPoolConfig)
         } else {
             JedisCluster(nodes, timeout, timeout, genericObjectPoolConfig)
         }
+        active = true
+        AlkaidRedis.register(this)
         return this
     }
 
     /**
      * 关闭连接
      */
+    @Synchronized
     override fun close() {
-        cluster.close()
+        if (active) {
+            active = false
+            cluster.close()
+        }
+        AlkaidRedis.unregister(this)
     }
 
     /**

module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnection.kt

@@ -19,42 +19,54 @@ import redis.clients.jedis.Jedis
 import redis.clients.jedis.JedisPool
 import redis.clients.jedis.JedisPubSub
 import redis.clients.jedis.exceptions.JedisConnectionException
-import taboolib.common.Inject
-import taboolib.common.LifeCycle
 import taboolib.common.PrimitiveIO
-import taboolib.common.platform.Awake
 import taboolib.module.configuration.Configuration
 import taboolib.module.configuration.Type
 import java.io.Closeable
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
+import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 
-class SingleRedisConnection(internal var pool: JedisPool, internal val connector: SingleRedisConnector): Closeable, IRedisConnection {
+class SingleRedisConnection(@Volatile internal var pool: JedisPool, internal val connector: SingleRedisConnector): Closeable, IRedisConnection {
 
+    private val closed = AtomicBoolean(false)
+    private val subscriptions = CopyOnWriteArrayList<Closeable>()
     private val service: ExecutorService = Executors.newCachedThreadPool()
+    private val reconnectService: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor()
 
-    private fun <T> exec(loop: Boolean = false, func: (Jedis) -> T): T {
+    init {
+        AlkaidRedis.register(this)
+    }
+
+    private fun <T> exec(func: (Jedis) -> T): T {
+        check(!closed.get()) { "Redis connection is closed" }
+        val currentPool = pool
         return try {
-            pool.resource.use { func(it) }
+            currentPool.resource.use { func(it) }
         } catch (ex: JedisConnectionException) {
             PrimitiveIO.error("Redis connection failed: ${ex.message}")
-            // 如果是循环模式则等待一段时间
-            if (loop) {
-                Thread.sleep(connector.reconnectDelay)
-            }
-            // 重连
-            pool = connector.connect().pool!!
-            // 重新执行
-            if (loop) {
-                exec(true, func)
-            } else {
-                pool.resource.use { func(it) }
-            }
+            reconnect(currentPool).resource.use { func(it) }
         }
     }
 
+    @Synchronized
+    private fun reconnect(failedPool: JedisPool): JedisPool {
+        check(!closed.get()) { "Redis connection is closed" }
+        if (pool !== failedPool) {
+            return pool
+        }
+        val connectorPool = connector.pool
+        if (connectorPool != null && connectorPool !== failedPool) {
+            pool = connectorPool
+            return connectorPool
+        }
+        connector.connect()
+        return connector.pool!!.also { pool = it }
+    }
+
     override fun eval(script: String, keys: List<String>, args: List<String>): Any? {
         return exec {
             it.eval(script, keys, args)
@@ -71,7 +83,19 @@ class SingleRedisConnection(internal var pool: JedisPool, internal val connector
      * 关闭连接
      */
     override fun close() {
-        pool.destroy()
+        if (!closed.compareAndSet(false, true)) {
+            return
+        }
+        subscriptions.forEach { runCatching { it.close() } }
+        subscriptions.clear()
+        reconnectService.shutdownNow()
+        service.shutdownNow()
+        runCatching {
+            synchronized(this) {
+                pool.close()
+            }
+        }
+        AlkaidRedis.unregister(this)
     }
 
     /**
@@ -151,17 +175,35 @@ class SingleRedisConnection(internal var pool: JedisPool, internal val connector
      * @param func 信息处理函数
      */
     override fun subscribe(vararg channel: String, patternMode: Boolean, func: RedisMessage.() -> Unit) {
-        service.submit {
-            try {
-                exec(true) { jedis ->
-                    if (patternMode) {
-                        jedis.psubscribe(createPubSub(true, func), *channel)
-                    } else {
-                        jedis.subscribe(createPubSub(false, func), *channel)
+        submitSubscription(channel, patternMode, createPubSub(patternMode, func))
+    }
+
+    private fun submitSubscription(channel: Array<out String>, patternMode: Boolean, pubSub: JedisPubSub) {
+        if (closed.get()) {
+            return
+        }
+        runCatching {
+            service.submit {
+                try {
+                    exec { jedis ->
+                        if (patternMode) {
+                            jedis.psubscribe(pubSub, *channel)
+                        } else {
+                            jedis.subscribe(pubSub, *channel)
+                        }
+                    }
+                } catch (ex: Throwable) {
+                    if (!closed.get()) {
+                        PrimitiveIO.error("Redis subscription failed: ${ex.message}")
+                        runCatching {
+                            reconnectService.schedule(
+                                { submitSubscription(channel, patternMode, pubSub) },
+                                connector.reconnectDelay,
+                                TimeUnit.MILLISECONDS
+                            )
+                        }
                     }
                 }
-            } catch (ex: Throwable) {
-                ex.printStackTrace()
             }
         }
     }
@@ -170,7 +212,7 @@ class SingleRedisConnection(internal var pool: JedisPool, internal val connector
         return object : JedisPubSub() {
 
             init {
-                resources.add(Closeable {
+                subscriptions.add(Closeable {
                     if (patternMode) {
                         punsubscribe()
                     } else {
@@ -324,15 +366,4 @@ class SingleRedisConnection(internal var pool: JedisPool, internal val connector
     override fun type(key: String): String {
         return exec { it.type(key) }
     }
-
-    @Inject
-    internal companion object {
-
-        val resources = CopyOnWriteArrayList<Closeable>()
-
-        @Awake(LifeCycle.DISABLE)
-        private fun onDisable() {
-            resources.forEach { runCatching { it.close() } }
-        }
-    }
 }

module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnector.kt

@@ -37,22 +37,29 @@ class SingleRedisConnector: Closeable {
      *
      * @return [SingleRedisConnector]
      */
+    @Synchronized
     fun connect(): SingleRedisConnector {
         config.maxTotal = connect
+        val previousPool = pool
         pool = when {
             auth != null && pass != null -> JedisPool(config, host, port, timeout, auth, pass)
             auth != null -> JedisPool(config, host, port, timeout, auth, null)
             pass != null -> JedisPool(config, host, port, timeout, pass)
             else -> JedisPool(config, host, port, timeout)
         }
+        previousPool?.close()
+        AlkaidRedis.register(this)
         return this
     }
 
     /**
      * 关闭连接
      */
+    @Synchronized
     override fun close() {
-        pool?.destroy()
+        pool?.close()
+        pool = null
+        AlkaidRedis.unregister(this)
     }
 
     /**