使用 JMS 线程池协程复制线程

Question

我有一些逻辑，它使用了 JMS 队列线程池。总体情况：

[request thread] -> jms -> JMS listener [thread pool] -> reactive web client

现在我收到了重复的消息（例如 - 错误）

ERROR [DefaultMessageListenerContainer-5 @coroutine#2]
ERROR [DefaultMessageListenerContainer-4 @coroutine#4]
ERROR [DefaultMessageListenerContainer-1 @coroutine#1]
ERROR [DefaultMessageListenerContainer-3 @coroutine#3]

此类协程由 runBlocking 启动。

@JmsListener(destination = ...)
fun someHandleFunction(...) = runBlocking {
   someWebClientService.createRequest(...)
}

// SomeWebClientService class
suspend fun createRequest(...) = try {
   val response = webClient
       .post()
       .uri { ... }
       ...
       .exchange()
       .awaitFirst()

   // Check response status here
} catch (e: Exception) { 
   // Some error handling here
}

是的，我知道在 runBlockin 协程的情况下可以 behave in such way... 但是为什么协程线程与新线程关联？

这种重复有什么问题？

Answer 1

问题与 JMS 配置有关。

出于某种原因，如果没有此配置，单独的 JMS 线程将处理一个任务

spring:
     artemis.embedded:
           persistent: true
           data-directory: ./target/some-queue-storage
           queues: ...

使用 JMS 线程池协程复制线程

Coroutines duplicated threads with JMS thread pool

reactive-programming

kotlin

kotlin-coroutines