feat: 引擎等MQ场景接入SCS框架 TencentBlueKing#7443 增加延迟队列声明

royalhuang · Sep 27, 2022 · 12b9a21 · 12b9a21
1 parent afd0710
commit 12b9a21
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 3 deletions.
diff --git a/...common-event/src/main/kotlin/com/tencent/devops/common/event/dispatcher/pipeline/mq/MQ.kt b/...common-event/src/main/kotlin/com/tencent/devops/common/event/dispatcher/pipeline/mq/MQ.kt
@@ -217,6 +217,11 @@ object MQ {
     const val ROUTE_MEASURE_REQUEST_EVENT = "r.measure.request.event"
     const val QUEUE_MEASURE_REQUEST_EVENT = "q.measure.request.event"
 
+    // 流水线设置发生变化的事件广播
+    const val EXCHANGE_PIPELINE_SETTING_CHANGE_FANOUT = "e.engine.pipeline.setting.change.fanout"
+    const val ROUTE_PIPELINE_SETTING_CHANGE = "r.engine.pipeline.setting.change"
+    const val QUEUE_PIPELINE_SETTING_CHANGE = "q.engine.pipeline.setting.change"
+
     // webSocket消息
     const val EXCHANGE_WEBSOCKET_TMP_FANOUT = "e.websocket.fanout"
     const val ROUTE_WEBSOCKET_TMP_EVENT = "r.websocket.file"

diff --git a/...n/kotlin/com/tencent/devops/common/stream/config/StreamBindingEnvironmentPostProcessor.kt b/...n/kotlin/com/tencent/devops/common/stream/config/StreamBindingEnvironmentPostProcessor.kt
@@ -111,9 +111,12 @@ class StreamBindingEnvironmentPostProcessor : EnvironmentPostProcessor, Ordered
                 definition.add(bindingName)
                 // 如果注解中指定了订阅组，则直接设置
                 // 如果未指定则取当前服务名作为订阅组，保证所有分布式服务再同一个组内
-                val prefix = "spring.cloud.stream.bindings.$bindingName-in-0"
-                setProperty("$prefix.destination", streamConsumer.destination)
-                setProperty("$prefix.group", streamConsumer.group)
+                val bindingPrefix = "spring.cloud.stream.bindings.$bindingName-in-0"
+                val rabbitPropPrefix = "spring.cloud.stream.rabbit.bindings.$bindingName-in-0"
+                setProperty("$bindingPrefix.destination", streamConsumer.destination)
+                setProperty("$bindingPrefix.group", streamConsumer.group)
+                setProperty("$bindingPrefix.group", streamConsumer.group)
+                setProperty("$rabbitPropPrefix.consumer.delayed-exchange", "true")
             }
 
             // 声明所有扫描结果的函数式声明