Блог

Главная - Вопросы по программированию - Почему KafkaServer должен быть потокобезопасным

Почему KafkaServer должен быть потокобезопасным

#apache-kafka

#apache-kafka

Вопрос:

Я некоторое время был пользователем Kafka, и сегодня мне нужно прочитать его основной исходный код.

Ну, я обнаружил, что код о запуске брокера Kafka был здесь: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaServer.scala

В этом файле scala есть функция startup() , которая вызывается при запуске брокера Kafka.

Что меня смущает, так это то, что кажется, что некоторые атомарные переменные используются для обеспечения потокобезопасности этой функции. Как вы можете видеть, есть некоторые атомарные логические переменные, такие как isShuttingDown , startupComplete , isStartingUp .

 def startup(): Unit = {
    try {
      info("starting")

      if (isShuttingDown.get)
        throw new IllegalStateException("Kafka server is still shutting down, cannot re-start!")

      if (startupComplete.get)
        return

      val canStartup = isStartingUp.compareAndSet(false, true)
      if (canStartup) {
        brokerState.newState(Starting)

        /* setup zookeeper */
        initZkClient(time)
        ...
        ...

Но я вроде как не знаю, зачем нам нужен такой потокобезопасный механизм. Насколько я понимаю, один процесс Kafka broker имеет только один KafkaServer , что означает, что функция startup имеет только один шанс быть выполненной. Если мы попытаемся запустить более одного процесса Kafka broker за короткий промежуток времени на одном компьютере, это просто проблема между процессами, а не проблема с многопоточностью. Потокобезопасность, похоже, не требуется.

Может кто-нибудь сказать мне, почему здесь используются некоторые атомарные переменные? Почему здесь необходима потокобезопасность?

Комментарии:

1. что произойдет, если вы отправите два последовательных вызова завершения работы брокеру от двух разных клиентов (или от одного и того же), если isShuttingDown он не был атомарным?

2. baeldung.com/cs/race-conditions

3. Вы так не думаете, но именно так это работает. Это сервер. Он прослушивает команды. Как вы думаете, если мы оба отправим команду, она подождет, пока ваша команда не будет завершена, а затем запустит мою? Он не имеет синхронного поведения. Когда вы вызываете www.google.com , вы думаете, сервер просто ждет выполнения предыдущих команд, а затем показывает мне страницу?

4. Это нормально, потому что он реализует механизм для АТОМАРНОЙ проверки его состояния. Исходный код, который вы видите, эти команды, вызываются ИЗВНЕ брокера. Если вызываются две последовательные команды завершения работы без того, чтобы логическое значение было атомарным, обе команды могут считать его ложным, запуская два завершения работы. Повторяю, это сервер, прослушивающий команды с конечной точки.

5. но ВЫ вызываете это, как я мог бы также назвать это, выполнив kafka-server-start.sh в то же время, что и вы. Вы здесь не поняли. Он получает команды и выполняет методы. Если предыдущая вызванная команда не может быть отражена в последовательных, вы не можете гарантировать, что брокер не запустится дважды или не попытается завершить работу пять раз!