kotlin - 如何在 Kotlin 中实现方法上的互斥并优先处理一个线程？

Question

我有两个 kafka 主题 my_priorized_topic 和 my_not_so_priorized_topic。我想在 EventProcessor.doLogic 上设置互斥锁，并且始终优先处理来自 my_prioritized_topic 的消息，然后再处理来自 my_not_so_prioritized_topic

的消息

谁能给我一些建议，告诉我如何使用 Kotlin 解决这个问题，也许是协程？

class EventProcessor {
  fun doLogic(message: String) {
    ... // code which cannot be parallelized
  }
}

class KafkaConsumers(private val eventProcessor: EventProcessor) {
  @KafkaConsumer(topic = "my_priorized_topic")
  fun consumeFromPriorizedTopic(message: String) {
    eventProcessor.doLogic(message)
  }

  @KafkaConsumer(topic = "my_not_so_priorized_topic")
  fun consumeFromNotSoPrioritizedTopic(message: String) {
    eventProcessor.doLogic(message)  
  }
}

Answer 1

您可以为高优先级任务和低优先级任务创建两个Channel。然后使用协程的 select 来消费 channel 中的事件。表达式并将高优先级任务 channel 放在第一位。

例子(字符串是偶数):

fun process(value: String) {
    // do what you want with the event
}

suspend fun selectFromHighAndLow(highPriorityChannel: ReceiveChannel<String>, lowPriorityChannel: ReceiveChannel<String>): String =
        select<String> {
            highPriorityChannel.onReceive { value ->
                value
            }
            lowPriorityChannel.onReceive { value ->
                value
            }
        }


val highPriorityChannel = Channel<String>()
val lowPriorityChannel = Channel<String>()
while (true) {
    process(selectFromHighAndLow(highPriorityChannel, lowPriorityChannel))
}

要向这些 channel 发送内容，您可以使用 channel.send(event)。