Предисловие

В предыдущей статьеПопрощайтесь с Zookeeper, используйте две команды для контейнеризации KafkaСледуйте официальной документацииdocker,На облачном сервере построен одноузловой кластер Kafka.,Успешно подключено к облачному серверу,Когда я пытаюсь подключиться с помощью Spark на своем ноутбуке,Невозможно использовать данные.

Как вы можете видеть на картинке выше, я нахожусь в потребительском сегменте. configЧжунминминминцзянbootstrap.serversНастройка как гибкая общедоступная сетьIP,А также можно подключиться к сети Kafka через Telnet на ноутбуке.

Но когда я запустил программу Spark, программа должна была вывести данные, потребляемые из какфы, но программа зависла.

Как видно из последней строки лога, соединение здесь на самом деле локальное, что определенно означает, что данные не могут быть использованы. Но я явно использовал в программе эластичный публичный IP, но почему здесь он стал локальным...

Анализ проблемы

Столкнувшись с проблемой такого рода, я сначала проверил проблему на сервере Kafka, поэтому пошел проверить конфигурацию Kafka в контейнере, чтобы увидеть, какой вариант конфигурации использует localhost, а затем исправил проблему.

проходитьdocker execКоманда вводитkafkaконтейнер,Затем начните поиск server.properties в разных каталогах.,Файл конфигурации был найден в каталоге /etc/kafk/docker.

Grep localhost обнаружил всего два параметра. Я зашел на официальный сайт, чтобы найти инструкции по настройке, и обнаружил, что вторая конфигурация была «виновником», вызвавшим эту проблему.

advertised.listeners

Сначала просмотрите описание опции Advertised.listeners.

Затем купите один и получите другой бесплатно, и эффект обучения для слушателей будет лучше.

listenersиadvertised.listenersОбщий смысл таков.,прослушиватели — это адреса, используемые сервером Kafka для прослушивания клиентских подключений.,Включает соединения между внешними клиентами и другими серверами Kafka.,Адрес, используемый для взаимодействия данных между клиентом и Kafka.

А Advertised.listeners — это адрес, транслируемый Kafka клиенту. Если сервер Kafka находится во внутренней сети, а клиент — во внешней сети, адрес в Advertised.listeners необходимо настроить как адрес, доступный извне.

Таким образом, когда клиент подключается, он сначала ищет адрес, транслируемый рекламодателями.listeners в Zookeeper или KRaft, а затем сопоставляет адрес, настроенный в прослушивателях для подключения к Kafka для чтения данных.

Обычный способ записи слушателей следующий:

PLAINTEXT://:9092
PLAINTEXT://0.0.0.0:9092
PLAINTEXT://ip:9092

Кажется, нет никакой разницы между первыми двумя способами записи. Оба они отслеживают все адреса, включая эластичные общедоступные IP-адреса. Третий способ записи — мониторинг фиксированных адресов. В конфигурации докер-контейнера по умолчанию стоит первый способ записи, поэтому менять его нет необходимости. Нам нужно только изменить локальный хост в dvertized.listeners на эластичный общедоступный IP-адрес.

решение

Я использую vi для редактирования server.properties, и при сохранении он предлагает readOnly. Убедитесь, что нет прав доступа к файлам. Chmod не удалось изменить разрешения. Поскольку вы не являетесь пользователем root, вы используете su и sudo для получения прав root. Этих двух команд нет.

Мы можем найти только другой путь,Разве вы не установили клиент Kafka на хост-компьютер раньше?,Мы можем изменить файл конфигурации в клиенте,Затем поместите его в каталог с файлом конфигурации.,Не лучше ли его заменить? Если нет клиента,Воляkafkaконтейнер Конфигурационные файлы вdocker cpВы также можете выйти и внести изменения.。

Друзья, знакомые с докером, знают, что у нас есть два способа замены файла конфигурации в контейнере.

1. Замена в образе: перепишите Dockerfile и завершите замену в процессе создания нового образа с использованием исходного образа.
2. Замена в контейнере: скопируйте измененный файл конфигурации непосредственно в контейнер Kafka.

Заменить конфигурацию образа

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

Создайте новый Dockerfile:

FROM apache/kafka:3.7.0
COPY server.properties /etc/kafka/docker

В Dockerfile всего две команды.,Скопируйте файл конфигурации в исходный образ Kafka.,Другие команды,НапримерCMDнаследоватьПросто исходное изображение。а затем выполнитьdocker buildЗаказ,на основеDockerFileСоздайте файл с именемapache/kakfa_kraft:3.7.0зеркало。

Напоминаем, что файл server.properties и файл Dockerfile должны находиться в одном каталоге, иначе будет сообщено об ошибке.

После создания образа мы используем его для запуска контейнера с именем kafka1.

Контейнер запускается нормально и может нормально подключаться.

Заменить конфигурацию контейнера

Второй метод относительно прост. Изначально мы запустили контейнер Kafka, используя официальный образ. Выполнение следующей команды может перезаписать исходный файл конфигурации.

docker cp server.properties kafka:/etc/kafka/docker/

После завершения замены перезапустите контейнер Kafka.

Проверка результата

Тестовый код искры выглядит следующим образом:

object Monitor {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("aqi").setMaster("local[1]")
    val ssc = new StreamingContext(conf, Seconds(10))
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "121.91.xx.xx:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "aqi",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("aqi_test")
    val stream: DStream[String] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    ).map(_.value)
    stream.foreachRDD(rdd => {
      rdd.foreach(x => {
        println(x)
      })
    })
    ssc.start
    ssc.awaitTermination
  }
}

Для одной и той же программы после изменения конфигурации с использованием двух вышеуказанных методов программа запуска может использовать данные.

Другие попытки

На первом этапе контейнера, после того как не было разрешения на изменение файла конфигурации, я попытался добавить некоторые параметры при создании контейнера Kafka.

Судя по логам, нет никаких сомнений в том, что все они потерпели неудачу, поэтому два моих метода выше проще.

Заключение

Это повторение и решение проблемы невозможности доступа к внешней сети при развертывании Kafka на моем облачном сервере. Вся статья не очень техническая и в основном посвящена использованию некоторых основных команд докера.

Я участвую в последнем конкурсе эссе для специального учебного лагеря Tencent Technology Creation 2024, приходите и разделите со мной приз!