diff --git a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/AlkaidRedis.kt b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/AlkaidRedis.kt index ad735dc68..44a43410f 100644 --- a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/AlkaidRedis.kt +++ b/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() + + fun 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 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 连接器 * diff --git a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnection.kt b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnection.kt index 86279cc08..2890e1252 100644 --- a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnection.kt +++ b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnection.kt @@ -16,9 +16,6 @@ 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 @@ -26,20 +23,16 @@ 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() private val service: ExecutorService = Executors.newCachedThreadPool() - @Inject - internal companion object { - - val resources = CopyOnWriteArrayList() - - @Awake(LifeCycle.DISABLE) - private fun onDisable() { - resources.forEach { runCatching { it.close() } } - } + init { + AlkaidRedis.register(this) } override fun eval(script: String, keys: List, args: List): 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 { diff --git a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnector.kt b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnector.kt index 308a17d27..bd9944b6d 100644 --- a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/ClusterRedisConnector.kt +++ b/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 = linkedSetOf() val genericObjectPoolConfig = GenericObjectPoolConfig() + @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) } /** diff --git a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnection.kt b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnection.kt index 7bc12d4e9..56e118363 100644 --- a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnection.kt +++ b/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() private val service: ExecutorService = Executors.newCachedThreadPool() + private val reconnectService: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor() - private fun exec(loop: Boolean = false, func: (Jedis) -> T): T { + init { + AlkaidRedis.register(this) + } + + private fun 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, args: List): 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, 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() - - @Awake(LifeCycle.DISABLE) - private fun onDisable() { - resources.forEach { runCatching { it.close() } } - } - } } diff --git a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnector.kt b/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnector.kt index dbcf60c47..489cb7138 100644 --- a/module/database/database-alkaid-redis/src/main/kotlin/taboolib/expansion/SingleRedisConnector.kt +++ b/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) } /** diff --git a/module/database/database-alkaid-redis/src/test/kotlin/taboolib/expansion/RedisConnectionRegistryTest.kt b/module/database/database-alkaid-redis/src/test/kotlin/taboolib/expansion/RedisConnectionRegistryTest.kt new file mode 100644 index 000000000..1cc28993a --- /dev/null +++ b/module/database/database-alkaid-redis/src/test/kotlin/taboolib/expansion/RedisConnectionRegistryTest.kt @@ -0,0 +1,49 @@ +package taboolib.expansion + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import java.io.Closeable +import java.util.concurrent.atomic.AtomicInteger + +class RedisConnectionRegistryTest { + + @Test + fun `registered connections close exactly once`() { + val registry = RedisConnectionRegistry() + val closeCount = AtomicInteger() + val connection = Closeable { closeCount.incrementAndGet() } + + registry.register(connection) + registry.register(connection) + registry.closeAll() + registry.closeAll() + + assertEquals(1, closeCount.get()) + assertEquals(0, registry.size()) + } + + @Test + fun `unregistered connection remains caller owned`() { + val registry = RedisConnectionRegistry() + val closeCount = AtomicInteger() + val connection = Closeable { closeCount.incrementAndGet() } + + registry.register(connection) + registry.unregister(connection) + registry.closeAll() + + assertEquals(0, closeCount.get()) + } + + @Test + fun `connection registered after shutdown closes immediately`() { + val registry = RedisConnectionRegistry() + val closeCount = AtomicInteger() + + registry.closeAll() + registry.register(Closeable { closeCount.incrementAndGet() }) + + assertEquals(1, closeCount.get()) + assertEquals(0, registry.size()) + } +} diff --git a/module/database/database-lettuce-redis/build.gradle.kts b/module/database/database-lettuce-redis/build.gradle.kts index 8d35c1a79..298ec264d 100644 --- a/module/database/database-lettuce-redis/build.gradle.kts +++ b/module/database/database-lettuce-redis/build.gradle.kts @@ -7,12 +7,14 @@ dependencies { // 使用 api 传递依赖 api("io.lettuce:lettuce-core:7.2.1.RELEASE") compileOnly("org.apache.commons:commons-pool2:2.12.1") + testImplementation("org.apache.commons:commons-pool2:2.12.1") compileOnly(project(":common")) compileOnly(project(":common-env")) compileOnly(project(":common-util")) compileOnly(project(":common-platform-api")) compileOnly(project(":module:basic:basic-configuration")) + testImplementation(project(":module:basic:basic-configuration")) } tasks { diff --git a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceClusterRedisClient.kt b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceClusterRedisClient.kt index a6a28a0d5..20cdb936a 100644 --- a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceClusterRedisClient.kt +++ b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceClusterRedisClient.kt @@ -24,198 +24,365 @@ import taboolib.expansion.lettuce.IRedisChannel import taboolib.expansion.lettuce.IRedisClient import taboolib.expansion.lettuce.cluster.IRedisClusterCommand import taboolib.expansion.lettuce.cluster.IRedisClusterPubSub +import java.util.concurrent.CancellationException import java.util.concurrent.CompletableFuture +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference import kotlin.collections.plusAssign import kotlin.time.toJavaDuration @Suppress("DuplicatedCode") -class LettuceClusterRedisClient(val redisConfig: LettuceRedisConfig): IRedisClient, IRedisChannel, IRedisClusterCommand, IRedisClusterPubSub { +class LettuceClusterRedisClient(val redisConfig: LettuceRedisConfig): IRedisClient, IRedisChannel, IRedisClusterCommand, IRedisClusterPubSub, LettuceRedisResource { + @Volatile lateinit var client: RedisClusterClient + @Volatile lateinit var pool: GenericObjectPool> + + @Volatile lateinit var asyncPool: BoundedAsyncPool> + @Volatile lateinit var pubSubConnection: StatefulRedisClusterPubSubConnection + + @Volatile lateinit var resources: DefaultClientResources + private val startup = AtomicReference?>() + private val stopped = AtomicBoolean(false) + private val shutdownStarted = AtomicBoolean(false) + private val lifecycleLock = Any() + @OptIn(ExperimentalStdlibApi::class) override fun start(autoRelease: Boolean): CompletableFuture { val completableFuture = CompletableFuture() - val resource = DefaultClientResources.builder() - - if (redisConfig.ioThreadPoolSize != 0) { - resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + if (!startup.compareAndSet(null, completableFuture)) { + val existingStartup = startup.get()!! + if (stopped.get() && !existingStartup.isCompletedExceptionally && !existingStartup.isCancelled) { + return CompletableFuture().also { + it.completeExceptionally(CancellationException("Redis cluster client is stopped")) + } + } + return existingStartup } - if (redisConfig.computationThreadPoolSize != 0) { - resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + if (stopped.get()) { + completableFuture.completeExceptionally(CancellationException("Redis cluster client is stopped")) + return completableFuture } + try { + val resource = DefaultClientResources.builder() + if (redisConfig.ioThreadPoolSize != 0) { + resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + } + if (redisConfig.computationThreadPoolSize != 0) { + resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + } - val cluster = redisConfig.cluster - - val uris = cluster.nodes.map { - it.redisURIBuilder().build() - } - val clientOptions = ClusterClientOptions.builder() + val cluster = redisConfig.cluster + val uris = cluster.nodes.map { it.redisURIBuilder().build() } + val clientOptions = ClusterClientOptions.builder() + if (redisConfig.ssl) { + clientOptions.sslOptions(redisConfig.sslOptions) + } - if (redisConfig.ssl) { - clientOptions.sslOptions(redisConfig.sslOptions) - } + val topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() + .enablePeriodicRefresh(cluster.enablePeriodicRefresh) + .refreshTriggersReconnectAttempts(cluster.refreshTriggersReconnectAttempts) + .dynamicRefreshSources(cluster.dynamicRefreshSources) + .closeStaleConnections(cluster.closeStaleConnections) + + // Lettuce 7.0+ 默认启用所有自适应触发器，需要禁用未配置的触发器 + val configuredTriggers = cluster.enableAdaptiveRefreshTrigger.toSet() + if (configuredTriggers.isEmpty()) { + topologyRefreshOptions.disableAllAdaptiveRefreshTriggers() + } else { + val triggersToDisable = ClusterTopologyRefreshOptions.RefreshTrigger.values() + .filter { it !in configuredTriggers } + .toTypedArray() + if (triggersToDisable.isNotEmpty()) { + topologyRefreshOptions.disableAdaptiveRefreshTrigger(*triggersToDisable) + } + } - val topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() - .enablePeriodicRefresh(cluster.enablePeriodicRefresh) - .refreshTriggersReconnectAttempts(cluster.refreshTriggersReconnectAttempts) - .dynamicRefreshSources(cluster.dynamicRefreshSources) - .closeStaleConnections(cluster.closeStaleConnections) - - // Lettuce 7.0+ 默认启用所有自适应触发器，需要禁用未配置的触发器 - val configuredTriggers = cluster.enableAdaptiveRefreshTrigger.toSet() - if (configuredTriggers.isEmpty()) { - // 如果未配置任何触发器，禁用所有 - topologyRefreshOptions.disableAllAdaptiveRefreshTriggers() - } else { - // 禁用未配置的触发器 - val triggersToDisable = ClusterTopologyRefreshOptions.RefreshTrigger.values() - .filter { it !in configuredTriggers } - .toTypedArray() - if (triggersToDisable.isNotEmpty()) { - topologyRefreshOptions.disableAdaptiveRefreshTrigger(*triggersToDisable) + cluster.adaptiveRefreshTriggersTimeout?.toJavaDuration()?.let { + topologyRefreshOptions.adaptiveRefreshTriggersTimeout(it) + } + cluster.refreshPeriod?.toJavaDuration()?.let { + topologyRefreshOptions.refreshPeriod(it) + } + clientOptions + .topologyRefreshOptions(topologyRefreshOptions.build()) + .autoReconnect(redisConfig.autoReconnect) + .maxRedirects(cluster.maxRedirects) + .validateClusterNodeMembership(cluster.validateClusterNodeMembership) + .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) + + val newResources = resource.build() + val newClient = try { + RedisClusterClient.create(newResources, uris).apply { + setOptions(clientOptions.build()) + } + } catch (ex: Throwable) { + newResources.shutdown() + throw ex + } + synchronized(lifecycleLock) { + resources = newResources + client = newClient + } + if (stopped.get()) { + closeResources() + return completableFuture } - } - cluster.adaptiveRefreshTriggersTimeout?.toJavaDuration()?.let { topologyRefreshOptions.adaptiveRefreshTriggersTimeout(it) } - cluster.refreshPeriod?.toJavaDuration()?.let { topologyRefreshOptions.refreshPeriod(it) } - clientOptions - .topologyRefreshOptions(topologyRefreshOptions.build()) - .autoReconnect(redisConfig.autoReconnect) - .maxRedirects(cluster.maxRedirects) - .validateClusterNodeMembership(cluster.validateClusterNodeMembership) - .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) - - resources = resource.build() - client = RedisClusterClient.create(resources, uris) - client.setOptions(clientOptions.build()) - - // 连接 pub/sub 通道 - pubSubConnection = client.connectPubSub() - // 连接同步 - pool = ConnectionPoolSupport.createGenericObjectPool( - { client.connect().apply { - if (redisConfig.enableSlaves) { - val slaves = redisConfig.slaves - readFrom = slaves.readFrom + // 异步连接 pub/sub 通道，避免 start() 阻塞调用线程 + val pubSubReady = client.connectPubSubAsync(StringCodec.UTF8).thenAccept { + pubSubConnection = it + if (stopped.get()) { + it.closeAsync() } - } }, - redisConfig.pool.clusterPoolConfig() - ) - // 连接异步 - AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( - { client.connectAsync(StringCodec.UTF8).whenComplete { v, _ -> - if (redisConfig.enableSlaves) { - val slaves = redisConfig.slaves - v.readFrom = slaves.readFrom + }.toCompletableFuture() + // 连接同步 + pool = ConnectionPoolSupport.createGenericObjectPool( + { + client.connect().apply { + if (redisConfig.enableSlaves) { + readFrom = redisConfig.slaves.readFrom + } + } + }, + redisConfig.pool.clusterPoolConfig() + ) + if (stopped.get()) { + pool.close() + } + // 连接异步 + val poolReady = AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( + { + client.connectAsync(StringCodec.UTF8).whenComplete { value, _ -> + if (redisConfig.enableSlaves) { + value.readFrom = redisConfig.slaves.readFrom + } + } + }, + redisConfig.asyncPool.poolConfig() + ).thenAccept { + asyncPool = it + if (stopped.get()) { + it.closeAsync() } - } }, - redisConfig.asyncPool.poolConfig() - ).thenAccept { - asyncPool = it - completableFuture.complete(null) - } - if (autoRelease) { - LettuceRedis.clusterClients += this + }.toCompletableFuture() + coordinateStart(completableFuture, autoRelease, pubSubReady, poolReady) + } catch (ex: Throwable) { + failStart(completableFuture, ex) } return completableFuture } @OptIn(ExperimentalStdlibApi::class) override fun startSync(autoRelease: Boolean) { - val resource = DefaultClientResources.builder() - - if (redisConfig.ioThreadPoolSize != 0) { - resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + val completableFuture = CompletableFuture() + if (!startup.compareAndSet(null, completableFuture)) { + val existingStartup = startup.get()!! + check(!stopped.get() && existingStartup.isDone && !existingStartup.isCompletedExceptionally && !existingStartup.isCancelled) { + "Redis cluster client is already starting or failed to start" + } + return } - if (redisConfig.computationThreadPoolSize != 0) { - resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + if (stopped.get()) { + val error = IllegalStateException("Redis cluster client is stopped") + completableFuture.completeExceptionally(error) + throw error } + try { + val resource = DefaultClientResources.builder() + if (redisConfig.ioThreadPoolSize != 0) { + resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + } + if (redisConfig.computationThreadPoolSize != 0) { + resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + } - val cluster = redisConfig.cluster + val cluster = redisConfig.cluster + val uris = cluster.nodes.map { it.redisURIBuilder().build() } + val clientOptions = ClusterClientOptions.builder() + if (redisConfig.ssl) { + clientOptions.sslOptions(redisConfig.sslOptions) + } - val uris = cluster.nodes.map { - it.redisURIBuilder().build() - } - val clientOptions = ClusterClientOptions.builder() + val topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() + .enablePeriodicRefresh(cluster.enablePeriodicRefresh) + .refreshTriggersReconnectAttempts(cluster.refreshTriggersReconnectAttempts) + .dynamicRefreshSources(cluster.dynamicRefreshSources) + .closeStaleConnections(cluster.closeStaleConnections) + + // Lettuce 7.0+ 默认启用所有自适应触发器，需要禁用未配置的触发器 + val configuredTriggers = cluster.enableAdaptiveRefreshTrigger.toSet() + if (configuredTriggers.isEmpty()) { + topologyRefreshOptions.disableAllAdaptiveRefreshTriggers() + } else { + val triggersToDisable = ClusterTopologyRefreshOptions.RefreshTrigger.values() + .filter { it !in configuredTriggers } + .toTypedArray() + if (triggersToDisable.isNotEmpty()) { + topologyRefreshOptions.disableAdaptiveRefreshTrigger(*triggersToDisable) + } + } - if (redisConfig.ssl) { - clientOptions.sslOptions(redisConfig.sslOptions) - } + cluster.adaptiveRefreshTriggersTimeout?.toJavaDuration()?.let { + topologyRefreshOptions.adaptiveRefreshTriggersTimeout(it) + } + cluster.refreshPeriod?.toJavaDuration()?.let { + topologyRefreshOptions.refreshPeriod(it) + } + clientOptions + .topologyRefreshOptions(topologyRefreshOptions.build()) + .autoReconnect(redisConfig.autoReconnect) + .maxRedirects(cluster.maxRedirects) + .validateClusterNodeMembership(cluster.validateClusterNodeMembership) + .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) + + val newResources = resource.build() + val newClient = try { + RedisClusterClient.create(newResources, uris).apply { + setOptions(clientOptions.build()) + } + } catch (ex: Throwable) { + newResources.shutdown() + throw ex + } + synchronized(lifecycleLock) { + resources = newResources + client = newClient + } + if (stopped.get()) { + closeResources() + error("Redis cluster client was stopped during startup") + } - val topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() - .enablePeriodicRefresh(cluster.enablePeriodicRefresh) - .refreshTriggersReconnectAttempts(cluster.refreshTriggersReconnectAttempts) - .dynamicRefreshSources(cluster.dynamicRefreshSources) - .closeStaleConnections(cluster.closeStaleConnections) - - // Lettuce 7.0+ 默认启用所有自适应触发器，需要禁用未配置的触发器 - val configuredTriggers = cluster.enableAdaptiveRefreshTrigger.toSet() - if (configuredTriggers.isEmpty()) { - // 如果未配置任何触发器，禁用所有 - topologyRefreshOptions.disableAllAdaptiveRefreshTriggers() - } else { - // 禁用未配置的触发器 - val triggersToDisable = ClusterTopologyRefreshOptions.RefreshTrigger.values() - .filter { it !in configuredTriggers } - .toTypedArray() - if (triggersToDisable.isNotEmpty()) { - topologyRefreshOptions.disableAdaptiveRefreshTrigger(*triggersToDisable) + // 连接 pub/sub 通道 + pubSubConnection = client.connectPubSub() + if (stopped.get()) { + pubSubConnection.closeAsync() + error("Redis cluster client was stopped during startup") + } + // 连接同步 + pool = ConnectionPoolSupport.createGenericObjectPool( + { + client.connect().apply { + if (redisConfig.enableSlaves) { + readFrom = redisConfig.slaves.readFrom + } + } + }, + redisConfig.pool.clusterPoolConfig() + ) + if (stopped.get()) { + pool.close() + error("Redis cluster client was stopped during startup") } + // 连接异步（同步方式创建） + asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool( + { + client.connectAsync(StringCodec.UTF8).whenComplete { value, _ -> + if (redisConfig.enableSlaves) { + value.readFrom = redisConfig.slaves.readFrom + } + } + }, + redisConfig.asyncPool.poolConfig() + ) + if (stopped.get()) { + asyncPool.closeAsync() + error("Redis cluster client was stopped during startup") + } + completeStart(completableFuture, autoRelease) + } catch (ex: Throwable) { + failStart(completableFuture, ex) + throw ex } + } - cluster.adaptiveRefreshTriggersTimeout?.toJavaDuration()?.let { topologyRefreshOptions.adaptiveRefreshTriggersTimeout(it) } - cluster.refreshPeriod?.toJavaDuration()?.let { topologyRefreshOptions.refreshPeriod(it) } - clientOptions - .topologyRefreshOptions(topologyRefreshOptions.build()) - .autoReconnect(redisConfig.autoReconnect) - .maxRedirects(cluster.maxRedirects) - .validateClusterNodeMembership(cluster.validateClusterNodeMembership) - .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) - - resources = resource.build() - client = RedisClusterClient.create(resources, uris) - client.setOptions(clientOptions.build()) - - // 连接 pub/sub 通道 - pubSubConnection = client.connectPubSub() - // 连接同步 - pool = ConnectionPoolSupport.createGenericObjectPool( - { client.connect().apply { - if (redisConfig.enableSlaves) { - val slaves = redisConfig.slaves - readFrom = slaves.readFrom - } - } }, - redisConfig.pool.clusterPoolConfig() - ) - // 连接异步（同步方式创建） - asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool( - { client.connectAsync(StringCodec.UTF8).whenComplete { v, _ -> - if (redisConfig.enableSlaves) { - val slaves = redisConfig.slaves - v.readFrom = slaves.readFrom - } - } }, - redisConfig.asyncPool.poolConfig() + override fun stop() { + if (!stopped.compareAndSet(false, true)) { + return + } + LettuceRedis.unregister(this) + startup.get()?.completeExceptionally(CancellationException("Redis cluster client is stopped")) + closeResources() + } + + private fun coordinateStart( + completableFuture: CompletableFuture, + autoRelease: Boolean, + vararg stages: CompletableFuture<*> + ) { + val coordinator = AsyncStartupCoordinator( + stages.size, + onSuccess = { completeStart(completableFuture, autoRelease) }, + onFailure = { failStart(completableFuture, it) }, + onSettled = { if (stopped.get()) closeResources() }, ) + stages.forEach { stage -> + stage.whenComplete { _, error -> coordinator.complete(error) } + } + } + + private fun completeStart(completableFuture: CompletableFuture, autoRelease: Boolean) { + if (stopped.get()) { + completableFuture.completeExceptionally(CancellationException("Redis cluster client is stopped")) + closeResources() + return + } if (autoRelease) { - LettuceRedis.clusterClients += this + LettuceRedis.register(this) } + if (stopped.get()) { + LettuceRedis.unregister(this) + completableFuture.completeExceptionally(CancellationException("Redis cluster client is stopped")) + closeResources() + return + } + completableFuture.complete(null) } - override fun stop() { - pubSubConnection.close() - asyncPool.close() - pool.close() - client.shutdown() - resources.shutdown() + private fun failStart(completableFuture: CompletableFuture, error: Throwable) { + stopped.set(true) + LettuceRedis.unregister(this) + completableFuture.completeExceptionally(error) + closeResources() + } + + private fun closeResources() { + val (clientToClose, resourcesToClose) = synchronized(lifecycleLock) { + val currentClient = if (::client.isInitialized) client else null + val currentResources = if (::resources.isInitialized) resources else null + currentClient to currentResources + } + if (clientToClose == null || !shutdownStarted.compareAndSet(false, true)) { + return + } + val closing = ArrayList>() + if (::pubSubConnection.isInitialized) { + runCatching { closing += pubSubConnection.closeAsync() } + } + if (::asyncPool.isInitialized) { + runCatching { closing += asyncPool.closeAsync() } + } + if (::pool.isInitialized) { + runCatching { pool.close() } + } + val connectionsClosed = if (closing.isEmpty()) { + CompletableFuture.completedFuture(null) + } else { + CompletableFuture.allOf(*closing.toTypedArray()) + } + connectionsClosed.handle { _, _ -> null }.thenCompose { + clientToClose.shutdownAsync() + }.whenComplete { _, _ -> + runCatching { resourcesToClose?.shutdown() } + } } override fun useCommands(block: (RedisClusterCommands) -> T): T? { diff --git a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedis.kt b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedis.kt index dcb3768a7..12a5a4c39 100644 --- a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedis.kt +++ b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedis.kt @@ -5,6 +5,75 @@ import taboolib.common.LifeCycle import taboolib.common.env.RuntimeDependencies import taboolib.common.env.RuntimeDependency import taboolib.common.platform.Awake +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger + +internal fun interface LettuceRedisResource { + + fun stop() +} + +internal class AsyncStartupCoordinator( + stageCount: Int, + private val onSuccess: () -> Unit, + private val onFailure: (Throwable) -> Unit, + private val onSettled: () -> Unit, +) { + + private val remaining = AtomicInteger(stageCount) + private val failed = AtomicBoolean(false) + + init { + require(stageCount > 0) { "stageCount must be positive" } + } + + fun complete(error: Throwable?) { + if (error != null && failed.compareAndSet(false, true)) { + onFailure(error) + } + onSettled() + if (remaining.decrementAndGet() == 0 && !failed.get()) { + onSuccess() + } + } +} + +internal class LettuceRedisResourceRegistry { + + private val closed = AtomicBoolean(false) + private val resources = ConcurrentHashMap.newKeySet() + + fun register(resource: LettuceRedisResource) { + if (closed.get()) { + runCatching { resource.stop() } + return + } + resources += resource + if (closed.get() && resources.remove(resource)) { + runCatching { resource.stop() } + } + } + + fun unregister(resource: LettuceRedisResource) { + resources.remove(resource) + } + + fun closeAll() { + if (!closed.compareAndSet(false, true)) { + return + } + resources.toList().forEach { resource -> + if (resources.remove(resource)) { + runCatching { resource.stop() } + } + } + } + + internal fun size(): Int { + return resources.size + } +} @Inject @RuntimeDependencies( @@ -107,16 +176,18 @@ import taboolib.common.platform.Awake ) object LettuceRedis { - internal val clients = mutableListOf() - internal val clusterClients = mutableListOf() + private val resources = LettuceRedisResourceRegistry() + + internal fun register(resource: LettuceRedisResource) { + resources.register(resource) + } + + internal fun unregister(resource: LettuceRedisResource) { + resources.unregister(resource) + } @Awake(LifeCycle.DISABLE) internal fun stop() { - clients.forEach { - it.stop() - } - clusterClients.forEach { - it.stop() - } + resources.closeAll() } } \ No newline at end of file diff --git a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedisClient.kt b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedisClient.kt index 0cad9bd92..01c066e59 100644 --- a/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedisClient.kt +++ b/module/database/database-lettuce-redis/src/main/kotlin/taboolib/expansion/LettuceRedisClient.kt @@ -25,165 +25,352 @@ import taboolib.expansion.lettuce.IRedisChannel import taboolib.expansion.lettuce.IRedisClient import taboolib.expansion.lettuce.IRedisCommand import taboolib.expansion.lettuce.IRedisPubSub +import java.util.concurrent.CancellationException import java.util.concurrent.CompletableFuture +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference @Suppress("DuplicatedCode") -class LettuceRedisClient(val redisConfig: LettuceRedisConfig): IRedisClient, IRedisChannel, IRedisCommand, IRedisPubSub { +class LettuceRedisClient(val redisConfig: LettuceRedisConfig): IRedisClient, IRedisChannel, IRedisCommand, IRedisPubSub, LettuceRedisResource { + @Volatile lateinit var client: RedisClient + @Volatile lateinit var pool: GenericObjectPool> + + @Volatile lateinit var asyncPool: BoundedAsyncPool> + @Volatile lateinit var masterReplicaPool: GenericObjectPool> + + @Volatile lateinit var masterAsyncReplicaPool: BoundedAsyncPool> + @Volatile lateinit var pubSubConnection: StatefulRedisPubSubConnection + + @Volatile lateinit var resources: DefaultClientResources var enabledSlaves = false + private val startup = AtomicReference?>() + private val stopped = AtomicBoolean(false) + private val shutdownStarted = AtomicBoolean(false) + private val lifecycleLock = Any() + override fun start(autoRelease: Boolean): CompletableFuture { val completableFuture = CompletableFuture() - val resource = DefaultClientResources.builder() - - if (redisConfig.ioThreadPoolSize != 0) { - resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + if (!startup.compareAndSet(null, completableFuture)) { + val existingStartup = startup.get()!! + if (stopped.get() && !existingStartup.isCompletedExceptionally && !existingStartup.isCancelled) { + return CompletableFuture().also { + it.completeExceptionally(CancellationException("Redis client is stopped")) + } + } + return existingStartup } - if (redisConfig.computationThreadPoolSize != 0) { - resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + if (stopped.get()) { + completableFuture.completeExceptionally(CancellationException("Redis client is stopped")) + return completableFuture } + try { + val resource = DefaultClientResources.builder() + if (redisConfig.ioThreadPoolSize != 0) { + resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + } + if (redisConfig.computationThreadPoolSize != 0) { + resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + } - val clientOptions = ClientOptions.builder() - .autoReconnect(redisConfig.autoReconnect) - .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) - - if (redisConfig.ssl) { - clientOptions.sslOptions(redisConfig.sslOptions) - } - val uri = redisConfig.redisURIBuilder().build() + val clientOptions = ClientOptions.builder() + .autoReconnect(redisConfig.autoReconnect) + .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) + if (redisConfig.ssl) { + clientOptions.sslOptions(redisConfig.sslOptions) + } + val uri = redisConfig.redisURIBuilder().build() - resources = resource.build() - client = RedisClient.create(resources, uri).apply { - options = clientOptions.build() - } - // 连接 pub/sub 通道 - pubSubConnection = client.connectPubSub() - - if (redisConfig.enableSlaves) { - enabledSlaves = true - val slaves = redisConfig.slaves - - // 连接同步 - masterReplicaPool = ConnectionPoolSupport.createGenericObjectPool( - { MasterReplica.connect(client, StringCodec.UTF8, uri).apply { - readFrom = slaves.readFrom - } }, - redisConfig.pool.slavesPoolConfig() - ) - // 连接异步 - AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( - { MasterReplica.connectAsync(client, StringCodec.UTF8, uri).whenComplete { v, _ -> - v.readFrom = slaves.readFrom - } }, - redisConfig.asyncPool.poolConfig() - ).thenAccept { - masterAsyncReplicaPool = it - completableFuture.complete(null) + val newResources = resource.build() + val newClient = try { + RedisClient.create(newResources, uri).apply { + options = clientOptions.build() + } + } catch (ex: Throwable) { + newResources.shutdown() + throw ex } - } else { - // 连接同步 - pool = ConnectionPoolSupport.createGenericObjectPool( - { client.connect() }, - redisConfig.pool.poolConfig() - ) - // 连接异步 - AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( - { client.connectAsync(StringCodec.UTF8, uri) }, - redisConfig.asyncPool.poolConfig() - ).thenAccept { - asyncPool = it - completableFuture.complete(null) + synchronized(lifecycleLock) { + resources = newResources + client = newClient } - } - if (autoRelease) { - LettuceRedis.clients += this + if (stopped.get()) { + closeResources() + return completableFuture + } + // 异步连接 pub/sub 通道，避免 start() 阻塞调用线程 + val pubSubReady = client.connectPubSubAsync(StringCodec.UTF8, uri).thenAccept { + pubSubConnection = it + if (stopped.get()) { + it.closeAsync() + } + }.toCompletableFuture() + + if (redisConfig.enableSlaves) { + enabledSlaves = true + val slaves = redisConfig.slaves + // 连接同步 + masterReplicaPool = ConnectionPoolSupport.createGenericObjectPool( + { + MasterReplica.connect(client, StringCodec.UTF8, uri).apply { + readFrom = slaves.readFrom + } + }, + redisConfig.pool.slavesPoolConfig() + ) + if (stopped.get()) { + masterReplicaPool.close() + } + // 连接异步 + val poolReady = AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( + { + MasterReplica.connectAsync(client, StringCodec.UTF8, uri).whenComplete { value, _ -> + value.readFrom = slaves.readFrom + } + }, + redisConfig.asyncPool.poolConfig() + ).thenAccept { + masterAsyncReplicaPool = it + if (stopped.get()) { + it.closeAsync() + } + }.toCompletableFuture() + coordinateStart(completableFuture, autoRelease, pubSubReady, poolReady) + } else { + // 连接同步 + pool = ConnectionPoolSupport.createGenericObjectPool( + { client.connect() }, + redisConfig.pool.poolConfig() + ) + if (stopped.get()) { + pool.close() + } + // 连接异步 + val poolReady = AsyncConnectionPoolSupport.createBoundedObjectPoolAsync( + { client.connectAsync(StringCodec.UTF8, uri) }, + redisConfig.asyncPool.poolConfig() + ).thenAccept { + asyncPool = it + if (stopped.get()) { + it.closeAsync() + } + }.toCompletableFuture() + coordinateStart(completableFuture, autoRelease, pubSubReady, poolReady) + } + } catch (ex: Throwable) { + failStart(completableFuture, ex) } return completableFuture } override fun startSync(autoRelease: Boolean) { - val resource = DefaultClientResources.builder() - - if (redisConfig.ioThreadPoolSize != 0) { - resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + val completableFuture = CompletableFuture() + if (!startup.compareAndSet(null, completableFuture)) { + val existingStartup = startup.get()!! + check(!stopped.get() && existingStartup.isDone && !existingStartup.isCompletedExceptionally && !existingStartup.isCancelled) { + "Redis client is already starting or failed to start" + } + return } - if (redisConfig.computationThreadPoolSize != 0) { - resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + if (stopped.get()) { + val error = IllegalStateException("Redis client is stopped") + completableFuture.completeExceptionally(error) + throw error } + try { + val resource = DefaultClientResources.builder() + if (redisConfig.ioThreadPoolSize != 0) { + resource.ioThreadPoolSize(redisConfig.ioThreadPoolSize) + } + if (redisConfig.computationThreadPoolSize != 0) { + resource.computationThreadPoolSize(redisConfig.computationThreadPoolSize) + } + + val clientOptions = ClientOptions.builder() + .autoReconnect(redisConfig.autoReconnect) + .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) + if (redisConfig.ssl) { + clientOptions.sslOptions(redisConfig.sslOptions) + } + val uri = redisConfig.redisURIBuilder().build() + + val newResources = resource.build() + val newClient = try { + RedisClient.create(newResources, uri).apply { + options = clientOptions.build() + } + } catch (ex: Throwable) { + newResources.shutdown() + throw ex + } + synchronized(lifecycleLock) { + resources = newResources + client = newClient + } + if (stopped.get()) { + closeResources() + error("Redis client was stopped during startup") + } + // 连接 pub/sub 通道 + pubSubConnection = client.connectPubSub() + if (stopped.get()) { + pubSubConnection.closeAsync() + error("Redis client was stopped during startup") + } - val clientOptions = ClientOptions.builder() - .autoReconnect(redisConfig.autoReconnect) - .pingBeforeActivateConnection(redisConfig.pingBeforeActivateConnection) + if (redisConfig.enableSlaves) { + enabledSlaves = true + val slaves = redisConfig.slaves + // 连接同步 + masterReplicaPool = ConnectionPoolSupport.createGenericObjectPool( + { + MasterReplica.connect(client, StringCodec.UTF8, uri).apply { + readFrom = slaves.readFrom + } + }, + redisConfig.pool.slavesPoolConfig() + ) + if (stopped.get()) { + masterReplicaPool.close() + error("Redis client was stopped during startup") + } + // 连接异步（同步方式创建） + masterAsyncReplicaPool = AsyncConnectionPoolSupport.createBoundedObjectPool( + { + MasterReplica.connectAsync(client, StringCodec.UTF8, uri).whenComplete { value, _ -> + value.readFrom = slaves.readFrom + } + }, + redisConfig.asyncPool.poolConfig() + ) + if (stopped.get()) { + masterAsyncReplicaPool.closeAsync() + error("Redis client was stopped during startup") + } + } else { + // 连接同步 + pool = ConnectionPoolSupport.createGenericObjectPool( + { client.connect() }, + redisConfig.pool.poolConfig() + ) + if (stopped.get()) { + pool.close() + error("Redis client was stopped during startup") + } + // 连接异步（同步方式创建） + asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool( + { client.connectAsync(StringCodec.UTF8, uri) }, + redisConfig.asyncPool.poolConfig() + ) + if (stopped.get()) { + asyncPool.closeAsync() + error("Redis client was stopped during startup") + } + } + completeStart(completableFuture, autoRelease) + } catch (ex: Throwable) { + failStart(completableFuture, ex) + throw ex + } + } - if (redisConfig.ssl) { - clientOptions.sslOptions(redisConfig.sslOptions) + override fun stop() { + if (!stopped.compareAndSet(false, true)) { + return } - val uri = redisConfig.redisURIBuilder().build() + LettuceRedis.unregister(this) + startup.get()?.completeExceptionally(CancellationException("Redis client is stopped")) + closeResources() + } - resources = resource.build() - client = RedisClient.create(resources, uri).apply { - options = clientOptions.build() + private fun coordinateStart( + completableFuture: CompletableFuture, + autoRelease: Boolean, + vararg stages: CompletableFuture<*> + ) { + val coordinator = AsyncStartupCoordinator( + stages.size, + onSuccess = { completeStart(completableFuture, autoRelease) }, + onFailure = { failStart(completableFuture, it) }, + onSettled = { if (stopped.get()) closeResources() }, + ) + stages.forEach { stage -> + stage.whenComplete { _, error -> coordinator.complete(error) } } - // 连接 pub/sub 通道 - pubSubConnection = client.connectPubSub() - - if (redisConfig.enableSlaves) { - enabledSlaves = true - val slaves = redisConfig.slaves - - // 连接同步 - masterReplicaPool = ConnectionPoolSupport.createGenericObjectPool( - { MasterReplica.connect(client, StringCodec.UTF8, uri).apply { - readFrom = slaves.readFrom - } }, - redisConfig.pool.slavesPoolConfig() - ) - // 连接异步（同步方式创建） - masterAsyncReplicaPool = AsyncConnectionPoolSupport.createBoundedObjectPool( - { MasterReplica.connectAsync(client, StringCodec.UTF8, uri).whenComplete { v, _ -> - v.readFrom = slaves.readFrom - } }, - redisConfig.asyncPool.poolConfig() - ) - } else { - // 连接同步 - pool = ConnectionPoolSupport.createGenericObjectPool( - { client.connect() }, - redisConfig.pool.poolConfig() - ) - // 连接异步（同步方式创建） - asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool( - { client.connectAsync(StringCodec.UTF8, uri) }, - redisConfig.asyncPool.poolConfig() - ) + } + + private fun completeStart(completableFuture: CompletableFuture, autoRelease: Boolean) { + if (stopped.get()) { + completableFuture.completeExceptionally(CancellationException("Redis client is stopped")) + closeResources() + return } if (autoRelease) { - LettuceRedis.clients += this + LettuceRedis.register(this) } + if (stopped.get()) { + LettuceRedis.unregister(this) + completableFuture.completeExceptionally(CancellationException("Redis client is stopped")) + closeResources() + return + } + completableFuture.complete(null) } - override fun stop() { - pubSubConnection.close() - if (enabledSlaves) { - masterAsyncReplicaPool.close() - masterReplicaPool.close() + private fun failStart(completableFuture: CompletableFuture, error: Throwable) { + stopped.set(true) + LettuceRedis.unregister(this) + completableFuture.completeExceptionally(error) + closeResources() + } + + private fun closeResources() { + val (clientToClose, resourcesToClose) = synchronized(lifecycleLock) { + val currentClient = if (::client.isInitialized) client else null + val currentResources = if (::resources.isInitialized) resources else null + currentClient to currentResources + } + if (clientToClose == null || !shutdownStarted.compareAndSet(false, true)) { + return + } + val closing = ArrayList>() + if (::pubSubConnection.isInitialized) { + runCatching { closing += pubSubConnection.closeAsync() } + } + if (::masterAsyncReplicaPool.isInitialized) { + runCatching { closing += masterAsyncReplicaPool.closeAsync() } + } + if (::asyncPool.isInitialized) { + runCatching { closing += asyncPool.closeAsync() } + } + if (::masterReplicaPool.isInitialized) { + runCatching { masterReplicaPool.close() } + } + if (::pool.isInitialized) { + runCatching { pool.close() } + } + val connectionsClosed = if (closing.isEmpty()) { + CompletableFuture.completedFuture(null) } else { - asyncPool.close() - pool.close() + CompletableFuture.allOf(*closing.toTypedArray()) + } + connectionsClosed.handle { _, _ -> null }.thenCompose { + clientToClose.shutdownAsync() + }.whenComplete { _, _ -> + runCatching { resourcesToClose?.shutdown() } } - client.shutdown() - resources.shutdown() } override fun useCommands(block: (RedisCommands) -> T): T? { diff --git a/module/database/database-lettuce-redis/src/test/kotlin/taboolib/expansion/LettuceRedisResourceRegistryTest.kt b/module/database/database-lettuce-redis/src/test/kotlin/taboolib/expansion/LettuceRedisResourceRegistryTest.kt new file mode 100644 index 000000000..1ec753468 --- /dev/null +++ b/module/database/database-lettuce-redis/src/test/kotlin/taboolib/expansion/LettuceRedisResourceRegistryTest.kt @@ -0,0 +1,137 @@ +package taboolib.expansion + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import taboolib.library.configuration.ConfigurationSection +import java.lang.reflect.Proxy +import java.util.concurrent.atomic.AtomicInteger + +class LettuceRedisResourceRegistryTest { + + @Test + fun `registered clients stop exactly once`() { + val registry = LettuceRedisResourceRegistry() + val stopCount = AtomicInteger() + val resource = LettuceRedisResource { stopCount.incrementAndGet() } + + registry.register(resource) + registry.register(resource) + registry.closeAll() + registry.closeAll() + + assertEquals(1, stopCount.get()) + assertEquals(0, registry.size()) + } + + @Test + fun `unregistered client remains caller managed`() { + val registry = LettuceRedisResourceRegistry() + val stopCount = AtomicInteger() + val resource = LettuceRedisResource { stopCount.incrementAndGet() } + + registry.register(resource) + registry.unregister(resource) + registry.closeAll() + + assertEquals(0, stopCount.get()) + } + + @Test + fun `client registered after shutdown stops immediately`() { + val registry = LettuceRedisResourceRegistry() + val stopCount = AtomicInteger() + + registry.closeAll() + registry.register(LettuceRedisResource { stopCount.incrementAndGet() }) + + assertEquals(1, stopCount.get()) + assertEquals(0, registry.size()) + } + + @Test + fun `startup coordinator reports the first failure immediately`() { + val successCount = AtomicInteger() + val failureCount = AtomicInteger() + val settledCount = AtomicInteger() + val coordinator = AsyncStartupCoordinator( + stageCount = 2, + onSuccess = { successCount.incrementAndGet() }, + onFailure = { failureCount.incrementAndGet() }, + onSettled = { settledCount.incrementAndGet() }, + ) + + coordinator.complete(IllegalStateException("failed")) + + assertEquals(0, successCount.get()) + assertEquals(1, failureCount.get()) + assertEquals(1, settledCount.get()) + + coordinator.complete(null) + + assertEquals(0, successCount.get()) + assertEquals(1, failureCount.get()) + assertEquals(2, settledCount.get()) + } + + @Test + fun `startup coordinator succeeds after every stage settles`() { + val successCount = AtomicInteger() + val failureCount = AtomicInteger() + val coordinator = AsyncStartupCoordinator( + stageCount = 2, + onSuccess = { successCount.incrementAndGet() }, + onFailure = { failureCount.incrementAndGet() }, + onSettled = {}, + ) + + coordinator.complete(null) + assertEquals(0, successCount.get()) + + coordinator.complete(null) + assertEquals(1, successCount.get()) + assertEquals(0, failureCount.get()) + } + + @Test + fun `synchronous start after stop throws`() { + val client = LettuceRedisClient(testConfig()) + client.stop() + + assertThrows(IllegalStateException::class.java) { + client.startSync() + } + } + + @Test + fun `asynchronous start after stop returns failed future`() { + val client = LettuceRedisClient(testConfig()) + client.stop() + + assertTrue(client.start().isCompletedExceptionally) + } + + private fun testConfig(): LettuceRedisConfig { + val configuration = Proxy.newProxyInstance( + ConfigurationSection::class.java.classLoader, + arrayOf(ConfigurationSection::class.java), + ) { _, method, args -> + when (method.name) { + "getString" -> when (args?.getOrNull(0)) { + "host" -> "127.0.0.1" + "timeout" -> "1s" + else -> args?.getOrNull(1) + } + "getInt" -> args?.getOrNull(1) ?: 0 + "getBoolean" -> args?.getOrNull(1) ?: false + "getConfigurationSection" -> null + "getKeys" -> emptySet() + "getStringList", "getEnumList" -> emptyList() + "contains" -> false + else -> null + } + } as ConfigurationSection + return LettuceRedisConfig(configuration) + } +} diff --git a/module/database/database-player-redis/src/main/kotlin/taboolib/expansion/RedisDatabaseHandler.kt b/module/database/database-player-redis/src/main/kotlin/taboolib/expansion/RedisDatabaseHandler.kt index 1fdbb0925..bd66e7148 100644 --- a/module/database/database-player-redis/src/main/kotlin/taboolib/expansion/RedisDatabaseHandler.kt +++ b/module/database/database-player-redis/src/main/kotlin/taboolib/expansion/RedisDatabaseHandler.kt @@ -5,6 +5,7 @@ import taboolib.common.platform.function.getDataFolder import taboolib.common.platform.function.pluginId import taboolib.library.configuration.ConfigurationSection import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean /** * 创建 Redis 数据管理器 @@ -32,12 +33,14 @@ class RedisDatabaseHandler( clearFlags: Boolean = false, ssl: String? = null, dataFile: String = "data.db", -) { +) : AutoCloseable { val database: Database private var connector: SingleRedisConnector? = null var connection: SingleRedisConnection? = null + private val closed = AtomicBoolean(false) + /** * 玩家Redis数据容器。 * @@ -48,17 +51,24 @@ class RedisDatabaseHandler( val redisDataContainer = ConcurrentHashMap() init { - table = conf.getConfigurationSection("Database")!!.getString("table", pluginId)!! - database = if (conf.getBoolean("enable")) { - buildPlayerDatabase(conf, table, flags, clearFlags, ssl) + val databaseConfig = conf.getConfigurationSection("Database")!! + table = databaseConfig.getString("table", table.ifEmpty { pluginId })!! + database = if (databaseConfig.getBoolean("enable")) { + buildPlayerDatabase(databaseConfig, table, flags, clearFlags, ssl) } else { buildPlayerDatabase(newFile(getDataFolder(), dataFile), table) } - val redis = conf.getConfigurationSection("Redis")!! - if (redis.getBoolean("enable")) { - connector = AlkaidRedis.create().fromConfig(redis) - connection?.close() - connection = connector!!.connect().connection() + try { + val redis = conf.getConfigurationSection("Redis")!! + if (redis.getBoolean("enable")) { + val newConnector = AlkaidRedis.create().fromConfig(redis) + connector = newConnector + connection = newConnector.connect().connection() + } + } catch (ex: Throwable) { + connector?.close() + database.close() + throw ex } } @@ -84,4 +94,37 @@ class RedisDatabaseHandler( redisDataContainer.remove(user) } + /** + * 释放 Redis 连接、连接器以及当前处理器拥有的数据库连接池。 + */ + override fun close() { + if (!closed.compareAndSet(false, true)) { + return + } + redisDataContainer.clear() + val currentConnection = connection + val currentConnector = connector + connection = null + connector = null + + var failure: Throwable? = null + fun closeResource(resource: AutoCloseable?) { + try { + resource?.close() + } catch (ex: Throwable) { + val firstFailure = failure + if (firstFailure == null) { + failure = ex + } else { + firstFailure.addSuppressed(ex) + } + } + } + + closeResource(currentConnection) + closeResource(currentConnector) + closeResource(database) + failure?.let { throw it } + } + } diff --git a/module/database/database-player/build.gradle.kts b/module/database/database-player/build.gradle.kts index 360bd131d..b76633d1d 100644 --- a/module/database/database-player/build.gradle.kts +++ b/module/database/database-player/build.gradle.kts @@ -5,4 +5,12 @@ dependencies { compileOnly(project(":module:database")) compileOnly(project(":module:basic:basic-configuration")) compileOnly("ink.ptms.core:v11701:11701-minimize:universal") + + testImplementation(project(":common")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) + testImplementation(project(":module:database")) + testImplementation(project(":module:basic:basic-configuration")) + testImplementation("com.zaxxer:HikariCP:4.0.3") + testImplementation("org.xerial:sqlite-jdbc:3.42.0.0") } \ No newline at end of file diff --git a/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt b/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt index 369f29f36..35b812949 100644 --- a/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt +++ b/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt @@ -1,12 +1,25 @@ package taboolib.expansion +import java.util.IdentityHashMap import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean import javax.sql.DataSource -class Database(val type: Type, val dataSource: DataSource = type.host().createDataSource()) { +class Database(val type: Type, val dataSource: DataSource = createOwnedDataSource(type)) : AutoCloseable { + + val ownsDataSource = takeOwnership(dataSource) + + private val closed = AtomicBoolean(false) + + constructor(type: Type, dataSource: DataSource, ownsDataSource: Boolean) : this(type, markOwnership(dataSource, ownsDataSource)) init { - type.tableVar().createTable(dataSource) + try { + type.tableVar().createTable(dataSource) + } catch (ex: Throwable) { + close() + throw ex + } } /** @@ -113,4 +126,49 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa where("user" eq user and ("key" eq key)) } } + + /** + * 关闭由当前实例创建的数据源。 + * + * 外部传入的数据源默认由调用方管理，可通过三参数构造函数显式转移所有权。 + */ + override fun close() { + if (!closed.compareAndSet(false, true) || !ownsDataSource) { + return + } + (dataSource as? AutoCloseable)?.close() + } + + companion object { + + private val ownedDataSources = ThreadLocal.withInitial { IdentityHashMap() } + + private fun createOwnedDataSource(type: Type): DataSource { + return type.host().createDataSource().also { + ownedDataSources.get()[it] = Unit + } + } + + private fun markOwnership(dataSource: DataSource, ownsDataSource: Boolean): DataSource { + val ownership = ownedDataSources.get() + if (ownsDataSource) { + ownership[dataSource] = Unit + } else { + ownership.remove(dataSource) + } + if (ownership.isEmpty()) { + ownedDataSources.remove() + } + return dataSource + } + + private fun takeOwnership(dataSource: DataSource): Boolean { + val ownership = ownedDataSources.get() + val ownsDataSource = ownership.remove(dataSource) != null + if (ownership.isEmpty()) { + ownedDataSources.remove() + } + return ownsDataSource + } + } } \ No newline at end of file diff --git a/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt new file mode 100644 index 000000000..134ac11c6 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt @@ -0,0 +1,7 @@ +package com.zaxxer.hikari_4_0_3 + +/** + * 测试环境使用 database 模块 shadow 产物，字节码会引用重定位后的 HikariConfig。 + * 生产环境由 TabooLib 运行时依赖提供，这里只在 test classpath 代理到原始 Hikari。 + */ +class HikariConfig : com.zaxxer.hikari.HikariConfig() diff --git a/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt new file mode 100644 index 000000000..b6ab0ba52 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt @@ -0,0 +1,7 @@ +package com.zaxxer.hikari_4_0_3 + +/** + * 测试环境使用 database 模块 shadow 产物，字节码会引用重定位后的 HikariDataSource。 + * 生产环境由 TabooLib 运行时依赖提供，这里只在 test classpath 代理到原始 Hikari。 + */ +class HikariDataSource(config: HikariConfig) : com.zaxxer.hikari.HikariDataSource(config) diff --git a/module/database/database-player/src/test/kotlin/taboolib/expansion/DatabaseLifecycleTest.kt b/module/database/database-player/src/test/kotlin/taboolib/expansion/DatabaseLifecycleTest.kt new file mode 100644 index 000000000..442cbfb28 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/taboolib/expansion/DatabaseLifecycleTest.kt @@ -0,0 +1,75 @@ +package taboolib.expansion + +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import taboolib.module.configuration.Configuration +import java.lang.reflect.Proxy +import java.nio.file.Path +import javax.sql.DataSource + +class DatabaseLifecycleTest { + + @TempDir + lateinit var tempDir: Path + + @BeforeEach + fun setupDatabaseSettings() { + taboolib.module.database.Database.settingsFile = Proxy.newProxyInstance( + Configuration::class.java.classLoader, + arrayOf(Configuration::class.java), + ) { _, method, args -> + when (method.name) { + "contains" -> false + "getBoolean", "getInt", "getLong", "getString" -> args?.getOrNull(1) + "getConfigurationSection", "getFile" -> null + "getReloadGeneration" -> 0 + "saveToString" -> "" + else -> null + } + } as Configuration + } + + @Test + fun `default data source is owned and closed idempotently`() { + val database = Database(TypeSQLite(tempDir.resolve("owned.db").toFile(), "owned_data")) + val dataSource = database.dataSource + + assertTrue(database.ownsDataSource) + database.close() + database.close() + + assertTrue(dataSource.isClosed()) + } + + @Test + fun `injected data source remains caller owned by default`() { + val type = TypeSQLite(tempDir.resolve("borrowed.db").toFile(), "borrowed_data") + val dataSource = type.host().createDataSource(autoRelease = false) + val database = Database(type, dataSource) + + assertFalse(database.ownsDataSource) + database.close() + + assertFalse(dataSource.isClosed()) + (dataSource as AutoCloseable).close() + } + + @Test + fun `injected data source can transfer ownership explicitly`() { + val type = TypeSQLite(tempDir.resolve("transferred.db").toFile(), "transferred_data") + val dataSource = type.host().createDataSource(autoRelease = false) + val database = Database(type, dataSource, ownsDataSource = true) + + assertTrue(database.ownsDataSource) + database.close() + + assertTrue(dataSource.isClosed()) + } + + private fun DataSource.isClosed(): Boolean { + return javaClass.getMethod("isClosed").invoke(this) as Boolean + } +}