pref:微服务的主机IP列表队列配置自动删除 #9663

src/backend/ci/core/common/common-db-sharding/src/main/kotlin/com/tencent/devops/common/db/config/BkShardingMQConfiguration.kt

@@ -31,7 +31,6 @@ import com.tencent.devops.common.db.listener.BkShardingRoutingRuleListener
 import com.tencent.devops.common.event.dispatcher.pipeline.mq.MQ
 import com.tencent.devops.common.event.dispatcher.pipeline.mq.Tools
 import com.tencent.devops.common.service.utils.BkServiceUtil
-import org.slf4j.LoggerFactory
 import org.springframework.amqp.core.Binding
 import org.springframework.amqp.core.BindingBuilder
 import org.springframework.amqp.core.FanoutExchange
@@ -53,17 +52,13 @@ import org.springframework.core.Ordered
 @AutoConfigureAfter(BkShardingRoutingRuleListener::class)
 class BkShardingMQConfiguration {
 
-    companion object {
-        private val logger = LoggerFactory.getLogger(BkShardingMQConfiguration::class.java)
-    }
-
     @Bean
     fun rabbitAdmin(connectionFactory: ConnectionFactory): RabbitAdmin {
         return RabbitAdmin(connectionFactory)
     }
 
     @Bean
-    fun shardingRoutingRuleQueue() = Queue(BkServiceUtil.getDynamicMqQueue())
+    fun shardingRoutingRuleQueue() = Queue(BkServiceUtil.getDynamicMqQueue(), true, false, true)
 
     @Bean
     fun shardingRoutingRuleExchange(): FanoutExchange {
@@ -86,8 +81,6 @@ class BkShardingMQConfiguration {
         @Autowired shardingRoutingRuleQueue: Queue,
         @Autowired bkShardingRoutingRuleListener: BkShardingRoutingRuleListener
     ): SimpleMessageListenerContainer {
-        // 增加动态队列清理钩子
-        addDynamicMqQueueClearHook(rabbitAdmin, shardingRoutingRuleQueue)
         val adapter = MessageListenerAdapter(bkShardingRoutingRuleListener, bkShardingRoutingRuleListener::execute.name)
         adapter.setMessageConverter(messageConverter)
         return Tools.createSimpleMessageListenerContainerByAdapter(
@@ -101,27 +94,4 @@ class BkShardingMQConfiguration {
             maxConcurrency = 5
         )
     }
-
-    private fun addDynamicMqQueueClearHook(rabbitAdmin: RabbitAdmin, shardingRoutingRuleQueue: Queue) {
-        try {
-            Runtime.getRuntime().addShutdownHook(object : Thread() {
-                override fun run() {
-                    val queueName = shardingRoutingRuleQueue.name
-                    logger.info("delete dynamicMqQueue($queueName) start!")
-                    rabbitAdmin.purgeQueue(queueName)
-                    rabbitAdmin.deleteQueue(queueName)
-                    val queueProperties = rabbitAdmin.getQueueProperties(queueName)
-                    logger.info("delete dynamicMqQueue($queueName) queueProperties:$queueProperties")
-                    if (queueProperties != null) {
-                        // 队列属性不为空说明删除未成功，打印失败日志
-                        logger.info("delete dynamicMqQueue($queueName) fail!")
-                        return
-                    }
-                    logger.info("delete dynamicMqQueue($queueName) success!")
-                }
-            })
-        } catch (t: Throwable) {
-            logger.warn("Fail to add dynamicMqQueueClear shutdown hook", t)
-        }
-    }
 }

src/backend/ci/core/common/common-web/src/main/kotlin/com/tencent/devops/common/web/WebAutoConfiguration.kt

@@ -116,8 +116,7 @@ class WebAutoConfiguration {
     ) = BkServiceInstanceApplicationRunner(
         compositeDiscoveryClient = compositeDiscoveryClient,
         bkTag = bkTag,
-        redisOperation = redisOperation,
-        rabbitAdmin = rabbitAdmin
+        redisOperation = redisOperation
     )
 
     @Bean

src/backend/ci/core/common/common-web/src/main/kotlin/com/tencent/devops/common/web/runner/BkServiceInstanceApplicationRunner.kt

@@ -6,19 +6,15 @@ import com.tencent.devops.common.service.BkTag
 import com.tencent.devops.common.service.utils.BkServiceUtil
 import com.tencent.devops.common.service.utils.CommonUtils
 import com.tencent.devops.common.service.utils.KubernetesUtils
-import com.tencent.devops.common.service.utils.SpringContextUtil
 import org.slf4j.LoggerFactory
-import org.springframework.amqp.rabbit.core.RabbitAdmin
 import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
 import org.springframework.cloud.client.discovery.composite.CompositeDiscoveryClient
-import org.springframework.core.env.Environment
 
 class BkServiceInstanceApplicationRunner constructor(
     private val compositeDiscoveryClient: CompositeDiscoveryClient,
     private val bkTag: BkTag,
-    private val redisOperation: RedisOperation,
-    private val rabbitAdmin: RabbitAdmin
+    private val redisOperation: RedisOperation
 ) : ApplicationRunner {
 
     companion object {
@@ -47,53 +43,11 @@ class BkServiceInstanceApplicationRunner constructor(
                         )
                     }"
                 )
-                val environment: Environment = SpringContextUtil.getBean(Environment::class.java)
-                val shardingDbFlag = environment.containsProperty("spring.datasource.dataSourceConfigs[0].url")
-                logger.info("initServiceHostInfo serviceName:$serviceName shardingDbFlag:$shardingDbFlag")
-                if (shardingDbFlag) {
-                    // 如果微服务的DB采用了分库分表方案，那么需要清理规则动态队列
-                    try {
-                        deleteDynamicMqQueue(cacheKey, serviceName, serviceHosts)
-                    } catch (ignored: Throwable) {
-                        logger.warn(
-                            "serviceName:$serviceName delete dynamicMqQueue fail!", ignored
-                        )
-                    }
-                }
                 // 清空redis中微服务的主机IP列表
                 redisOperation.delete(cacheKey)
                 // 把微服务的最新主机IP列表写入redis中
                 redisOperation.sadd(cacheKey, *serviceHosts)
             }
-
-            private fun deleteDynamicMqQueue(cacheKey: String, serviceName: String, serviceHosts: Array<String>) {
-                // 将微服务在redis的主机IP列表取出
-                val historyServiceHosts = redisOperation.getSetMembers(cacheKey)?.toMutableSet()
-                logger.info("initServiceHostInfo serviceName:$serviceName historyServiceHosts:$historyServiceHosts")
-                if (historyServiceHosts.isNullOrEmpty()) {
-                    return
-                }
-                historyServiceHosts.removeAll(serviceHosts.toSet())
-                historyServiceHosts.forEach { historyServiceHost ->
-                    val queueName = BkServiceUtil.getDynamicMqQueue(serviceName, historyServiceHost)
-                    logger.info("serviceName:$serviceName delete dynamicMqQueue($queueName) start!")
-                    var queueProperties = rabbitAdmin.getQueueProperties(queueName)
-                    if (queueProperties == null) {
-                        // 队列属性为空说明删除成功
-                        logger.info("serviceName:$serviceName dynamicMqQueue($queueName) does not exist!")
-                        return
-                    }
-                    rabbitAdmin.purgeQueue(queueName)
-                    rabbitAdmin.deleteQueue(queueName)
-                    queueProperties = rabbitAdmin.getQueueProperties(queueName)
-                    if (queueProperties != null) {
-                        // 队列属性不为空说明删除未成功，把队列名称写入redis中
-                        logger.info("serviceName:$serviceName delete dynamicMqQueue($queueName) fail!")
-                    } else {
-                        logger.info("serviceName:$serviceName delete dynamicMqQueue($queueName) success!")
-                    }
-                }
-            }
         }.start()
     }
 }