From c3acb73c2233e1064b82decc86499a81e842089a Mon Sep 17 00:00:00 2001 From: twisti <76837088+twisti-dev@users.noreply.github.com> Date: Mon, 1 Jun 2026 19:22:56 +0200 Subject: [PATCH 1/2] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor(rabbitmq):=20?= =?UTF-8?q?restructure=20RabbitClient=20to=20manage=20shared=20resources?= =?UTF-8?q?=20with=20ExecutorService?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - introduce sharedConsumerExecutor for managing consumer threads - rename closeEventLoopGroup to closeSharedResources for clarity - update shutdown logic to ensure proper resource cleanup --- .../rabbitmq/common/RabbitMQCommonInstance.kt | 4 +- .../common/connection/client/RabbitClient.kt | 40 ++++++++++++++----- 2 files changed, 33 insertions(+), 11 deletions(-) diff --git a/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/RabbitMQCommonInstance.kt b/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/RabbitMQCommonInstance.kt index 702730d..1645ce4 100644 --- a/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/RabbitMQCommonInstance.kt +++ b/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/RabbitMQCommonInstance.kt @@ -21,11 +21,11 @@ abstract class RabbitMQCommonInstance : RabbitMQInstance { @MustBeInvokedByOverriders open suspend fun onDisable() { withContext(NonCancellable) { - RabbitClient.closeEventLoopGroup() + RabbitClient.closeSharedResources() } } companion object { fun get(): RabbitMQCommonInstance = RabbitMQInstance.instance as RabbitMQCommonInstance } -} \ No newline at end of file +} diff --git a/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/connection/client/RabbitClient.kt b/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/connection/client/RabbitClient.kt index b1a1c5b..af13b14 100644 --- a/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/connection/client/RabbitClient.kt +++ b/surf-rabbitmq-common/src/main/kotlin/dev/slne/surf/rabbitmq/common/connection/client/RabbitClient.kt @@ -23,11 +23,11 @@ import io.netty.channel.socket.nio.NioSocketChannel import io.netty.channel.uring.IoUring import io.netty.channel.uring.IoUringIoHandler import io.netty.channel.uring.IoUringSocketChannel -import kotlinx.coroutines.delay +import org.jetbrains.annotations.Blocking import java.lang.AutoCloseable +import java.time.Duration import java.time.Instant -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.ConcurrentLinkedQueue +import java.util.concurrent.* import kotlin.time.Duration.Companion.seconds class RabbitClient private constructor( @@ -79,6 +79,7 @@ class RabbitClient private constructor( } private val sharedEventLoopGroup: MultiThreadIoEventLoopGroup + private val sharedConsumerExecutor: ExecutorService private val activeClients = ConcurrentHashMap() init { @@ -100,6 +101,22 @@ class RabbitClient private constructor( .factory() sharedEventLoopGroup = MultiThreadIoEventLoopGroup(8, nettyThreadFactory, transport.ioHandlerFactory) + sharedConsumerExecutor = Executors.newFixedThreadPool( + 16, + Thread.ofPlatform() + .name("rabbitmq-consumer-thread-", 0) + .uncaughtExceptionHandler { thread, throwable -> + log.atSevere() + .withCause(throwable) + .log( + "Uncaught exception in RabbitMQ consumer thread (%s): %s", + thread.name, + throwable + ) + } + .daemon() + .factory() + ) } fun create( @@ -121,6 +138,7 @@ class RabbitClient private constructor( requestedHeartbeat = 60 connectionTimeout = config.getTimeout().seconds.inWholeMilliseconds.toInt() + setSharedExecutor(sharedConsumerExecutor) netty().eventLoopGroup(sharedEventLoopGroup) netty().bootstrapCustomizer { bootstrap -> bootstrap.channel(transport.channelClass) @@ -156,14 +174,14 @@ class RabbitClient private constructor( return client } - - suspend fun closeEventLoopGroup() { + @Blocking + fun closeSharedResources() { val stillActive = activeClients.values.toList() if (stillActive.isNotEmpty()) { log.atWarning() .log( - "RabbitMQ Netty EventLoopGroup is being shut down while %s RabbitClient(s) are still active. " + + "RabbitMQ shared resources are being shut down while %s RabbitClient(s) are still active. " + "These plugins probably did not call RabbitMQApi.disconnect(): %s", stillActive.size, stillActive.joinToString { it.connectionName } @@ -172,11 +190,11 @@ class RabbitClient private constructor( log.atWarning() .log( "Any RabbitMQ connection recovery errors that appear after this message are expected follow-up " + - "errors caused by shutting down the shared Netty EventLoopGroup while RabbitMQ clients are " + + "errors caused by shutting down shared RabbitMQ resources while RabbitMQ clients are " + "still active. Fix the plugins listed above by calling RabbitMQApi.disconnect() during shutdown." ) - delay(10.seconds) + Thread.sleep(Duration.ofSeconds(10)) stillActive.forEach { info -> log.atWarning() @@ -193,6 +211,10 @@ class RabbitClient private constructor( } } + sharedConsumerExecutor.shutdown() + if (!sharedConsumerExecutor.awaitTermination(10, TimeUnit.SECONDS)) { + sharedConsumerExecutor.shutdownNow() + } sharedEventLoopGroup.shutdownGracefully().syncUninterruptibly() } } @@ -242,4 +264,4 @@ class RabbitClient private constructor( activeClients.remove(this) } } -} \ No newline at end of file +} From 955318d8ef609567cc65892448ef54c6f7d78ed4 Mon Sep 17 00:00:00 2001 From: twisti <76837088+twisti-dev@users.noreply.github.com> Date: Mon, 1 Jun 2026 19:23:06 +0200 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=94=A7=20chore:=20update=20version=20?= =?UTF-8?q?to=201.5.3=20in=20gradle.properties?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gradle.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gradle.properties b/gradle.properties index 3e78762..745fb7b 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,4 +1,4 @@ kotlin.code.style=official kotlin.stdlib.default.dependency=false org.gradle.parallel=true -version=1.5.2 \ No newline at end of file +version=1.5.3 \ No newline at end of file