Руководство по Vert.x Core

В основе Vert.x лежит набор Java API, которые мы называем Vert.x Core

Vert.x core предоставляет функциональность для:

Создание TCP клиентов и серверов
Создание HTTP клиентов и серверов, включая поддержку WebSocket
Мессенджер (Event bus)
Общие данные — локальные и кластерные распределённые карты
Периодические и отложенные действия
Развёртывание и свёртывание Verticles
UDP Sockets
Клиент DNS
Доступ к файловой системе
Виртуальные потоки
Высокая доступность
Нативные транспорты
Кластеризация

Функциональность в core довольно низкого уровня — вы не найдете здесь таких вещей, как доступ к базам данных, авторизацию или высокоуровневую функциональность веб-приложений. Всё это находится в Vert.x ext (расширениях).

Vert.x core небольшой и лёгкий. Вы используете только необходимые части. Он также полностью встраиваемый в ваши существующие приложения — мы не заставляем вас структурировать ваши приложения каким-то специальным образом, чтобы использовать Vert.x.

Вы можете использовать core из других языков, поддерживаемых Vert.x. Но есть интересная особенность — мы не заставляем вас использовать Java API напрямую, скажем, из JavaScript или Ruby. Ведь разные языки имеют разные соглашения и выражения, и было бы странно навязывать Java-стиль разработчикам Ruby (например). Вместо этого мы автоматически генерируем соответствующие эквиваленты Java API для каждого языка.

Отныне мы будем использовать слово core для обозначения Vert.x core.

Если вы используете Maven или Gradle, добавьте следующую зависимость в раздел dependencies вашего проекта для доступа к API Vert.x Core:

Maven (в вашем pom.xml):

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-core</artifactId>
  <version>5.0.0</version>
</dependency>

Gradle (в вашем файле build.gradle):

dependencies {
  compile 'io.vertx:vertx-core:5.0.0'
}

Давайте обсудим различные концепции и возможности в core.

Вначале был Vert.x

Вы не сможете сделать много в Vert.x, если не сможете взаимодействовать с объектом Vertx!

Это контрольный центр Vert.x, с помощью которого вы делаете практически всё, включая создание клиентов и серверов, получение ссылки на шину событий, установку таймеров и многое другое.

Так как получить экземпляр?

Если вы встраиваете Vert.x, то создаёте экземпляр следующим образом:

Vertx vertx = Vertx.vertx();

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

Указание параметров при создании объекта Vertx

При создании объекта Vert.x вы также можете указать параметры, если значения по умолчанию вам не подходят:

Vertx vertx = Vertx.vertx(new VertxOptions().setWorkerPoolSize(40));

Объект VertxOptions имеет множество настроек и позволяет настраивать такие вещи, как кластеризацию, высокую доступность, размеры пулов и различные другие настройки.

Создание кластеризованного объекта Vert.x

Если вы создаёте кластеризованный Vert.x (см. раздел о шине событий для получения дополнительной информации о кластеризации шины событий), то обычно используете асинхронный вариант для создания объекта Vertx.

Это связано с тем, что разным экземплярам Vert.x в кластере обычно требуется некоторое время (возможно, несколько секунд) для объединения. В это время мы не хотим блокировать вызывающую нить, поэтому мы передаём результат асинхронно.

Вы знаете флюэнтный подход?

Вы, возможно, заметили, что в предыдущих примерах использовался флюэнтный API.

Флюэнтный API — это когда несколько вызовов методов можно объединить в цепочку. Например:

request.response().putHeader("Content-Type", "text/plain").end("some text");

Это распространённый шаблон во всех API Vert.x, так что привыкайте к нему.

Объединение вызовов таким образом позволяет писать код немного короче. Конечно, если вам не нравится флюэнтный подход, мы не заставляем вас использовать его таким образом, вы можете спокойно его игнорировать, если предпочитаете, и написать свой код так:

HttpServerResponse response = request.response();
response.putHeader("Content-Type", "text/plain");
response.end("some text");

Не звоните нам, мы позвоним вам.

API Vert.x в основном событийно-ориентированы. Это означает, что когда происходят события, которые вас интересуют, Vert.x сообщит вам об этом, отправив событие.

Вот некоторые примеры событий:

сработал таймер
данные поступили по сокету,
данные были прочитаны с диска
произошла ошибка
HTTP-сервер получил запрос

Вы обрабатываете события, предоставив обработчики API Vert.x. Например, для получения события таймера каждую секунду вы должны сделать так:

vertx.setPeriodic(1000, id -> {
  // This handler will get called every second
  System.out.println("timer fired!");
});

Или для получения HTTP-запроса:

server.requestHandler(request -> {
  // This handler will be called every time an HTTP request is received at the server
  request.response().end("hello world!");
});

Через некоторое время, когда Vert.x имеет событие, которое нужно передать вашему обработчику, Vert.x вызовет его асинхронно.

Это приводит нас к некоторым важным понятиям в Vert.x:

Не блокируйте меня!

За очень немногими исключениями (например, некоторые операции с файловой системой, заканчивающиеся на «Sync»), ни один из API Vert.x не блокирует вызывающую нить.

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

Поскольку ни один из API Vert.x не блокирует потоки, вы можете использовать Vert.x для обработки большого количества параллельности, используя небольшое количество потоков.

С традиционным блокирующим API вызывающая нить может заблокироваться, когда:

Чтение данных из сокета
Запись данных на диск
Отправка сообщения получателю и ожидание ответа.
… Многие другие ситуации

Во всех вышеперечисленных случаях, когда ваш поток ожидает результата, он не может делать ничего другого — он фактически бесполезен.

Это означает, что если вы хотите много параллельности, используя блокирующие API, вам нужно много потоков, чтобы предотвратить остановку приложения.

Потоки имеют издержки с точки зрения требуемой памяти (например, для стека) и контекстного переключения.

Для уровней параллельности, необходимых во многих современных приложениях, блокирующий подход просто не масштабируется.

Реактор и многопотоковый реактор

Ранее мы упоминали, что API Vert.x ориентированы на события — Vert.x передает события обработчикам, когда они становятся доступны.

В большинстве случаев Vert.x вызывает ваши обработчики, используя поток, называемый циклом событий.

Поскольку ничего в Vert.x или вашем приложении не блокирует, цикл событий может свободно выполнять доставку событий различным обработчикам по мере их поступления.

Благодаря отсутствию блокировок, цикл событий потенциально может обрабатывать огромное количество событий за короткий промежуток времени. Например, один цикл событий может очень быстро обрабатывать множество HTTP-запросов.

Мы называем это паттерном Reactor.

Вы могли уже слышать об этом — например, Node.js реализует этот паттерн.

В стандартной реализации реактора существует один поток цикла событий, который циклически доставляет все события всем обработчикам по мере их поступления.

Проблема с одним потоком заключается в том, что он может работать только на одном ядре в любой момент времени. Поэтому, если вы хотите, чтобы ваше приложение реактора с одним потоком (например, ваше приложение Node.js) масштабировалось на многоядерном сервере, вам нужно запускать и управлять множеством разных процессов.

Vert.x работает иначе. Вместо одного цикла событий, каждый экземпляр Vertx поддерживает несколько циклов событий. По умолчанию мы выбираем количество, исходя из доступного количества ядер на машине, но это можно переопределить.

Это означает, что один процесс Vertx может масштабироваться по вашему серверу, в отличие от Node.js.

Мы называем этот паттерн многопотоковым реактором, чтобы отличать его от паттерна реактора с одним потоком.

Несмотря на то, что экземпляр Vertx поддерживает несколько циклов событий, любой конкретный обработчик никогда не будет выполняться одновременно, и в большинстве случаев (за исключением worker verticles) всегда будет вызываться с использованием того же самого цикла событий.

Золотое правило — не блокируйте цикл событий

Мы уже знаем, что API Vert.x являются неблокирующими и не будут блокировать цикл событий, но это мало помогает, если вы сами блокируете цикл событий в обработчике.

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

Поэтому не делайте этого! Вы предупреждены.

Примеры блокировок включают:

Thread.sleep()
Ожидание блокировки
Ожидание мьютекса или монитора (например, синхронизированный блок)
Выполнение длительной операции с базой данных и ожидание результата
Выполнение сложного вычисления, которое занимает значительное время.
Бесконечный цикл

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

Так… что такое значительный период времени?

Сколько длится кусок веревки? Это действительно зависит от вашего приложения и необходимого уровня конкурентности.

Если у вас есть один цикл событий и вы хотите обрабатывать 10000 запросов HTTP в секунду, то понятно, что обработка каждого запроса не может занимать более 0,1 мс, поэтому вы не можете блокировать его на большее время.

Математика не сложная и будет оставлена в качестве упражнения для читателя.

Если ваше приложение не реагирует, это может быть признаком блокировки цикла событий где-то. Чтобы помочь вам диагностировать такие проблемы, Vert.x будет автоматически регистрировать предупреждения, если обнаружит, что цикл событий не вернулся в течение некоторого времени. Если вы видите такие предупреждения в своих логах, вам следует провести расследование.

Thread vertx-eventloop-thread-3 has been blocked for 20458 ms

Vert.x также предоставит трассировки стека, чтобы точно определить, где происходит блокировка.

Если вы хотите отключить эти предупреждения или изменить настройки, вы можете сделать это в объекте VertxOptions перед созданием объекта Vertx.

Будущие результаты

Vert.x 4 использует фьючерсы для представления асинхронных результатов.

Любой асинхронный метод возвращает объект Future для результата вызова: успех или ошибка.

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

FileSystem fs = vertx.fileSystem();

Future<FileProps> future = fs.props("/my_file.txt");

future.onComplete((AsyncResult<FileProps> ar) -> {
  if (ar.succeeded()) {
    FileProps props = ar.result();
    System.out.println("File size = " + props.size());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

Не путайте фьючерсы с обещаниями.

Если фьючерсы представляют «сторону чтения» асинхронного результата, обещания — это «сторона записи». Они позволяют отложить действие по предоставлению результата.

В большинстве случаев вам не нужно создавать обещания в приложении Vert.x. Композиция фьючерсов и Координация фьючерсов предоставляют инструменты для преобразования и объединения асинхронных результатов.

Терминальные операции, такие как onSuccess, onFailure и onComplete, не гарантируют порядок вызова обратных вызовов.

Представьте фьючерс, на который зарегистрированы 2 обратных вызова:

future.onComplete(ar -> {
  // Do something
});
future.onComplete(ar -> {
  // May be invoked first
});

Возможен случай, когда второй обратный вызов будет вызван раньше первого.

Если вам нужна такая гарантия, рассмотрите композицию фьючерсов с andThen.

Композиция фьючерсов

compose можно использовать для цепочки фьючерсов:

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

FileSystem fs = vertx.fileSystem();

Future<Void> future = fs
  .createFile("/foo")
  .compose(v -> {
    // When the file is created (fut1), execute this:
    return fs.writeFile("/foo", Buffer.buffer());
  })
  .compose(v -> {
    // When the file is written (fut2), execute this:
    return fs.move("/foo", "/bar");
  });

В этом примере 3 операции объединены в цепочку:

создание файла
запись данных в файл
перемещение файла

При успешном выполнении всех 3 шагов, конечный фьючерс (future) также завершится успешно. В случае неудачи одного из шагов, конечный фьючерс завершится неудачей.

Помимо этого, Future предлагает больше: map, recover, otherwise, andThen и даже flatMap, которое является псевдонимом для compose

Будущее согласование

Согласование нескольких будущих событий может быть достигнуто с помощью Vert.x futures. Оно поддерживает одновременное выполнение (запуск нескольких асинхронных операций параллельно) и последовательное выполнение (цепочки асинхронных операций).

Future.all принимает несколько аргументов-будущих (до 6) и возвращает будущее, которое успешно выполняется, когда все будущие события успешно завершаются, и неудачно, когда хотя бы одно из будущих событий завершается неудачно:

Future<HttpServer> httpServerFuture = httpServer.listen();

Future<NetServer> netServerFuture = netServer.listen();

Future.all(httpServerFuture, netServerFuture).onComplete(ar -> {
  if (ar.succeeded()) {
    // All servers started
  } else {
    // At least one server failed
  }
});

Операции выполняются параллельно, Handler, прикреплённый к возвращённому будущему, вызывается по завершении композиции. Когда одна из операций завершается неудачно (одно из переданных будущих событий помечено как неудачное), возвращаемое будущее также отмечается как неудачное. Если все операции завершаются успешно, возвращаемое будущее завершается успешно.

В случае успеха метод resultAt гарантирует результаты в том же порядке, в котором они указаны в вызове Future.all. В приведённом примере, независимо от того, какой элемент завершился первым, результат httpServer может быть получен с помощью resultAt(0), а результат netServer - с помощью resultAt(1).

В качестве альтернативы вы можете передать список (возможно, пустой) будущих событий:

Future.all(Arrays.asList(future1, future2, future3));

В то время как композиция all ждёт, пока все будущие события будут успешными (или одно из них завершится неудачно), композиция any ждёт первого успешно завершённого будущего. Future.any принимает несколько аргументов-будущих (до 6) и возвращает будущее, которое успешно завершается, когда одно из будущих событий завершается успешно, и неудачно, когда все будущие события завершаются неудачно:

Future.any(future1, future2).onComplete(ar -> {
  if (ar.succeeded()) {
    // At least one is succeeded
  } else {
    // All failed
  }
});

Можно также использовать список будущих событий:

Future.any(Arrays.asList(f1, f2, f3));

Композиция join ждёт, пока все будущие события будут завершены, либо успешно, либо неудачно. Future.join принимает несколько аргументов-будущих (до 6) и возвращает будущее, которое успешно завершается, когда все будущие события завершаются успешно, и неудачно, когда все будущие события завершаются, и по крайней мере одно из них завершается неудачно:

Future.join(future1, future2, future3).onComplete(ar -> {
  if (ar.succeeded()) {
    // All succeeded
  } else {
    // All completed and at least one failed
  }
});

Также можно использовать список будущих событий:

Future.join(Arrays.asList(future1, future2, future3));

Взаимодействие с CompletionStage

API Vert.x Future предлагает совместимость с и из CompletionStage, что является интерфейсом JDK для составных асинхронных операций.

Мы можем перейти от Vert.x Future к CompletionStage, используя метод toCompletionStage, как в примере:

Future<String> future = vertx.createDnsClient().lookup("vertx.io");
future.toCompletionStage().whenComplete((ip, err) -> {
  if (err != null) {
    System.err.println("Could not resolve vertx.io");
    err.printStackTrace();
  } else {
    System.out.println("vertx.io => " + ip);
  }
});

Мы можем, наоборот, перейти от CompletionStage к Vert.x Future, используя Future.fromCompletionStage. Существуют 2 варианта:

первый вариант принимает только CompletionStage и вызывает методы Future из потока, который разрешает экземпляр CompletionStage, и
второй вариант принимает дополнительный параметр Context, чтобы вызвать методы Future в контексте Vert.x.

В большинстве случаев вариант с CompletionStage и Context является тем, который вы захотите использовать, чтобы соблюдать модель потоков Vert.x, так как Vert.x Future чаще всего используются с кодом, библиотеками и клиентами Vert.x.

Вот пример перехода от CompletionStage к Vert.x Future и диспетчеризации в контексте:

Future.fromCompletionStage(completionStage, vertx.getOrCreateContext())
  .flatMap(str -> {
    String key = UUID.randomUUID().toString();
    return storeInDb(key, str);
  })
  .onSuccess(str -> {
    System.out.println("We have a result: " + str);
  })
  .onFailure(err -> {
    System.err.println("We have a problem");
    err.printStackTrace();
  });

Вертикали

Vert.x поставляется с простой, масштабируемой моделью развертывания и одновременного выполнения, похожей на акторы, которую можно использовать для экономии времени.

Эта модель полностью необязательна, и Vert.x не обязывает вас создавать свои приложения таким образом, если вы этого не хотите.

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

Для использования этой модели необходимо написать код в виде набора вертикалей.

Вертикали — это фрагменты кода, которые развертываются и выполняются Vert.x. Экземпляр Vert.x по умолчанию поддерживает N потоков циклов событий (где N по умолчанию равно core*2). Вертикали могут быть написаны на любом из поддерживаемых Vert.x языках, и одно приложение может включать вертикали, написанные на нескольких языках.

Вертикаль можно рассматривать как нечто подобное актору в модели акторов.

Обычно приложение состоит из множества экземпляров вертикалей, выполняющихся одновременно в одном экземпляре Vert.x. Различные экземпляры вертикалей общаются друг с другом, отправляя сообщения в событийной шине.

Создание вертикалей

Классы вертикалей должны реализовывать интерфейс Deployable.

Они могут реализовывать его напрямую, но обычно проще расширить абстрактный класс VerticleBase.

Вот пример вертикали:

class MyVerticle extends VerticleBase {

  // Called when verticle is deployed
  public Future<?> start() throws Exception {
    return super.start();
  }

  // Optional - called when verticle is un-deployed
  public Future<?> stop() throws Exception {
    return super.stop();
  }
}

Обычно вы переопределяете метод start, как в приведённом выше примере.

Когда Vert.x развертывает вертикаль, он вызывает метод start, и когда возвращённое методом будущее завершится, вертикаль считается запущенной.

Вы также можете по желанию переопределить метод stop. Он будет вызван Vert.x при развёртывании вертикали, и когда возвращённое методом будущее завершится, вертикаль будет считаться остановленной.

Вот более подробный пример:

class MyVerticle extends VerticleBase {

  private HttpServer server;

  @Override
  public Future<?> start() {
    server = vertx.createHttpServer().requestHandler(req -> {
      req.response()
        .putHeader("content-type", "text/plain")
        .end("Hello from Vert.x!");
    });

    // Now bind the server:
    return server.listen(8080);
  }
}

Вы даже можете написать вертикаль в одну строку:

Deployable verticle = context -> vertx
  .createHttpServer()
  .requestHandler(req -> req.response()
    .putHeader("content-type", "text/plain")
    .end("Hello from Vert.x!"))
  .listen(8080);

Вам не нужно вручную останавливать HTTP-сервер, запущенный вертикалью, в методе stop вертикали. Vert.x автоматически остановит любой запущенный сервер при развёртывании вертикали.

Что случилось с контрактами Vert.x 4 Verticle и AbstractVerticle?

Контракт, определённый Verticle и AbstractVerticle, больше не был удобным с будущим Vert.x 5:

class MyVerticle extends AbstractVerticle {
  @Override
  public void start(Promise<Void> startPromise) throws Exception {
    Future<String> future = bindService();

    // Requires to write
    future.onComplete(ar -> {
      if (ar.succeeded()) {
        startPromise.complete();
      } else {
        startPromise.fail(ar.cause());
      }
    });

    // Or
    future
      .<Void>mapEmpty()
      .onComplete(startPromise);
  }
}

Тем не менее, Verticle и AbstractVerticle не устарели в Vert.x 5. Можно их использовать, но это больше не рекомендуемый вариант по умолчанию.

Типы вертикалей

Существуют два разных типа вертикалей:

Стандартные вертикали: Это наиболее распространённый и полезный тип — они всегда выполняются в потоке цикла событий. Мы обсудим это подробнее в следующей части.
Вертикали рабочих потоков: Они выполняются с помощью потока из пула рабочих потоков. Экземпляр никогда не выполняется одновременно более чем одним потоком.

Стандартные вертиксы

Стандартным вертиксам присваивается поток цикла событий при их создании, и метод start вызывается с этим циклом событий. Когда вы вызываете любые другие методы, которые принимают обработчик из ядра API из цикла событий, Vert.x гарантирует, что эти обработчики при вызове будут выполняться в том же цикле событий.

Это означает, что мы можем гарантировать, что весь код в экземпляре вашей вертиксы всегда выполняется в том же цикле событий (пока вы не создаете собственные потоки и не вызываете их!).

Это означает, что вы можете писать весь код своего приложения как однопоточный и позволить Vert.x позаботиться о потоках и масштабировании. Больше не нужно беспокоиться о synchronized и volatile, а также вы избегаете многих других проблем с гонками и тупиками, так распространенных при разработке многопоточных приложений «традиционным» способом.

Вертиксы для работы

Вертика для работы — это такая же, как стандартная вертикcа, но она выполняется с помощью потока из пула рабочих потоков Vert.x, а не с помощью цикла событий.

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

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

Если вы хотите развернуть вертиксу как рабочую, вы делаете это с помощью setThreadingModel.

DeploymentOptions options = new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER);
vertx.deployVerticle(new MyOrderProcessorVerticle(), options);

Экземпляры рабочих вертиксов никогда не выполняются одновременно Vert.x более чем одним потоком, но могут выполняться разными потоками в разное время.

Вертиксы виртуальных потоков

Вертика виртуального потока — это такая же, как стандартная вертикcа, но она выполняется с помощью виртуальных потоков, а не с помощью цикла событий.

Вертиксы виртуальных потоков предназначены для использования модели async/await с будущими значениями Vert.x.

Если вы хотите развернуть вертиксу как вертиксу виртуального потока, вы делаете это с помощью setThreadingModel.

DeploymentOptions options = new DeploymentOptions().setThreadingModel(ThreadingModel.VIRTUAL_THREAD);
vertx.deployVerticle(new MyOrderProcessorVerticle(), options);

эта функция требует Java 21

Развертывание вертиксов программно

Вы можете развернуть вертиксу, используя один из deployVerticle методов, указав имя вертиксы, или вы можете передать экземпляр вертиксы, которую вы уже создали сами.

VerticleBase myVerticle = new MyVerticle();
vertx.deployVerticle(myVerticle);

Вы также можете развертывать вертиксы, указав имя вертиксы.

Имя вертиксы используется для поиска конкретной VerticleFactory, которая будет использоваться для создания фактического экземпляра(ов) вертиксы.

Вот пример развертывания Java вертиксы, используя её имя класса:

vertx.deployVerticle("com.mycompany.MyOrderProcessorVerticle");

Правила сопоставления имени вертикаля с фабрикой вертикалей

При развертывании вертикаля(ей) с использованием имени, это имя используется для выбора фактической фабрики вертикалей, которая будет инстанцировать вертикaль(ей).

Имена вертикалей могут иметь префикс — строку, за которой следует двоеточие, которое, если присутствует, будет использовано для поиска фабрики, например:

 groovy:com.mycompany.SomeGroovyCompiledVerticle // Use the Groovy verticle factory

Если префикс отсутствует, Vert.x будет искать суффикс и использовать его для поиска фабрики, например:

 SomeScript.groovy // Will use the Groovy verticle factory

Если префикс или суффикс отсутствуют, Vert.x предположит, что это полное имя класса Java (FQCN), и попытается инстанцировать его.

Как располагаются фабрики вертикалей?

Большинство фабрик вертикалей загружаются из classpath и регистрируются при запуске Vert.x.

Вы также можете программно регистрировать и аннулировать фабрики вертикалей, используя registerVerticleFactory и unregisterVerticleFactory, если это необходимо.

Ожидание завершения развертывания

Развертывание вертикалей асинхронно и может завершиться некоторое время после возврата вызова deploy.

Если вы хотите получать уведомления о завершении развертывания, вы можете развернуть, указав обработчик завершения:

vertx
  .deployVerticle(new MyOrderProcessorVerticle())
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Deployment id is: " + res.result());
    } else {
      System.out.println("Deployment failed!");
    }
  });

Обработчик завершения получит результат, содержащий строку идентификатора развертывания, если развертывание прошло успешно.

Этот идентификатор развертывания можно использовать позже, если вы хотите развернуть развертывание.

Удаление развертывания вертикалей

Развертывания можно удалить с помощью undeploy.

Само удаление асинхронно, поэтому, если вы хотите получать уведомления о завершении удаления, вы можете указать обработчик завершения:

vertx
  .undeploy(deploymentID)
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Undeployed ok");
    } else {
      System.out.println("Undeploy failed!");
    }
  });

Указание количества экземпляров вертикалей

При развертывании вертикаля с помощью verticle, вы можете указать количество экземпляров вертикалей, которые вы хотите развернуть, вам также необходимо передать Callable<Deployable>, чтобы Vert.x мог инстанцировать ваши экземпляры вертикалей.

DeploymentOptions options = new DeploymentOptions().setInstances(16);
vertx.deployVerticle(() -> new MyOrderProcessorVerticle(), options);

Это полезно для легкого масштабирования на нескольких ядрах. Например, у вас может быть веб-серверная вертикaль для развертывания и несколько ядер на вашем компьютере, поэтому вы хотите развернуть несколько экземпляров, чтобы использовать все ядра.

Передача конфигурации вертикaли

Конфигурация в формате JSON может быть передана вертикaли во время развертывания:

JsonObject config = new JsonObject().put("name", "tim").put("directory", "/blah");
DeploymentOptions options = new DeploymentOptions().setConfig(config);
vertx.deployVerticle(new MyOrderProcessorVerticle(), options);

Эта конфигурация затем доступна через объект Context или напрямую с помощью метода config. Конфигурация возвращается как объект JSON, поэтому вы можете извлечь данные следующим образом:

System.out.println("Configuration: " + config().getString("name"));

Получение переменных окружения в Verticle

Переменные окружения и системные свойства доступны с помощью Java API:

System.getProperty("prop");
System.getenv("HOME");

Выход Vert.x

Потоки, поддерживаемые экземплярами Vert.x, не являются демонами, поэтому они препятствуют завершению JVM.

Если вы встраиваете Vert.x и закончили с ним, вы можете вызвать close, чтобы его закрыть.

Это приведет к завершению всех внутренних пулов потоков и закрытию других ресурсов, что позволит JVM завершиться.

Объект контекста

Когда Vert.x предоставляет событие обработчику или вызывает методы start или stop Verticle, выполнение связывается с Context. Обычно контекст — это контекст цикла событий и привязан к определённому потоку цикла событий. Таким образом, выполнение для этого контекста всегда происходит в том же потоке цикла событий. В случае с рабочими Verticle и выполнением инлайновых блокирующих кодов, с выполнением будет связан рабочий контекст, использующий поток из пула рабочих потоков.

Чтобы получить контекст, используйте метод getOrCreateContext:

Context context = vertx.getOrCreateContext();

Если текущий поток имеет связанный с ним контекст, он повторно использует объект контекста. Если нет, создаётся новый экземпляр контекста. Вы можете проверить тип полученного контекста:

Context context = vertx.getOrCreateContext();
if (context.isEventLoopContext()) {
  System.out.println("Context attached to Event Loop");
} else if (context.isWorkerContext()) {
  System.out.println("Context attached to Worker Thread");
} else if (! Context.isOnVertxThread()) {
  System.out.println("Context not attached to a thread managed by vert.x");
}

Когда вы получили объект контекста, вы можете выполнить код в этом контексте асинхронно. Другими словами, вы отправляете задачу, которая в конечном итоге будет выполнена в том же контексте, но позже:

vertx.getOrCreateContext().runOnContext( (v) -> {
  System.out.println("This will be executed asynchronously in the same context");
});

Когда несколько обработчиков работают в одном контексте, они могут захотеть обмениваться данными. Объект контекста предоставляет методы для хранения и извлечения данных, используемых в контексте. Например, он позволяет передавать данные в некоторое действие, выполняемое с runOnContext:

final Context context = vertx.getOrCreateContext();
context.put("data", "hello");
context.runOnContext((v) -> {
  String hello = context.get("data");
});

Объект контекста также позволяет получить доступ к конфигурации verticle, используя метод config. Подробности о данной конфигурации см. в разделе Передача конфигурации в verticle.

Выполнение периодических и отложенных действий

В Vert.x очень часто требуется выполнить действие после задержки или периодически.

В стандартных вертиклах невозможно просто заставить поток заснуть, чтобы ввести задержку, так как это заблокирует поток цикла событий.

Вместо этого используются таймеры Vert.x. Таймеры могут быть одноразовыми или периодическими. Мы рассмотрим оба варианта.

Одноразовые таймеры

Одноразовый таймер вызывает обработчик события через определённую задержку, выраженную в миллисекундах.

Чтобы установить таймер, который сработает один раз, используйте метод setTimer, передав задержку и обработчик.

long timerID = vertx.setTimer(1000, id -> {
  System.out.println("And one second later this is printed");
});

System.out.println("First this is printed");

Возвращаемое значение — уникальный идентификатор таймера, который впоследствии можно использовать для отмены таймера. Обработчик также получает идентификатор таймера.

Периодические таймеры

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

Будет первоначальная задержка, равная периоду.

Возвращаемое значение setPeriodic — уникальный идентификатор таймера (long). Он может быть использован позднее, если необходимо отменить таймер.

Аргументом, передаваемым в обработчик события таймера, также является уникальный идентификатор таймера:

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

В этом случае следует рассмотреть использование setTimer вместо него. После завершения обработки можно установить следующий таймер.

long timerID = vertx.setPeriodic(1000, id -> {
  System.out.println("And every second this is printed");
});

System.out.println("First this is printed");

Отмена таймеров

Для отмены периодического таймера вызовите cancelTimer, указав идентификатор таймера. Например:

vertx.cancelTimer(timerID);

Таймер как будущее

Timer объединяет одноразовый таймер и будущие в одном API.

Future<String> timer = vertx
  .timer(10, TimeUnit.SECONDS)
  .map(v -> "Success");

timer.onSuccess(value -> {
  System.out.println("Timer fired: " + value);
});
timer.onFailure(cause -> {
  System.out.println("Timer cancelled: " + cause.getMessage());
});

Будущее завершается успешно, когда таймер срабатывает, а, наоборот, cancelling — если таймер завершается неудачно.

Автоматическое удаление в вертиклах

Если вы создаёте таймеры внутри вертикла, эти таймеры будут автоматически закрыты при развёртывании вертикла.

Пул рабочих вертикла

Вертиклы используют пул рабочих Vert.x для выполнения блокирующих действий, т.е. executeBlocking или рабочих вертиклах.

Разный пул рабочих может быть указан в параметрах развертывания:

vertx.deployVerticle(new MyOrderProcessorVerticle(), new DeploymentOptions().setWorkerPoolName("the-specific-pool"));

Шина событий

event bus является нервной системой Vert.x.

Для каждого экземпляра Vert.x существует единственный экземпляр шины событий, который можно получить с помощью метода eventBus.

Шина событий позволяет различным частям вашего приложения взаимодействовать друг с другом, независимо от языка программирования, на котором они написаны, и независимо от того, находятся ли они в одном экземпляре Vert.x или в разных.

Она даже может быть подключена для взаимодействия с клиентской стороной JavaScript, работающей в браузере.

Шина событий образует распределённую систему обмена сообщениями «равный-равному», охватывающую несколько серверных узлов и несколько браузеров.

Шина событий поддерживает модели обмена сообщениями «опубликовать/подписаться», «точка-точка» и «запрос-ответ».

API шины событий очень прост. В основном он включает регистрацию обработчиков, отмену регистрации обработчиков и отправку и публикацию сообщений.

Сначала немного теории:

Теория

Обращения

Сообщения отправляются в шину событий по адресу.

Vert.x не утруждает себя сложными схемами адресации. В Vert.x адрес — это просто строка. Любая строка допустима. Однако рекомендуется использовать какую-либо схему, например, используя точки для разграничения пространства имён.

Примеры допустимых адресов: europe.news.feed1, acme.games.pacman, sausages и X.

Обработчики

Сообщения принимаются обработчиками. Вы регистрируете обработчик по адресу.

По одному и тому же адресу можно зарегистрировать несколько обработчиков.

Один обработчик можно зарегистрировать по многим адресам.

Публикация/подписка на сообщения

Шина событий поддерживает публикацию сообщений.

Сообщения публикуются по адресу. Публикация означает доставку сообщения всем обработчикам, зарегистрированным по этому адресу.

Это знакомый шаблон публикация/подписка.

Точечная передача и сообщения запроса/ответа

Шина событий также поддерживает точечную передачу сообщений.

Сообщения отправляются по адресу. Vert.x затем маршрутизирует их только одному из обработчиков, зарегистрированных по этому адресу.

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

При точечной передаче сообщений при отправке сообщения можно указать необязательный обработчик ответа.

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

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

Это распространённый шаблон обмена сообщениями, называемый шаблоном запрос/ответ.

Доставка с наилучшими усилиями

Vert.x делает всё возможное, чтобы доставить сообщения и сознательно не будет их выбрасывать. Это называется доставкой с наилучшими усилиями.

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

Если ваше приложение заботится о потерянных сообщениях, вы должны написать обработчики, которые являются идемпотентными, и ваши отправители должны повторять попытки после восстановления.

Типы сообщений

Vert.x из коробки позволяет отправлять любые примитивные/простые типы, строки или buffers в качестве сообщений.

Однако, в Vert.x принято и общепринятой практикой является отправка сообщений в формате JSON.

JSON очень легко создавать, читать и парсить на всех языках, поддерживаемых Vert.x, поэтому он стал своего рода lingua franca для Vert.x.

Однако вы не обязаны использовать JSON, если не хотите.

Шина событий очень гибкая и также поддерживает отправку произвольных объектов по шине событий. Для этого можно определить codec для объектов, которые вы хотите отправить.

API шины событий

Перейдём к API.

Получение шины событий

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

EventBus eb = vertx.eventBus();

В одном экземпляре Vert.x существует единственный экземпляр шины событий.

Регистрация обработчиков

Самый простой способ зарегистрировать обработчик — использовать consumer. Вот пример:

EventBus eb = vertx.eventBus();

eb.consumer("news.uk.sport", message -> {
  System.out.println("I have received a message: " + message.body());
});

При поступлении сообщения для вашего обработчика, ваш обработчик будет вызван, передавая message.

Объект, возвращаемый вызовом consumer(), является экземпляром MessageConsumer.

Этот объект впоследствии может использоваться для отмены регистрации обработчика или использования обработчика как потока.

В качестве альтернативы, вы можете использовать consumer, чтобы вернуть MessageConsumer без установленного обработчика, а затем установить обработчик на нём. Например:

EventBus eb = vertx.eventBus();

MessageConsumer<String> consumer = eb.consumer("news.uk.sport");
consumer.handler(message -> {
  System.out.println("I have received a message: " + message.body());
});

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

Если вы хотите получить уведомление об окончании этого процесса, вы можете использовать completion future на объекте MessageConsumer.

consumer.completion().onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("The handler registration has reached all nodes");
  } else {
    System.out.println("Registration failed!");
  }
});

Отмена регистрации обработчиков

Чтобы отменить регистрацию обработчика, вызовите unregister.

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

consumer
  .unregister()
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("The handler un-registration has reached all nodes");
    } else {
      System.out.println("Un-registration failed!");
    }
  });

Опубликование сообщений

Опубликование сообщения простое. Просто используйте publish, указав адрес, в который его опубликовать.

eventBus.publish("news.uk.sport", "Yay! Someone kicked a ball");

Это сообщение будет затем доставлено всем обработчикам, зарегистрированным по адресу news.uk.sport.

Отправка сообщений

Отправка сообщения приведет к тому, что только один обработчик, зарегистрированный по указанному адресу, получит сообщение. Это паттерн обмена сообщениями "точка-точка". Обработчик выбирается нестрогим алгоритмом круговой очереди.

Вы можете отправить сообщение с помощью send.

eventBus.send("news.uk.sport", "Yay! Someone kicked a ball");

Установка заголовков сообщений

Сообщения, отправленные по шине событий, также могут содержать заголовки. Это может быть задано путём предоставления DeliveryOptions при отправке или публикации:

DeliveryOptions options = new DeliveryOptions();
options.addHeader("some-header", "some-value");
eventBus.send("news.uk.sport", "Yay! Someone kicked a ball", options);

Порядок сообщений

Vert.x будет доставлять сообщения любому конкретному обработчику в том же порядке, в котором они были отправлены конкретным отправителем.

Объект сообщения

Объект, который вы получаете в обработчике сообщений, является Message.

body сообщения соответствует объекту, который был отправлен или опубликован.

Заголовки сообщения доступны с помощью headers.

Подтверждение сообщений/отправка ответов

При использовании send шина событий пытается доставить сообщение зарегистрированному на шине событий MessageConsumer.

В некоторых случаях для отправителя полезно знать, когда потребитель получил сообщение и "обработало" его с помощью паттерна запрос-ответ.

Чтобы подтвердить, что сообщение было обработано, потребитель может ответить на сообщение, вызвав reply.

В этом случае ответ отправляется обратно отправителю, и обработчик ответа вызывается с ответом.

Пример прояснит это:

Получатель:

MessageConsumer<String> consumer = eventBus.consumer("news.uk.sport");
consumer.handler(message -> {
  System.out.println("I have received a message: " + message.body());
  message.reply("how interesting!");
});

Отправитель:

eventBus
  .request("news.uk.sport", "Yay! Someone kicked a ball across a patch of grass")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Received reply: " + ar.result().body());
    }
  });

Ответ может содержать тело сообщения, которое может содержать полезную информацию.

Что подразумевается под "обработкой" определяется приложением и зависит исключительно от действий потребителя сообщений, и это не то, что шина событий Vert.x сама знает или о чем заботится.

Примеры:

Простой потребитель сообщений, реализующий службу, возвращающую текущее время, подтвердит сообщение ответом, содержащим время суток в теле ответа
Потребитель сообщений, реализующий постоянную очередь, может подтвердить сообщение true, если сообщение было успешно сохранено в хранилище, или false, если нет.
Потребитель сообщений, обрабатывающий заказ, может подтвердить сообщение true, когда заказ был успешно обработан, чтобы его можно было удалить из базы данных

Отправка с таймаутами

При отправке сообщения с обработчиком ответа вы можете указать таймаут в DeliveryOptions.

Если ответ не получен в течение этого времени, обработчик ответа будет вызван с ошибкой.

Значение таймаута по умолчанию составляет 30 секунд.

Ошибки отправки

Отправка сообщений может завершиться ошибкой по другим причинам, в том числе:

Нет доступных обработчиков для отправки сообщения
Получатель явно отклонил сообщение с помощью fail

Во всех случаях обработчик ответа будет вызван с конкретной ошибкой.

Кодеки сообщений

Вы можете отправлять любые объекты через шину событий, если вы определите и зарегистрируете message codec для них.

У кодеков сообщений есть имя, и вы указываете это имя в DeliveryOptions при отправке или публикации сообщения:

eventBus.registerCodec(myCodec);

DeliveryOptions options = new DeliveryOptions().setCodecName(myCodec.name());

eventBus.send("orders", new MyPOJO(), options);

Если вы всегда хотите использовать один и тот же кодек для определённого типа, то вы можете зарегистрировать для него кодек по умолчанию, а затем вам не нужно указывать кодек в параметрах доставки при каждом отправлении:

eventBus.registerDefaultCodec(MyPOJO.class, myCodec);

eventBus.send("orders", new MyPOJO());

Вы отменяете регистрацию кодека сообщений с помощью unregisterCodec.

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

Vert.x имеет встроенные кодеки для определённых типов данных:

базовые типы (строка, массив байтов, байт, целое, длинное целое, двойная точность, логическое значение, короткое целое, символ), или
некоторые типы данных Vert.x (буферы, массив JSON, объекты JSON), или
типы, реализующие интерфейс ClusterSerializable, или
типы, реализующие интерфейс java.io.Serializable.

В режиме кластеризации ClusterSerializable и java.io.Serializable объекты по умолчанию отклоняются по соображениям безопасности.

Вы можете определить разрешённые классы для кодирования и декодирования, предоставив функции, которые проверяют имя класса:

Кластеризованная шина событий

Шина событий существует не только в одном экземпляре Vert.x. Объединяя различные экземпляры Vert.x в вашей сети, они могут образовывать единую распределённую шину событий.

Если вы создаёте экземпляр Vert.x программно, вы получаете кластеризованную шину событий, настроив экземпляр Vert.x как кластеризованный:

VertxOptions options = new VertxOptions();
Vertx
  .clusteredVertx(options)
  .onComplete(res -> {
    if (res.succeeded()) {
      Vertx vertx = res.result();
      EventBus eventBus = vertx.eventBus();
      System.out.println("We now have a clustered event bus: " + eventBus);
    } else {
      System.out.println("Failed: " + res.cause());
    }
  });

Также убедитесь, что на вашем пути к выполнению есть одна из реализаций ClusterManager.

Автоматическая очистка в вертиклах

Если вы регистрируете обработчики шины событий изнутри вертиклов, эти обработчики будут автоматически дерегистрированы при развёртывании вертикла.

Настройка шины событий

Настройка шины событий возможна. Она особенно полезна при использовании кластеризованной шины событий. Под капотом шина событий использует TCP-соединения для отправки и получения сообщений, поэтому параметры EventBusOptions позволяют настроить все аспекты этих TCP-соединений. Поскольку шина событий выступает как сервер, так и клиент, конфигурация близка к конфигурациям NetClientOptions и NetServerOptions.

VertxOptions options = new VertxOptions()
    .setEventBusOptions(new EventBusOptions()
        .setSsl(true)
        .setKeyCertOptions(new JksOptions().setPath("keystore.jks").setPassword("wibble"))
        .setTrustOptions(new JksOptions().setPath("keystore.jks").setPassword("wibble"))
        .setClientAuth(ClientAuth.REQUIRED)
    );

Vertx
  .clusteredVertx(options)
  .onComplete(res -> {
    if (res.succeeded()) {
      Vertx vertx = res.result();
      EventBus eventBus = vertx.eventBus();
      System.out.println("We now have a clustered event bus: " + eventBus);
    } else {
      System.out.println("Failed: " + res.cause());
    }
  });

Предыдущий фрагмент кода демонстрирует, как можно использовать SSL-соединения для шины событий вместо обычных TCP-соединений.

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

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

Параметры EventBusOptions также позволяют указать, является ли шина событий кластеризованной, а также порт и хост.

При использовании в контейнерах можно также настроить публичный хост и порт:

VertxOptions options = new VertxOptions()
    .setEventBusOptions(new EventBusOptions()
        .setClusterPublicHost("whatever")
        .setClusterPublicPort(1234)
    );

Vertx
  .clusteredVertx(options)
  .onComplete(res -> {
    if (res.succeeded()) {
      Vertx vertx = res.result();
      EventBus eventBus = vertx.eventBus();
      System.out.println("We now have a clustered event bus: " + eventBus);
    } else {
      System.out.println("Failed: " + res.cause());
    }
  });

JSON

В отличие от некоторых других языков, Java не имеет встроенной поддержки JSON, поэтому мы предоставляем два класса, чтобы упростить работу с JSON в ваших приложениях Vert.x.

Объекты JSON

Класс JsonObject представляет объекты JSON.

Объект JSON — это по сути просто карта, у которой есть строковые ключи, а значения могут быть одного из поддерживаемых JSON типов (строка, число, логическое значение).

Объекты JSON также поддерживают значения null.

Создание объектов JSON

Пустые объекты JSON можно создать с помощью конструктора по умолчанию.

Вы можете создать объект JSON из строкового представления JSON следующим образом:

String jsonString = "{\"foo\":\"bar\"}";
JsonObject object = new JsonObject(jsonString);

Вы можете создать объект JSON из карты следующим образом:

Map<String, Object> map = new HashMap<>();
map.put("foo", "bar");
map.put("xyz", 3);
JsonObject object = new JsonObject(map);

Добавление элементов в объект JSON

Используйте методы put, чтобы добавить значения в объект JSON.

Вызовы методов можно объединять благодаря флюентному API:

JsonObject object = new JsonObject();
object.put("foo", "bar").put("num", 123).put("mybool", true);

Получение значений из объекта JSON

Вы получаете значения из объекта JSON, используя методы getXXX, например:

String val = jsonObject.getString("some-key");
int intVal = jsonObject.getInteger("some-other-key");

Преобразование между объектами JSON и Java-объектами

Вы можете создать объект JSON из полей Java-объекта следующим образом:

Вы можете создать Java-объект и заполнить его поля из объекта JSON следующим образом:

request.bodyHandler(buff -> {
  JsonObject jsonObject = buff.toJsonObject();
  User javaObject = jsonObject.mapTo(User.class);
});

Обратите внимание, что оба вышеуказанных направления преобразования используют Jackson’s ObjectMapper#convertValue() для выполнения преобразования. См. документацию Jackson для получения информации о влиянии видимости полей и конструкторов, предостережениях по сериализации и десериализации через ссылки на объекты и т. д.

Однако, в простейшем случае, как mapFrom, так и mapTo должны быть успешны, если все поля Java-класса являются общедоступными (или имеют общедоступные геттеры/сеттеры), и если существует общедоступный конструктор по умолчанию (или нет определенных конструкторов).

Ссылки на объекты будут транзитивно сериализованы/десериализованы в/из вложенных объектов JSON, если граф объектов не является циклическим.

Кодирование объекта JSON в строку

Для кодирования объекта в строку используйте encode.

JSON массивы

Класс JsonArray представляет JSON массивы.

JSON массив — это последовательность значений (строка, число, булево значение).

JSON массивы также могут содержать нулевые значения.

Создание JSON массивов

Пустые JSON массивы можно создать с помощью конструктора по умолчанию.

Вы можете создать JSON массив из строкового представления JSON следующим образом:

String jsonString = "[\"foo\",\"bar\"]";
JsonArray array = new JsonArray(jsonString);

Добавление элементов в JSON массив

Вы добавляете элементы в JSON массив, используя методы add.

JsonArray array = new JsonArray();
array.add("foo").add(123).add(false);

Получение значений из JSON массива

Вы получаете значения из JSON массива, используя методы getXXX, например:

String val = array.getString(0);
Integer intVal = array.getInteger(1);
Boolean boolVal = array.getBoolean(2);

Кодирование JSON массива в строку

Для кодирования массива в строковую форму используйте encode.

Создание произвольных JSON данных

Создание JSON объекта и массива предполагает, что вы используете допустимое строковое представление.

Если вы не уверены в валидности строки, используйте вместо этого Json.decodeValue.

Object object = Json.decodeValue(arbitraryJson);
if (object instanceof JsonObject) {
  // That's a valid json object
} else if (object instanceof JsonArray) {
  // That's a valid json array
} else if (object instanceof String) {
  // That's valid string
} else {
  // etc...
}

Настройка Jackson

Настройка ограничений чтения

Начиная с Jackson 2.15, добавлены верхние границы ограничений для ограничения накопленных байтов при разборе JSON входных данных.

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

vertx.jackson.defaultReadMaxNestingDepth: Максимальная глубина вложенности
vertx.jackson.defaultReadMaxDocumentLength: Максимальная длина документа
vertx.jackson.defaultReadMaxNumberLength: Максимальная длина числового значения
vertx.jackson.defaultReadMaxStringLength: Максимальная длина строкового значения
vertx.jackson.defaultReadMaxNameLength: Максимальная длина имени свойства
vertx.jackson.defaultMaxTokenCount: Максимальное количество токенов

Для получения дополнительной информации обратитесь к данному ресурсу.

Указатели на JSON

Vert.x предоставляет реализацию указателей на JSON из RFC6901. Вы можете использовать указатели для запросов и записи. Вы можете построить свои указатели JsonPointer, используя строку, URI или вручную добавляя пути:

JsonPointer pointer1 = JsonPointer.from("/hello/world");
// Build a pointer manually
JsonPointer pointer2 = JsonPointer.create()
  .append("hello")
  .append("world");

После создания указателя, используйте queryJson для запроса значения JSON. Вы можете обновить значение JSON с помощью writeJson:

Object result1 = objectPointer.queryJson(jsonObject);
// Query a JsonArray
Object result2 = arrayPointer.queryJson(jsonArray);
// Write starting from a JsonObject
objectPointer.writeJson(jsonObject, "new element");
// Write starting from a JsonObject
arrayPointer.writeJson(jsonArray, "new element");

Вы можете использовать указатель Vert.x Json с любой моделью объекта, предоставив пользовательскую реализацию JsonPointerIterator

Буферы

Большая часть данных перемещается внутри Vert.x с помощью буферов.

Буфер — это последовательность нуля или более байтов, к которой можно читать или записывать, и которая автоматически расширяется по мере необходимости для размещения любых записанных в неё байтов. Можно представить буфер как умный массив байтов.

Создание буферов

Буферы можно создать, используя один из статических методов.

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

Вот несколько примеров создания буферов:

Создать новый пустой буфер:

Buffer buff = Buffer.buffer();

Создать буфер из строки. Строка будет закодирована в буфере с помощью UTF-8.

Buffer buff = Buffer.buffer("some string");

Создать буфер из строки: Строка будет закодирована с помощью указанного кодирования, например:

Buffer buff = Buffer.buffer("some string", "UTF-16");

Создать буфер из byte[]

byte[] bytes = new byte[] {1, 3, 5};
Buffer buff = Buffer.buffer(bytes);

Создать буфер с начальным размером. Если вам известно, что в буфер будет записано определённое количество данных, вы можете создать буфер и указать этот размер. Это заставит буфер изначально выделить столько памяти, что более эффективно, чем многократное автоматическое изменение размера буфера по мере записи данных в него.

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

Buffer buff = Buffer.buffer(10000);

Запись в буфер

Существует два способа записи в буфер: добавление и произвольный доступ. В любом случае буферы всегда будут автоматически расширяться, чтобы вместить байты. Невозможно получить IndexOutOfBoundsException с помощью буфера.

Добавление в буфер

Для добавления в буфер используются методы.

Методы добавления существуют для добавления различных типов.

Возвращаемое значение методов — сам буфер, поэтому их можно объединять:

Buffer buff = Buffer.buffer();

buff.appendInt(123).appendString("hello\n");

socket.write(buff);

Запись в буфер с произвольным доступом

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

Буфер всегда будет расширяться по мере необходимости, чтобы вместить данные.

Buffer buff = Buffer.buffer();

buff.setInt(1000, 123);
buff.setString(0, "hello");

Чтение из буфера

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

Buffer buff = Buffer.buffer();
for (int i = 0; i < buff.length(); i += 4) {
  System.out.println("int value at " + i + " is " + buff.getInt(i));
}

Работа с беззнаковыми числами

Беззнаковые числа могут быть считываться или добавлены/установлены в буфер с помощью методов getUnsignedXXX, appendUnsignedXXX и setUnsignedXXX. Это полезно при реализации кодека для сетевого протокола, оптимизированного для минимизации использования полосы пропускания.

В следующем примере значение 200 устанавливается в указанной позиции с помощью одного байта:

Buffer buff = Buffer.buffer(128);
int pos = 15;
buff.setUnsignedByte(pos, (short) 200);
System.out.println(buff.getUnsignedByte(pos));

На консоли отображается '200'.

Длина буфера

Используйте метод length для получения длины буфера. Длина буфера - это индекс байта в буфере с наибольшим индексом + 1.

Копирование буферов

Используйте метод copy для создания копии буфера.

Нарезание буферов

Нарезанный буфер - это новый буфер, который опирается на исходный буфер, то есть он не копирует данные. Используйте метод slice для создания нарезанных буферов.

Повторное использование буфера

После записи буфера в сокет или другое подобное место, они больше не могут быть повторно использованы.

Создание серверов и клиентов TCP

Vert.x позволяет легко создавать неблокирующие TCP-клиенты и серверы.

Создание TCP-сервера

Самый простой способ создать TCP-сервер, используя все параметры по умолчанию, выглядит следующим образом:

NetServer server = vertx.createNetServer();

Настройка TCP-сервера

Если вам не нужны параметры по умолчанию, сервер можно настроить, передав объект настроек при его создании:

NetServerOptions options = new NetServerOptions().setPort(4321);
NetServer server = vertx.createNetServer(options);

Запуск сервера и прослушивание

Чтобы заставить сервер прослушивать входящие запросы, используйте один из вариантов listen.

Чтобы заставить сервер прослушивать указанный хост и порт:

NetServer server = vertx.createNetServer();
server.listen();

Или указать хост и порт в вызове listen, игнорируя конфигурацию в параметрах:

NetServer server = vertx.createNetServer();
server.listen(1234, "localhost");

По умолчанию хост — 0.0.0.0 (прослушивание на всех доступных адресах), а порт — 0 (специальное значение, которое заставляет сервер выбрать случайный свободный порт).

Фактическая привязка асинхронная, поэтому сервер может начать прослушивание не сразу после вызова listen.

Если вам нужно уведомить о фактическом начале прослушивания, можно передать обработчик в вызов listen. Например:

NetServer server = vertx.createNetServer();
server
  .listen(1234, "localhost")
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Server is now listening!");
    } else {
      System.out.println("Failed to bind!");
    }
  });

Прослушивание на случайном порту

Если используется 0 в качестве порта прослушивания, сервер выберет случайный свободный порт.

Чтобы узнать фактический порт прослушивания, можно вызвать actualPort.

NetServer server = vertx.createNetServer();
server
  .listen(0, "localhost")
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Server is now listening on actual port: " + server.actualPort());
    } else {
      System.out.println("Failed to bind!");
    }
  });

Прослушивание Unix-доменных сокетов

При работе с JDK 16+ или с использованием нативного транспорта, сервер может прослушивать Unix-доменные сокеты:

NetServer netServer = vertx.createNetServer();

// Only available when running on JDK16+, or using a native transport
SocketAddress address = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");

netServer
  .connectHandler(so -> {
  // Handle application
  })
  .listen(address)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // Bound to socket
    } else {
      // Handle failure
    }
  });

Получение уведомлений о входящих подключениях

Для получения уведомлений о подключении, необходимо установить обработчик connectHandler:

NetServer server = vertx.createNetServer();
server.connectHandler(socket -> {
  // Handle the connection in here
});

При подключении обработчик будет вызван с экземпляром NetSocket.

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

Чтение данных из сокета

Для чтения данных из сокета, необходимо установить обработчик handler на сокете.

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

NetServer server = vertx.createNetServer();
server.connectHandler(socket -> {
  socket.handler(buffer -> {
    System.out.println("I received some bytes: " + buffer.length());
  });
});

Запись данных в сокет

Для записи в сокет используйте один из write.

Buffer buffer = Buffer.buffer().appendFloat(12.34f).appendInt(123);
socket.write(buffer);

// Write a string in UTF-8 encoding
socket.write("some data");

// Write a string using the specified encoding
socket.write("some data", "UTF-16");

Операции записи асинхронные и могут не произойти до определенного момента после возврата вызова write.

Обработчик закрытия

Если вы хотите получать уведомления о закрытии сокета, можете установить closeHandler:

socket.closeHandler(v -> {
  System.out.println("The socket has been closed");
});

Обработка исключений

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

Вы можете установить exceptionHandler для получения любых исключений, которые произошли до того, как соединение передано connectHandler, например, во время рукопожатия TLS.

Обработчик записи Event Bus

Каждый сокет может зарегистрировать обработчик в Event Bus, который при получении буферов записывает их в себя. Это локальные подписки, недоступные для других узлов кластера.

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

Эта функция отключена по умолчанию, но её можно включить, используя setRegisterWriteHandler или setRegisterWriteHandler.

Адрес обработчика предоставляется writeHandlerID.

Локальные и удаленные адреса

Локальный адрес NetSocket можно получить, используя localAddress.

Удаленный адрес (адрес другого конца соединения) NetSocket можно получить, используя remoteAddress.

Отправка файлов или ресурсов из classpath

Файлы и ресурсы classpath могут быть записаны в сокет напрямую, используя sendFile. Это может быть очень эффективным способом отправки файлов, так как оно может обрабатываться ядром ОС напрямую, где это поддерживается операционной системой.

См. главу о обслуживании файлов из classpath для ограничений разрешения classpath или его отключения.

socket.sendFile("myfile.dat");

Потоковые сокеты

Экземпляры NetSocket также являются ReadStream и WriteStream экземплярами, поэтому их можно использовать для передачи данных в или из других потоковых чтений и записей.

См. главу о потоках для получения дополнительной информации.

Обновление соединений до SSL/TLS

Соединение без SSL/TLS может быть обновлено до SSL/TLS, используя upgradeToSsl.

Для корректной работы сервер или клиент должны быть настроены для SSL/TLS. См. главу по SSL/TLS для получения дополнительной информации.

Плавный завершение TCP-соединения

Вы можете выполнить плавное завершение server или client.

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

server
  .shutdown()
  .onSuccess(res -> {
    System.out.println("Server is now closed");
  });

Завершение ожидает, пока все сокеты не будут закрыты или не сработает таймаут завершения. Когда таймаут сработает, все сокеты будут закрыты принудительно.

Каждый открытый сокет уведомляется о событии завершения, позволяя выполнить закрытие на уровне протокола перед фактическим закрытием сокета.

socket.shutdownHandler(v -> {
  socket
    // Write close frame
    .write(closeFrame())
    // Wait until we receive the remote close frame
    .compose(success -> closeFrameHandler(socket))
    // Close the socket
    .eventually(() -> socket.close());
});

Значение по умолчанию для таймаута завершения составляет 30 секунд, но его можно переопределить.

server
  .shutdown(60, TimeUnit.SECONDS)
  .onSuccess(res -> {
    System.out.println("Server is now closed");
  });

Закрытие TCP-соединения

Вы можете закрыть server или client, чтобы немедленно закрыть все открытые подключения и освободить все ресурсы. В отличие от shutdown, здесь нет периода плавного завершения.

Закрытие является асинхронным и может не завершиться до некоторого времени после возвращения вызова. Вы можете использовать возвращенное будущее, чтобы получить уведомление о завершении фактического закрытия.

Это будущее завершается, когда закрытие полностью завершено.

server
  .close()
  .onSuccess(res -> {
    System.out.println("Server is now closed");
  });

Автоматическая очистка в вертексах

Если вы создаёте TCP-серверы и клиенты внутри вертексов, эти серверы и клиенты будут автоматически закрыты при развёртывании вертекса.

Масштабирование — совместное использование TCP-серверов

Обработчики любого TCP-сервера всегда выполняются в одном потоке цикла событий.

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

Для использования большего числа ядер сервера вам потребуется развернуть несколько экземпляров сервера.

Вы можете создать несколько экземпляров программно в своём коде:

class MyVerticle extends VerticleBase {

  NetServer server;

  @Override
  public Future<?> start() {
    server = vertx.createNetServer();
    server.connectHandler(socket -> {
      socket.handler(buffer -> {
        // Just echo back the data
        socket.write(buffer);
      });
    });
    return server.listen(1234, "localhost");
  }
}

// Create a few instances so we can utilise cores
vertx.deployVerticle(MyVerticle.class, new DeploymentOptions().setInstances(10));

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

В этот момент вы можете спросить себя: «Как можно иметь более одного сервера, прослушивающего один и тот же хост и порт? Разве вы не получите конфликты портов, как только попытаетесь развернуть более одного экземпляра?»

Vert.x использует небольшой трюк здесь.*

Когда вы развертываете другой сервер на том же хосте и порту, что и существующий сервер, он не пытается создать новый сервер, прослушивающий тот же хост/порт.

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

Следовательно, TCP-серверы Vert.x могут масштабироваться по доступным ядрам, при этом каждый экземпляр остаётся однопоточным.

Создание TCP-клиента

Самый простой способ создать TCP-клиента, используя все параметры по умолчанию, выглядит следующим образом:

NetClient client = vertx.createNetClient();

Настройка TCP-клиента

Если вам не требуется значение по умолчанию, клиент можно настроить, передав экземпляр NetClientOptions при его создании:

NetClientOptions options = new NetClientOptions().setConnectTimeout(10000);
NetClient client = vertx.createNetClient(options);

Установление соединений

Для подключения к серверу используется connect, указывая порт и хост сервера, а также обработчик, который будет вызван с результатом, содержащим NetSocket при успешном подключении, или с ошибкой в случае неудачного подключения.

NetClientOptions options = new NetClientOptions().setConnectTimeout(10000);
NetClient client = vertx.createNetClient(options);
client
  .connect(4321, "localhost")
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Connected!");
      NetSocket socket = res.result();
    } else {
      System.out.println("Failed to connect: " + res.cause().getMessage());
    }
  });

Установление соединений с сокетами Unix-домена

При работе на JDK 16+ или использовании нативного транспорта, клиент может подключаться к сокетам Unix-домена:

NetClient netClient = vertx.createNetClient();

// Only available when running on JDK16+, or using a native transport
SocketAddress addr = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");

// Connect to the server
netClient
  .connect(addr)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // Connected
    } else {
      // Handle failure
    }
  });

Настройка попыток подключения

Клиент можно настроить на автоматическое повторное подключение к серверу в случае невозможности подключения. Это настраивается с помощью setReconnectInterval и setReconnectAttempts.

В настоящее время Vert.x не будет пытаться повторно подключиться при сбое соединения, попытки повторного подключения и интервал применяются только к созданию начальных соединений.

NetClientOptions options = new NetClientOptions().
  setReconnectAttempts(10).
  setReconnectInterval(500);

NetClient client = vertx.createNetClient(options);

По умолчанию, многократные попытки подключения отключены.

Ведение журнала сетевой активности

Для отладки можно вести журнал сетевой активности:

NetServerOptions options = new NetServerOptions().setLogActivity(true);

NetServer server = vertx.createNetServer(options);

Вот вывод простого HTTP-сервера

id: 0x359e3df6, L:/127.0.0.1:8080 - R:/127.0.0.1:65351] READ: 78B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 47 45 54 20 2f 20 48 54 54 50 2f 31 2e 31 0d 0a |GET / HTTP/1.1..|
|00000010| 48 6f 73 74 3a 20 6c 6f 63 61 6c 68 6f 73 74 3a |Host: localhost:|
|00000020| 38 30 38 30 0d 0a 55 73 65 72 2d 41 67 65 6e 74 |8080..User-Agent|
|00000030| 3a 20 63 75 72 6c 2f 37 2e 36 34 2e 31 0d 0a 41 |: curl/7.64.1..A|
|00000040| 63 63 65 70 74 3a 20 2a 2f 2a 0d 0a 0d 0a       |ccept: */*....  |
+--------+-------------------------------------------------+----------------+
[id: 0x359e3df6, L:/127.0.0.1:8080 - R:/127.0.0.1:65351] WRITE: 50B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d |HTTP/1.1 200 OK.|
|00000010| 0a 63 6f 6e 74 65 6e 74 2d 6c 65 6e 67 74 68 3a |.content-length:|
|00000020| 20 31 31 0d 0a 0d 0a 48 65 6c 6c 6f 20 57 6f 72 | 11....Hello Wor|
|00000030| 6c 64                                           |ld              |
+--------+-------------------------------------------------+----------------+
[id: 0x359e3df6, L:/127.0.0.1:8080 - R:/127.0.0.1:65351] READ COMPLETE
[id: 0x359e3df6, L:/127.0.0.1:8080 - R:/127.0.0.1:65351] FLUSH

По умолчанию двоичные данные записываются в шестнадцатеричном формате.

Можно уменьшить подробность формата данных, чтобы выводилась только длина буфера вместо всех данных, установив формат лога данных.

NetServerOptions options = new NetServerOptions()
  .setLogActivity(true)
  .setActivityLogDataFormat(ByteBufFormat.SIMPLE);

NetServer server = vertx.createNetServer(options);

Вот тот же вывод с простым форматом буфера

[id: 0xda8d41dc, L:/127.0.0.1:8080 - R:/127.0.0.1:65399] READ: 78B
[id: 0xda8d41dc, L:/127.0.0.1:8080 - R:/127.0.0.1:65399] WRITE: 50B
[id: 0xda8d41dc, L:/127.0.0.1:8080 - R:/127.0.0.1:65399] READ COMPLETE
[id: 0xda8d41dc, L:/127.0.0.1:8080 - R:/127.0.0.1:65399] FLUSH
[id: 0xda8d41dc, L:/127.0.0.1:8080 - R:/127.0.0.1:65399] READ COMPLETE
[id: 0xda8d41dc, L:/127.0.0.1:8080 ! R:/127.0.0.1:65399] INACTIVE
[id: 0xda8d41dc, L:/127.0.0.1:8080 ! R:/127.0.0.1:65399] UNREGISTERED

Клиенты также могут регистрировать сетевую активность

NetClientOptions options = new NetClientOptions().setLogActivity(true);

NetClient client = vertx.createNetClient(options);

Сетевая активность регистрируется Netty на уровне DEBUG и с именем io.netty.handler.logging.LoggingHandler. При использовании ведения журнала сетевой активности следует учитывать несколько моментов:

Ведение журнала выполняется не Vert.x, а Netty
это не функция для рабочей среды

Вы должны прочитать раздел Ведение журнала Netty.

Ограничение пропускной способности входящего и исходящего трафика TCP-соединений

TCP-сервер (Net/Http) может быть настроен с параметрами управления трафиком для ограничения пропускной способности. Ограничение пропускной способности как для входящего, так и для исходящего трафика может быть установлено через TrafficShapingOptions. Для NetServer параметры управления трафиком можно установить через NetServerOptions, а для HttpServer — через HttpServerOptions.

NetServerOptions options = new NetServerOptions()
  .setHost("localhost")
  .setPort(1234)
  .setTrafficShapingOptions(new TrafficShapingOptions()
    .setInboundGlobalBandwidth(64 * 1024)
    .setOutboundGlobalBandwidth(128 * 1024));

NetServer server = vertx.createNetServer(options);

HttpServerOptions options = new HttpServerOptions()
  .setHost("localhost")
  .setPort(1234)
  .setTrafficShapingOptions(new TrafficShapingOptions()
    .setInboundGlobalBandwidth(64 * 1024)
    .setOutboundGlobalBandwidth(128 * 1024));

HttpServer server = vertx.createHttpServer(options);

Эти параметры управления трафиком также можно динамически обновлять после запуска сервера.

NetServerOptions options = new NetServerOptions()
                             .setHost("localhost")
                             .setPort(1234)
                             .setTrafficShapingOptions(new TrafficShapingOptions()
                                                         .setInboundGlobalBandwidth(64 * 1024)
                                                         .setOutboundGlobalBandwidth(128 * 1024));
NetServer server = vertx.createNetServer(options);
TrafficShapingOptions update = new TrafficShapingOptions()
                                 .setInboundGlobalBandwidth(2 * 64 * 1024) // twice
                                 .setOutboundGlobalBandwidth(128 * 1024); // unchanged
server
  .listen(1234, "localhost")
  // wait until traffic shaping handler is created for updates
  .onSuccess(v -> server.updateTrafficShapingOptions(update));

HttpServerOptions options = new HttpServerOptions()
                              .setHost("localhost")
                              .setPort(1234)
                              .setTrafficShapingOptions(new TrafficShapingOptions()
                                                          .setInboundGlobalBandwidth(64 * 1024)
                                                          .setOutboundGlobalBandwidth(128 * 1024));
HttpServer server = vertx.createHttpServer(options);
TrafficShapingOptions update = new TrafficShapingOptions()
                                 .setInboundGlobalBandwidth(2 * 64 * 1024) // twice
                                 .setOutboundGlobalBandwidth(128 * 1024); // unchanged
server
  .listen(1234, "localhost")
  // wait until traffic shaping handler is created for updates
  .onSuccess(v -> server.updateTrafficShapingOptions(update));

Настройка серверов и клиентов для работы с SSL/TLS

Клиенты и серверы TCP могут быть настроены для использования безопасного протокола передачи данных — более ранние версии TLS были известны как SSL.

API серверов и клиентов одинаковы, независимо от использования SSL/TLS, и оно включается путём конфигурации экземпляров NetClientOptions или NetServerOptions, используемых для создания серверов или клиентов.

Включение SSL/TLS на сервере

SSL/TLS включается с помощью ssl.

По умолчанию оно отключено.

Указание ключа/сертификата для сервера

SSL/TLS-серверы обычно предоставляют сертификаты клиентам для проверки их личности.

Сертификаты/ключи можно настроить для серверов несколькими способами:

Первый метод заключается в указании расположения хранилища ключей Java, содержащего сертификат и закрытый ключ.

Хранилища ключей Java можно управлять с помощью утилиты keytool, которая поставляется с JDK.

Пароль для хранилища ключей также должен быть предоставлен:

NetServerOptions options = new NetServerOptions().setSsl(true).setKeyCertOptions(
  new JksOptions().
    setPath("/path/to/your/server-keystore.jks").
    setPassword("password-of-your-keystore")
);
NetServer server = vertx.createNetServer(options);

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

Buffer myKeyStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/server-keystore.jks");
JksOptions jksOptions = new JksOptions().
  setValue(myKeyStoreAsABuffer).
  setPassword("password-of-your-keystore");
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(jksOptions);
NetServer server = vertx.createNetServer(options);

Ключ/сертификат в формате PKCS#12 (https://en.wikipedia.org/wiki/PKCS_12), обычно с расширением .pfx или .p12, также можно загрузить аналогичным образом, как хранилища ключей JKS:

NetServerOptions options = new NetServerOptions().setSsl(true).setKeyCertOptions(
  new PfxOptions().
    setPath("/path/to/your/server-keystore.pfx").
    setPassword("password-of-your-keystore")
);
NetServer server = vertx.createNetServer(options);

Также поддерживается конфигурация буфера:

Buffer myKeyStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/server-keystore.pfx");
PfxOptions pfxOptions = new PfxOptions().
  setValue(myKeyStoreAsABuffer).
  setPassword("password-of-your-keystore");
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(pfxOptions);
NetServer server = vertx.createNetServer(options);

Еще один способ предоставить закрытый ключ и сертификат сервера отдельно, используя файлы .pem.

NetServerOptions options = new NetServerOptions().setSsl(true).setKeyCertOptions(
  new PemKeyCertOptions().
    setKeyPath("/path/to/your/server-key.pem").
    setCertPath("/path/to/your/server-cert.pem")
);
NetServer server = vertx.createNetServer(options);

Также поддерживается конфигурация буфера:

Buffer myKeyAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/server-key.pem");
Buffer myCertAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/server-cert.pem");
PemKeyCertOptions pemOptions = new PemKeyCertOptions().
  setKeyValue(myKeyAsABuffer).
  setCertValue(myCertAsABuffer);
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(pemOptions);
NetServer server = vertx.createNetServer(options);

Vert.x поддерживает чтение нешифрованных RSA и/или ECC-ключей из файлов PKCS8 PEM. RSA-ключи также могут быть прочитаны из файлов PKCS1 PEM. Сертификаты X.509 могут быть прочитаны из файлов PEM, содержащих текстовое кодирование сертификата, как определено в RFC 7468, Раздел 5.

Имейте в виду, что ключи, содержащиеся в нешифрованных файлах PKCS8 или PKCS1 PEM, могут быть извлечены любым, кто может прочитать файл. Поэтому убедитесь, что на таких файлах PEM установлены надлежащие ограничения доступа, чтобы предотвратить злоупотребление.

Наконец, вы также можете загрузить общее хранилище ключей Java, что полезно для использования других реализаций хранилищ ключей, таких как Bouncy Castle:

NetServerOptions options = new NetServerOptions().setSsl(true).setKeyCertOptions(
  new KeyStoreOptions().
    setType("BKS").
    setPath("/path/to/your/server-keystore.bks").
    setPassword("password-of-your-keystore")
);
NetServer server = vertx.createNetServer(options);

Указание доверия для сервера

Серверы SSL/TLS могут использовать центр сертификации для проверки подлинности клиентов.

Центры сертификации могут быть настроены для серверов несколькими способами:

Хранилища доверия Java можно управлять с помощью утилиты keytool, которая поставляется с JDK.

Пароль для хранилища доверия также должен быть предоставлен:

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new JksOptions().
      setPath("/path/to/your/truststore.jks").
      setPassword("password-of-your-truststore")
  );
NetServer server = vertx.createNetServer(options);

В качестве альтернативы вы можете прочитать хранилище доверия как буфер и предоставить его напрямую:

Buffer myTrustStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/truststore.jks");
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new JksOptions().
      setValue(myTrustStoreAsABuffer).
      setPassword("password-of-your-truststore")
  );
NetServer server = vertx.createNetServer(options);

Центр сертификации в формате PKCS#12 (https://en.wikipedia.org/wiki/PKCS_12), обычно с расширением .pfx или .p12, также может быть загружен аналогичным образом, как хранилища доверия JKS:

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new PfxOptions().
      setPath("/path/to/your/truststore.pfx").
      setPassword("password-of-your-truststore")
  );
NetServer server = vertx.createNetServer(options);

Также поддерживается настройка буфера:

Buffer myTrustStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/truststore.pfx");
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new PfxOptions().
      setValue(myTrustStoreAsABuffer).
      setPassword("password-of-your-truststore")
  );
NetServer server = vertx.createNetServer(options);

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

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new PemTrustOptions().
      addCertPath("/path/to/your/server-ca.pem")
  );
NetServer server = vertx.createNetServer(options);

Также поддерживается настройка буфера:

Buffer myCaAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/server-ca.pfx");
NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setClientAuth(ClientAuth.REQUIRED).
  setTrustOptions(
    new PemTrustOptions().
      addCertValue(myCaAsABuffer)
  );
NetServer server = vertx.createNetServer(options);

Включение SSL/TLS на клиенте

Клиенты Net также могут быть легко настроены для использования SSL. У них есть точно такие же API при использовании SSL, как и при использовании стандартных сокетов.

Чтобы включить SSL на NetClient, вызывается функция setSSL(true).

Настройка доверия клиента

Если значение trustALl установлено в true на клиенте, то клиент будет доверять всем серверным сертификатам. Соединение всё ещё будет зашифровано, но этот режим уязвим к атакам «человек посередине». То есть, вы не можете быть уверены, с кем подключаетесь. Используйте с осторожностью. Значение по умолчанию — false.

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustAll(true);
NetClient client = vertx.createNetClient(options);

Если trustAll не задано, то необходимо настроить хранилище доверия клиента, которое должно содержать сертификаты серверов, которым доверяет клиент.

По умолчанию проверка хоста не настроена на клиенте. Это проверяет часть CN сертификата сервера по отношению к имени хоста сервера, чтобы избежать атак «человек посередине».

Вы должны настроить её явно на вашем клиенте

"" (пустая строка) отключает проверку хоста
"HTTPS" включает проверку HTTP по TLS
LDAPS включает расширение LDAP v3 для проверки TLS

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setHostnameVerificationAlgorithm(verificationAlgorithm);
NetClient client = vertx.createNetClient(options);

клиент Vert.x HTTP использует TCP-клиент и настраивает алгоритм проверки с помощью "HTTPS".

Как и настройка сервера, доверие клиента можно настроить несколькими способами:

Первый способ — указать расположение Java-хранилища доверия, которое содержит сертификат удостоверяющего центра.

Это стандартное Java-хранилище ключей, такое же, как хранилища ключей на стороне сервера. Расположение хранилища доверия клиента задаётся с помощью функции path на jks options. Если сервер предоставляет сертификат во время подключения, которого нет в хранилище доверия клиента, попытка подключения не будет успешной.

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new JksOptions().
      setPath("/path/to/your/truststore.jks").
      setPassword("password-of-your-truststore")
  );
NetClient client = vertx.createNetClient(options);

Также поддерживается настройка буфера:

Buffer myTrustStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/truststore.jks");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new JksOptions().
      setValue(myTrustStoreAsABuffer).
      setPassword("password-of-your-truststore")
  );
NetClient client = vertx.createNetClient(options);

Удостоверяющий центр в формате PKCS#12 (http://en.wikipedia.org/wiki/PKCS_12), обычно с расширением .pfx или .p12, также можно загрузить аналогичным образом, как и JKS-хранилища доверия:

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new PfxOptions().
      setPath("/path/to/your/truststore.pfx").
      setPassword("password-of-your-truststore")
  );
NetClient client = vertx.createNetClient(options);

Также поддерживается настройка буфера:

Buffer myTrustStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/truststore.pfx");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new PfxOptions().
      setValue(myTrustStoreAsABuffer).
      setPassword("password-of-your-truststore")
  );
NetClient client = vertx.createNetClient(options);

Ещё один способ предоставления удостоверяющего центра сервера с помощью списка .pem файлов.

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new PemTrustOptions().
      addCertPath("/path/to/your/ca-cert.pem")
  );
NetClient client = vertx.createNetClient(options);

Также поддерживается настройка буфера:

Buffer myTrustStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/ca-cert.pem");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(
    new PemTrustOptions().
      addCertValue(myTrustStoreAsABuffer)
  );
NetClient client = vertx.createNetClient(options);

Указание ключа/сертификата для клиента

Если сервер требует аутентификации клиента, то клиент должен представить свой собственный сертификат серверу при подключении. Клиента можно настроить несколькими способами:

Первый метод заключается в указании расположения хранилища ключей Java, содержащего ключ и сертификат. Опять же, это обычное хранилище ключей Java. Расположение хранилища ключей клиента устанавливается с помощью функции path в jks options.

NetClientOptions options = new NetClientOptions().setSsl(true).setKeyCertOptions(
  new JksOptions().
    setPath("/path/to/your/client-keystore.jks").
    setPassword("password-of-your-keystore")
);
NetClient client = vertx.createNetClient(options);

Также поддерживается конфигурация буфера:

Buffer myKeyStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/client-keystore.jks");
JksOptions jksOptions = new JksOptions().
  setValue(myKeyStoreAsABuffer).
  setPassword("password-of-your-keystore");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setKeyCertOptions(jksOptions);
NetClient client = vertx.createNetClient(options);

NetClientOptions options = new NetClientOptions().setSsl(true).setKeyCertOptions(
  new PfxOptions().
    setPath("/path/to/your/client-keystore.pfx").
    setPassword("password-of-your-keystore")
);
NetClient client = vertx.createNetClient(options);

Также поддерживается конфигурация буфера:

Buffer myKeyStoreAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/client-keystore.pfx");
PfxOptions pfxOptions = new PfxOptions().
  setValue(myKeyStoreAsABuffer).
  setPassword("password-of-your-keystore");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setKeyCertOptions(pfxOptions);
NetClient client = vertx.createNetClient(options);

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

NetClientOptions options = new NetClientOptions().setSsl(true).setKeyCertOptions(
  new PemKeyCertOptions().
    setKeyPath("/path/to/your/client-key.pem").
    setCertPath("/path/to/your/client-cert.pem")
);
NetClient client = vertx.createNetClient(options);

Также поддерживается конфигурация буфера:

Buffer myKeyAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/client-key.pem");
Buffer myCertAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/client-cert.pem");
PemKeyCertOptions pemOptions = new PemKeyCertOptions().
  setKeyValue(myKeyAsABuffer).
  setCertValue(myCertAsABuffer);
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setKeyCertOptions(pemOptions);
NetClient client = vertx.createNetClient(options);

Обратите внимание, что при конфигурации pem, закрытый ключ не зашифрован.

Обновление конфигурации SSL/TLS

Вы можете использовать метод updateSSLOptions для обновления ключей/сертификатов или доверия на сервере или клиенте TCP (например, для реализации вращения сертификатов).

Future<Boolean> fut = server.updateSSLOptions(new ServerSSLOptions()
  .setKeyCertOptions(
    new JksOptions()
      .setPath("/path/to/your/server-keystore.jks").
      setPassword("password-of-your-keystore")));

При успешном обновлении используется новая конфигурация SSL, в противном случае сохраняется предыдущая конфигурация.

Объект options сравнивается (с использованием equals) с существующими параметрами, чтобы предотвратить обновление, когда объекты равны, так как загрузка параметров может быть дорогостоящей. Если объекты равны, вы можете использовать параметр force, чтобы принудительно обновить.

Самозаверяемые сертификаты для целей тестирования и разработки

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

Часто требуются самозаверяемые сертификаты, будь то для юнит-тестирования/интеграционных тестов или для запуска тестовой версии приложения.

SelfSignedCertificate может использоваться для предоставления помощников самозаверяемых PEM-сертификатов и предоставления конфигураций KeyCertOptions и TrustOptions:

SelfSignedCertificate certificate = SelfSignedCertificate.create();

NetServerOptions serverOptions = new NetServerOptions()
  .setSsl(true)
  .setKeyCertOptions(certificate.keyCertOptions())
  .setTrustOptions(certificate.trustOptions());

vertx.createNetServer(serverOptions)
  .connectHandler(socket -> socket.end(Buffer.buffer("Hello!")))
  .listen(1234, "localhost");

NetClientOptions clientOptions = new NetClientOptions()
  .setSsl(true)
  .setKeyCertOptions(certificate.keyCertOptions())
  .setTrustOptions(certificate.trustOptions());

NetClient client = vertx.createNetClient(clientOptions);
client
  .connect(1234, "localhost")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      ar.result().handler(buffer -> System.out.println(buffer));
    } else {
      System.err.println("Woops: " + ar.cause().getMessage());
    }
  });

Клиент также может быть настроен на доверие ко всем сертификатам:

NetClientOptions clientOptions = new NetClientOptions()
  .setSsl(true)
  .setTrustAll(true);

Обратите внимание, что самозаверяемые сертификаты также работают для других TCP-протоколов, таких как HTTPS:

SelfSignedCertificate certificate = SelfSignedCertificate.create();

vertx.createHttpServer(new HttpServerOptions()
  .setSsl(true)
  .setKeyCertOptions(certificate.keyCertOptions())
  .setTrustOptions(certificate.trustOptions()))
  .requestHandler(req -> req.response().end("Hello!"))
  .listen(8080);

Отмена сертификатов уполномоченных органов

Доверие может быть настроено на использование списка отозванных сертификатов (CRL) для отозванных сертификатов, которым больше не следует доверять. Конфигурация crlPath настраивает использование списка crl:

NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(trustOptions).
  addCrlPath("/path/to/your/crl.pem");
NetClient client = vertx.createNetClient(options);

Также поддерживается конфигурация буфера:

Buffer myCrlAsABuffer = vertx.fileSystem().readFileBlocking("/path/to/your/crl.pem");
NetClientOptions options = new NetClientOptions().
  setSsl(true).
  setTrustOptions(trustOptions).
  addCrlValue(myCrlAsABuffer);
NetClient client = vertx.createNetClient(options);

Настройка набора шифров

По умолчанию конфигурация TLS будет использовать список наборов шифров SSL-движка:

Движок JDK SSL, когда используется JdkSSLEngineOptions
Движок OpenSSL, когда используется OpenSSLEngineOptions

Этот набор шифров можно настроить с помощью набора включенных шифров:

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions).
  addEnabledCipherSuite("ECDHE-RSA-AES128-GCM-SHA256").
  addEnabledCipherSuite("ECDHE-ECDSA-AES128-GCM-SHA256").
  addEnabledCipherSuite("ECDHE-RSA-AES256-GCM-SHA384").
  addEnabledCipherSuite("CDHE-ECDSA-AES256-GCM-SHA384");
NetServer server = vertx.createNetServer(options);

Когда набор включенных шифров определен (т.е. не пуст), он имеет приоритет над набором шифров по умолчанию SSL-движка.

Набор шифров можно указать в конфигурации NetServerOptions или NetClientOptions.

Настройка версий протокола TLS

По умолчанию в конфигурации TLS включены следующие протоколы: TLSv1.2 и TLSv1.3. Версии протоколов можно включить, явно добавив их:

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions).
  addEnabledSecureTransportProtocol("TLSv1.1");
NetServer server = vertx.createNetServer(options);

Их также можно удалить:

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions).
  removeEnabledSecureTransportProtocol("TLSv1.2");
NetServer server = vertx.createNetServer(options);

Версии протоколов можно указать в конфигурации NetServerOptions или NetClientOptions.

TLS 1.0 (TLSv1) и TLS 1.1 (TLSv1.1) широко устарели и были отключены по умолчанию с Vert.x 4.4.0.

SSL-движок

Реализация движка может быть настроена на использование OpenSSL вместо реализации JDK. До того, как JDK начал использовать аппаратные средства (инструкции процессора) для AES в Java 8 и для RSA в Java 9, OpenSSL обеспечивал значительно лучшие показатели производительности и использование процессора по сравнению с движком JDK.

Варианты движка для использования:

варианты getSslEngineOptions, если они установлены
в противном случае JdkSSLEngineOptions

NetServerOptions options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions);

// Use JDK SSL engine explicitly
options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions).
  setSslEngineOptions(new JdkSSLEngineOptions());

// Use OpenSSL engine
options = new NetServerOptions().
  setSsl(true).
  setKeyCertOptions(keyStoreOptions).
  setSslEngineOptions(new OpenSSLEngineOptions());

Указание имени сервера (SNI)

Указание имени сервера (SNI) — это расширение TLS, с помощью которого клиент указывает имя хоста, к которому пытается подключиться. Во время рукопожатия TLS клиент предоставляет имя сервера, и сервер может использовать его для ответа с помощью конкретного сертификата для этого имени сервера вместо стандартного развернутого сертификата. Если сервер требует аутентификации клиента, сервер может использовать конкретный доверенный сертификат CA в зависимости от указанного имени сервера.

Когда SNI активен, сервер использует

CN или SAN DNS (Subject Alternative Name с DNS) сертификата для точного соответствия, например www.example.com
CN или SAN DNS сертификата для соответствия шаблону имени, например *.example.com
в противном случае первый сертификат, когда клиент не предоставляет имя сервера или представленное имя сервера не может быть сопоставлено

Когда сервер дополнительно требует аутентификации клиента:

если JksOptions равно set on trust options, то выполняется точное соответствие с псевдонимом хранилища доверия
в противном случае используются доступные сертификаты CA таким же образом, как если бы SNI не было

сервер не может связаться с ошибкой, указывающей на неверный псевдоним

Вы можете включить SNI на сервере, установив setSni на true и настроив сервер с несколькими парами ключ/сертификат.

Файлы Java KeyStore или PKCS12 могут хранить несколько пар ключ/сертификат по умолчанию.

JksOptions keyCertOptions = new JksOptions().setPath("keystore.jks").setPassword("wibble");

NetServer netServer = vertx.createNetServer(new NetServerOptions()
    .setKeyCertOptions(keyCertOptions)
    .setSsl(true)
    .setSni(true)
);

PemKeyCertOptions можно настроить на хранение нескольких записей:

PemKeyCertOptions keyCertOptions = new PemKeyCertOptions()
    .setKeyPaths(Arrays.asList("default-key.pem", "host1-key.pem", "etc..."))
    .setCertPaths(Arrays.asList("default-cert.pem", "host2-key.pem", "etc...")
    );

NetServer netServer = vertx.createNetServer(new NetServerOptions()
    .setKeyCertOptions(keyCertOptions)
    .setSsl(true)
    .setSni(true)
);

Клиент неявно отправляет имя подключаемого хоста в качестве имени сервера SNI для полностью квалифицированного доменного имени (FQDN).

Вы можете указать явное имя сервера при подключении сокета

NetClient client = vertx.createNetClient(new NetClientOptions()
    .setTrustOptions(trustOptions)
    .setSsl(true)
);

// Connect to 'localhost' and present 'server.name' server name
client
  .connect(1234, "localhost", "server.name")
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Connected!");
      NetSocket socket = res.result();
    } else {
      System.out.println("Failed to connect: " + res.cause().getMessage());
    }
  });

Он может использоваться для различных целей:

указать имя сервера, отличное от имени хоста сервера
указать имя сервера при подключении к IP-адресу
обязательно указать имя сервера при использовании короткого имени

Переговоры о протоколе прикладного уровня (ALPN)

Переговоры о протоколе прикладного уровня (ALPN) — это расширение TLS для переговоров о протоколе прикладного уровня. Оно используется протоколом HTTP/2: во время рукопожатия TLS клиент предоставляет список принимаемых протоколов прикладного уровня, а сервер отвечает протоколом, который он поддерживает.

Java TLS поддерживает ALPN (Java 8 и более поздние версии).

Поддержка ALPN в OpenSSL

OpenSSL также поддерживает (родную) ALPN.

OpenSSL требует настройки setSslEngineOptions и использования jar-файла netty-tcnative в пути к классам. Использование tcnative может потребовать установки OpenSSL на вашей ОС в зависимости от реализации tcnative.

Использование прокси для подключений клиентов

NetClient поддерживает прокси HTTP/1.x CONNECT, SOCKS4a или SOCKS5.

Прокси можно настроить в NetClientOptions, задав объект ProxyOptions, содержащий тип прокси, имя хоста, порт и, необязательно, имя пользователя и пароль.

Вот пример:

NetClientOptions options = new NetClientOptions()
  .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5)
    .setHost("localhost").setPort(1080)
    .setUsername("username").setPassword("secret"));
NetClient client = vertx.createNetClient(options);

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

Можно использовать setNonProxyHosts для настройки списка хостов, обходящих прокси. Список принимает * подстановочный знак для сопоставления доменов:

NetClientOptions options = new NetClientOptions()
  .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5)
    .setHost("localhost").setPort(1080)
    .setUsername("username").setPassword("secret"))
  .addNonProxyHost("*.foo.com")
  .addNonProxyHost("localhost");
NetClient client = vertx.createNetClient(options);

Использование протокола HA PROXY

Протокол HA PROXY предоставляет удобный способ безопасной передачи информации о подключении, такой как адрес клиента, через несколько слоёв NAT или TCP прокси.

Протокол HA PROXY можно включить, задав опцию setUseProxyProtocol и добавив следующую зависимость в ваш класспаth:

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-codec-haproxy</artifactId>
  <!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

NetServerOptions options = new NetServerOptions().setUseProxyProtocol(true);
NetServer server = vertx.createNetServer(options);
server.connectHandler(so -> {
  // Print the actual client address provided by the HA proxy protocol instead of the proxy address
  System.out.println(so.remoteAddress());

  // Print the address of the proxy
  System.out.println(so.localAddress());
});

Написание HTTP серверов и клиентов

Vert.x позволяет легко писать неблокирующие HTTP клиенты и серверы.

Vert.x поддерживает протоколы HTTP/1.0, HTTP/1.1 и HTTP/2.

Основной API для HTTP одинаков для HTTP/1.x и HTTP/2, специфические возможности API доступны для работы с протоколом HTTP/2.

Создание HTTP сервера

Самый простой способ создать HTTP сервер, используя все параметры по умолчанию, выглядит следующим образом:

HttpServer server = vertx.createHttpServer();

Настройка HTTP сервера

Если вы не хотите использовать параметры по умолчанию, сервер можно настроить, передав объект HttpServerOptions при его создании:

HttpServerOptions options = new HttpServerOptions().setMaxWebSocketFrameSize(1000000);

HttpServer server = vertx.createHttpServer(options);

Настройка HTTP/2 сервера

Vert.x поддерживает HTTP/2 по TLS h2 и по TCP h2c.

h2 определяет протокол HTTP/2, когда используется по TLS, согласованный с Application-Layer Protocol Negotiation (ALPN)
h2c определяет протокол HTTP/2 при использовании в открытом тексте по TCP, такие подключения устанавливаются либо с помощью запроса обновления HTTP/1.1, либо непосредственно

Для обработки h2 запросов TLS должен быть включен вместе с setUseAlpn:

HttpServerOptions options = new HttpServerOptions()
    .setUseAlpn(true)
    .setSsl(true)
    .setKeyCertOptions(new JksOptions().setPath("/path/to/my/keystore"));

HttpServer server = vertx.createHttpServer(options);

ALPN — это расширение TLS, которое согласовывает протокол перед тем, как клиент и сервер начнут обмен данными.

Клиенты, которые не поддерживают ALPN, все равно смогут выполнить стандартное SSL соединение.

ALPN обычно согласовывает протокол h2, хотя http/1.1 может быть использован, если сервер или клиент так решат.

Для обработки h2c запросов TLS должен быть выключен, сервер будет обновлять подключение на HTTP/2 любой запрос HTTP/1.1, который хочет обновиться на HTTP/2. Он также будет принимать прямые h2c подключения, начинающиеся с PRI * HTTP/2.0\r\nSM\r\n префикса.

большинство браузеров не поддерживают h2c, поэтому для обслуживания веб-сайтов следует использовать h2, а не h2c.

Когда сервер принимает HTTP/2 соединение, он отправляет клиенту свои initial settings. Настройки определяют, как клиент может использовать подключение, значения по умолчанию для сервера следующие:

getMaxConcurrentStreams: 100, как рекомендуется в спецификации HTTP/2
Значения по умолчанию для других настроек HTTP/2.

Настройка поддерживаемых версий HTTP сервера

По умолчанию поддерживаемые версии HTTP зависят от конфигурации сервера.

при отключенном TLS
HTTP/1.1, HTTP/1.0
HTTP/2 при isHttp2ClearTextEnabled является true
при включенном TLS и отключенном ALPN
HTTP/1.1 и HTTP/1.0
при включенном TLS и включенном ALPN
протоколы, определенные в getAlpnVersions: по умолчанию HTTP/1.1 и HTTP/2

Если вы хотите отключить HTTP/2 на сервере - при отключенном TLS, установите setHttp2ClearTextEnabled в значение false - при включенном TLS - установите (isUseAlpn) в значение false - или удалите HTTP/2 из списка getAlpnVersions

Ведение журнала активности сетевого сервера

Для отладки может ведется журнал сетевой активности.

HttpServerOptions options = new HttpServerOptions().setLogActivity(true);

HttpServer server = vertx.createHttpServer(options);

См. главу по ведению журнала сетевой активности для подробного объяснения.

Запуск прослушивания сервера

Чтобы сообщить серверу о прослушивании входящих запросов, используйте один из вариантов listen.

Для прослушивания сервера на указанном в параметрах хосте и порту:

HttpServer server = vertx.createHttpServer();
server.listen();

Или для указания хоста и порта в вызове прослушивания, игнорируя конфигурацию в параметрах:

HttpServer server = vertx.createHttpServer();
server.listen(8080, "myhost.com");

По умолчанию хост - 0.0.0.0 (прослушивание на всех доступных адресах), а порт - 80.

Фактическое привязывание является асинхронным, поэтому сервер может не начать прослушивание сразу после возврата вызова прослушивания.

Если вы хотите получить уведомление о фактическом начале прослушивания сервером, вы можете предоставить обработчик для вызова listen. Например:

HttpServer server = vertx.createHttpServer();
server
  .listen(8080, "myhost.com")
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Server is now listening!");
    } else {
      System.out.println("Failed to bind!");
    }
  });

Прослушивание Unix-доменных сокетов

При работе на JDK 16+ или использовании нативного транспорта, сервер может прослушивать Unix-доменные сокеты:

HttpServer httpServer = vertx.createHttpServer();

// Only available when running on JDK16+, or using a native transport
SocketAddress address = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");

httpServer
  .requestHandler(req -> {
    // Handle application
  })
  .listen(address)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // Bound to socket
    } else {
      // Handle failure
    }
  });

Получение уведомлений о входящих запросах

Чтобы получить уведомления о прибытии запроса, необходимо установить requestHandler:

HttpServer server = vertx.createHttpServer();
server.requestHandler(request -> {
  // Handle the request in here
});

Обработка запросов

При поступлении запроса вызывается обработчик запроса, которому передается экземпляр HttpServerRequest. Этот объект представляет HTTP-запрос со стороны сервера.

Обработчик вызывается после полного чтения заголовков запроса.

Если запрос содержит тело, то это тело прибудет на сервер некоторое время спустя после вызова обработчика запроса.

Объект запроса сервера позволяет получить uri, path, params и headers, среди прочего.

Каждый объект запроса сервера связан с одним объектом ответа сервера. Вы используете response для получения ссылки на объект HttpServerResponse.

Вот простой пример сервера, обрабатывающего запрос и отвечающего на него «hello world».

vertx.createHttpServer().requestHandler(request -> {
  request.response().end("Hello world");
}).listen(8080);

Версия запроса

Версию HTTP, указанную в запросе, можно получить с помощью version

Метод запроса

Используйте method для получения метода HTTP-запроса. (т.е. GET, POST, PUT, DELETE, HEAD, OPTIONS и т.д.).

URI запроса

Используйте uri для получения URI запроса.

Обратите внимание, что это фактический URI, переданный в HTTP-запросе, и он почти всегда является относительным URI.

URI определен в разделе 5.1.2 спецификации HTTP — Request-URI

Путь запроса

Используйте path для возвращения части пути URI.

Например, если URI запроса был `a/b/c/page.html?param1=abc&param2=xyz`

Тогда путь будет /a/b/c/page.html

Параметр запроса

Используйте query для возвращения части параметра URI.

Например, если URI запроса был a/b/c/page.html?param1=abc&param2=xyz

Тогда параметр будет param1=abc&param2=xyz

Заголовки запроса

Используйте headers для возвращения заголовков HTTP-запроса.

Это возвращает экземпляр MultiMap — который похож на обычную Map или Hash, но допускает несколько значений для одного ключа — это связано с тем, что HTTP позволяет несколько значений заголовков с одним ключом.

Он также имеет регистронезависимые ключи, что означает, что вы можете сделать следующее:

MultiMap headers = request.headers();

// Get the User-Agent:
System.out.println("User agent is " + headers.get("user-agent"));

// You can also do this and get the same result:
System.out.println("User agent is " + headers.get("User-Agent"));

Власть запроса

Используйте authority для возвращения власти HTTP-запроса.

Для запросов HTTP/1.x возвращается заголовок host, для запросов HTTP/1 возвращается псевдозаголовок :authority.

Параметры запроса

Используйте params для возврата параметров HTTP-запроса.

Так же как и headers, это возвращает экземпляр MultiMap, так как может быть более одного параметра с одинаковым именем.

Параметры запроса передаются в URI запроса после пути. Например, если URI был /page.html?param1=abc&param2=xyz

Тогда параметры будут содержать следующее:

param1: 'abc'
param2: 'xyz

Обратите внимание, что эти параметры запроса извлекаются из URL запроса. Если у вас есть атрибуты формы, которые были отправлены в качестве части отправки HTML-формы, отправленной в теле multi-part/form-data запроса, то они не появятся в параметрах здесь.

Удаленный адрес

Адрес отправителя запроса можно получить с помощью remoteAddress.

Абсолютный URI

URI, переданный в HTTP-запросе, обычно относительный. Если вы хотите получить абсолютный URI, соответствующий запросу, вы можете получить его с помощью absoluteURI

Обработчик завершения

endHandler запроса вызывается, когда весь запрос, включая тело, полностью прочитан.

Чтение данных из тела запроса

Часто HTTP-запрос содержит тело, которое мы хотим прочитать. Как уже упоминалось, обработчик запроса вызывается, когда только пришли заголовки запроса, поэтому в объекте запроса нет тела в этот момент.

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

Чтобы получить тело, вы можете использовать handler в запросе, это будет вызываться каждый раз, когда приходит фрагмент тела запроса. Вот пример:

request.handler(buffer -> {
  System.out.println("I have received a chunk of the body of length " + buffer.length());
});

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

В некоторых случаях (например, если тело небольшое) вы захотите агрегировать все тело в памяти, поэтому вы можете сделать агрегацию самостоятельно следующим образом:

Buffer totalBuffer = Buffer.buffer();

request.handler(buffer -> {
  System.out.println("I have received a chunk of the body of length " + buffer.length());
  totalBuffer.appendBuffer(buffer);
});

request.endHandler(v -> {
  System.out.println("Full body received, length = " + totalBuffer.length());
});

Это настолько распространённый случай, что Vert.x предоставляет bodyHandler для этого. Обработчик тела вызывается один раз, когда всё тело получено:

request.bodyHandler(totalBuffer -> {
  System.out.println("Full body received, length = " + totalBuffer.length());
});

Потоковые запросы

Объект запроса является ReadStream, поэтому вы можете передать тело запроса в любой объект WriteStream.

См. главу о потоках для подробного объяснения.

Обработка HTML-форм

HTML-формы могут быть отправлены с типом контента application/x-www-form-urlencoded или multipart/form-data.

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

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

Формы с многочастью также могут содержать загрузку файлов.

Если вы хотите получить атрибуты формы с многочастью, вы должны сообщить Vert.x, что ожидаете получить такую форму перед чтением любого тела, вызвав setExpectMultipart с true, а затем получить фактические атрибуты с помощью formAttributes после того, как всё тело будет прочитано:

server.requestHandler(request -> {
  request.setExpectMultipart(true);
  request.endHandler(v -> {
    // The body has now been fully read, so retrieve the form attributes
    MultiMap formAttributes = request.formAttributes();
  });
});

Атрибуты формы имеют максимальный размер 8192 байта. Когда клиент отправляет форму с размером атрибута, превышающим это значение, загрузка файла вызывает исключение в обработчике исключений HttpServerRequest. Вы можете установить другой максимальный размер с помощью setMaxFormAttributeSize.

Обработка загрузки файлов формы

Vert.x также может обрабатывать загрузку файлов, закодированные в теле многочастичного запроса.

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

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

Объект, переданный в обработчик, является экземпляром HttpServerFileUpload.

server.requestHandler(request -> {
  request.setExpectMultipart(true);
  request.uploadHandler(upload -> {
    System.out.println("Got a file upload " + upload.name());
  });
});

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

request.uploadHandler(upload -> {
  upload.handler(chunk -> {
    System.out.println("Received a chunk of the upload of length " + chunk.length());
  });
});

Объект загрузки — это ReadStream, поэтому вы можете перенаправить тело запроса в любой экземпляр WriteStream. Подробное объяснение см. в главе о потоках.

Если вы хотите просто загрузить файл на диск, вы можете использовать streamToFileSystem:

request.uploadHandler(upload -> {
  upload.streamToFileSystem("myuploads_directory/" + upload.filename());
});

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

Обработка файлов cookie

Для получения файла cookie по имени используйте getCookie, или для получения всех файлов cookie используйте cookies.

Для удаления файла cookie используйте removeCookie.

Для добавления файла cookie используйте addCookie.

Набор файлов cookie будет автоматически записан обратно в ответ при записи заголовков ответа, чтобы браузер мог их сохранить.

Файлы cookie описываются экземплярами Cookie. Это позволяет получить имя, значение, домен, путь и другие обычные свойства файлов cookie.

Файлы cookie SameSite позволяют серверам требовать, чтобы файл cookie не отправлялся с запросами на другие сайты (где сайт определяется регистрируемым доменом), что обеспечивает некоторую защиту от межсайтовых поддельных запросов. Такой тип файлов cookie активируется с помощью метода установки: setSameSite.

Файлы cookie SameSite могут иметь одно из 3 значений:

None — браузер будет отправлять файлы cookie как с межсайтовыми, так и с односайтовыми запросами.
Strict — браузер будет отправлять файлы cookie только для односайтовых запросов (запросы, исходящие с сайта, который установил файл cookie). Если запрос исходил из другого URL-адреса, чем URL-адрес текущего местоположения, ни один из файлов cookie, помеченных атрибутом Strict, не будет включен.
Lax — файлы cookie SameSite удерживаются при межсайтовых подзапросах, таких как вызовы для загрузки изображений или фреймов, но будут отправляться при переходе пользователя по URL-адресу с внешнего сайта; например, при переходе по ссылке.

Вот пример запроса и добавления файлов cookie:

Cookie someCookie = request.getCookie("mycookie");
String cookieValue = someCookie.getValue();

// Do something with cookie...

// Add a cookie - this will get written back in the response automatically
request.response().addCookie(Cookie.cookie("othercookie", "somevalue"));

Обработка сжатых тел

Vert.x может обрабатывать сжатые тела полезной нагрузки, закодированные клиентом с помощью алгоритмов deflate, gzip, snappy или brotli.

Для включения распаковки установите setDecompressionSupported в параметрах при создании сервера.

Snappy поддерживается без внешних зависимостей.

Для распаковки Brotli вам нужна библиотека Brotli4j в классе, а для Zstandard — Zstd-jni:

Maven (в вашем pom.xml):

<dependency>
  <groupId>com.aayushatharva.brotli4j</groupId>
  <artifactId>brotli4j</artifactId>
  <version>${brotli4j.version}</version>
</dependency>
<dependency>
  <groupId>com.github.luben</groupId>
  <artifactId>zstd-jni</artifactId>
  <version>${zstd-jini.version}</version>
</dependency>

Gradle (в вашем файле build.gradle):

dependencies {
  implementation 'com.aayushatharva.brotli4j:brotli4j:${brotli4j.version}'
  runtimeOnly 'com.aayushatharva.brotli4j:native-$system-and-arch:${brotli4j.version}'
  implementation 'com.github.luben:zstd-jni:${zstd-jini.version}'
}

При использовании Gradle вам нужно вручную добавить нативные библиотеки runtime в зависимости от вашей ОС и архитектуры. Дополнительные сведения см. в разделе Gradle в Brotli4j.

По умолчанию распаковка отключена.

Прием пользовательских кадров HTTP/2

HTTP/2 — это протокол в виде кадров с различными кадрами для модели запроса/ответа HTTP. Протокол позволяет отправлять и получать другие типы кадров.

Для приема пользовательских кадров вы можете использовать customFrameHandler в запросе, этот метод будет вызываться каждый раз при поступлении пользовательского кадра. Вот пример:

request.customFrameHandler(frame -> {

  System.out.println("Received a frame type=" + frame.type() +
      " payload" + frame.payload().toString());
});

Кадры HTTP/2 не подлежат управлению потоком — обработчик кадров будет вызываться немедленно при поступлении пользовательского кадра, независимо от того, приостановлен ли запрос или нет.

Отправка ответов

Объект ответа сервера является экземпляром HttpServerResponse и получается из запроса с помощью response.

Вы используете объект ответа для записи ответа клиенту HTTP.

Установка кода и сообщения состояния

По умолчанию код состояния HTTP для ответа равен 200, что обозначает OK.

Для установки другого кода используйте setStatusCode.

Вы также можете указать пользовательское сообщение состояния с помощью setStatusMessage.

Если вы не укажете сообщение состояния, будет использовано значение по умолчанию, соответствующее коду состояния.

для HTTP/2 код состояния отсутствует в ответе, так как протокол не передаёт сообщение клиенту

Запись HTTP ответов

Для записи данных в HTTP ответ используйте одну из операций write.

Эти операции могут быть вызваны несколько раз до завершения ответа. Они могут быть вызваны несколькими способами:

С одним буфером:

HttpServerResponse response = request.response();
response.write(buffer);

С строкой. В этом случае строка будет закодирована в UTF-8, а результат записан в сеть.

HttpServerResponse response = request.response();
response.write("hello world!");

Со строкой и кодировкой. В этом случае строка будет закодирована с помощью указанной кодировки, а результат записан в сеть.

HttpServerResponse response = request.response();
response.write("hello world!", "UTF-16");

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

Если вы просто записываете одну строку или буфер в HTTP ответ, вы можете записать его и завершить ответ одним вызовом end.

Первый вызов записи приводит к записи заголовков ответа. Следовательно, если вы не используете HTTP чанкинг, вы должны установить заголовок Content-Length до записи в ответ, иначе будет слишком поздно. Если вы используете HTTP чанкинг, вам об этом беспокоиться не нужно.

Завершение HTTP ответов

После завершения работы с HTTP ответом вы должны end его.

Это можно сделать несколькими способами:

Без аргументов ответ просто завершается.

HttpServerResponse response = request.response();
response.write("hello world!");
response.end();

Его также можно вызвать со строкой или буфером таким же образом, как вызывается write. В этом случае это то же самое, что вызов write со строкой или буфером, за которым следует вызов end без аргументов. Например:

HttpServerResponse response = request.response();
response.end("hello world!");

Закрытие базового TCP-соединения

Вы можете закрыть базовое TCP-соединение с помощью close.

Соединения без поддержания соединения (keep-alive) будут автоматически закрыты Vert.x, когда ответ будет завершён.

Соединения с поддержанием соединения (keep-alive) по умолчанию не закрываются автоматически Vert.x. Если вы хотите, чтобы соединения с поддержанием соединения закрывались после простоя, настройте setIdleTimeout.

Соединения HTTP/2 отправляют кадр GOAWAY перед закрытием ответа.

Установка заголовков ответа

Заголовки HTTP-ответа можно добавить, напрямую добавив их в headers:

HttpServerResponse response = request.response();
MultiMap headers = response.headers();
headers.set("content-type", "text/html");
headers.set("other-header", "wibble");

Или вы можете использовать putHeader

HttpServerResponse response = request.response();
response.putHeader("content-type", "text/html").putHeader("other-header", "wibble");

Все заголовки должны быть добавлены до записи каких-либо частей тела ответа.

Размеченные HTTP-ответы и трейлеры

Vert.x поддерживает кодирование HTTP-ответа частями.

Это позволяет записать тело HTTP-ответа частями, что обычно используется, когда большой объём ответа передаётся клиенту, а его общий размер не известен заранее.

Вы переводите HTTP-ответ в режим работы частями следующим образом:

HttpServerResponse response = request.response();
response.setChunked(true);

По умолчанию режим работы частями не используется. При работе в режиме частями каждый вызов одного из write методов приведет к записи нового фрагмента HTTP.

При работе в режиме частями вы также можете записать трейлеры HTTP-ответа в ответ. Они фактически записываются в последнем фрагменте ответа.

режим работы частями не влияет на поток HTTP/2

Чтобы добавить трейлеры в ответ, добавьте их напрямую в trailers.

HttpServerResponse response = request.response();
response.setChunked(true);
MultiMap trailers = response.trailers();
trailers.set("X-wibble", "woobble").set("X-quux", "flooble");

Или используйте putTrailer.

HttpServerResponse response = request.response();
response.setChunked(true);
response.putTrailer("X-wibble", "woobble").putTrailer("X-quux", "flooble");

Прямая подача файлов с диска или из класса

Если вы пишете веб-сервер, один из способов подачи файла с диска — открыть его как AsyncFile и направить его в ответ HTTP.

Или вы можете загрузить его целиком с помощью readFile и записать его напрямую в ответ.

В качестве альтернативы, Vert.x предоставляет метод, позволяющий подавать файл с диска или из файловой системы в ответ HTTP за одну операцию. Там, где это поддерживается основной операционной системой, это может привести к тому, что ОС напрямую переместит байты из файла в сокет, вообще не копируя их через пользовательское пространство.

Это делается с помощью sendFile, и обычно более эффективно для больших файлов, но может быть медленнее для небольших файлов.

Вот очень простой веб-сервер, который подаёт файлы из файловой системы с помощью sendFile:

vertx.createHttpServer().requestHandler(request -> {
  String file = "";
  if (request.path().equals("/")) {
    file = "index.html";
  } else if (!request.path().contains("..")) {
    file = request.path();
  }
  request.response().sendFile("web/" + file);
}).listen(8080);

Заголовок HTTP-ответа использует расширение имени файла для установки заголовка типа содержимого HTTP-ответа, когда расширение имени файла хорошо известно MimeMapping (поиск нечувствителен к регистру).

Отправка файла выполняется асинхронно и может не завершиться до некоторого времени после возвращения вызова. Если вы хотите получить уведомление о завершении записи файла, можете использовать sendFile.

Обратитесь к главе о подачи файлов из класса для ограничений по разрешению класса или его отключению.

Если вы используете sendFile при использовании HTTPS, он будет копировать через пользовательское пространство, так как если ядро копирует данные напрямую с диска в сокет, это не дает нам возможности применять какое-либо шифрование.

Если вы собираетесь писать веб-серверы напрямую с помощью Vert.x, будьте внимательны, чтобы пользователи не могли использовать путь для доступа к файлам за пределами каталога, из которого вы хотите их подавать, или из класса. Возможно, безопаснее вместо этого использовать Vert.x Web.

Когда необходимо подать только часть файла, например, начиная с заданного байта, вы можете сделать это следующим образом:

vertx.createHttpServer().requestHandler(request -> {
  long offset = 0;
  try {
    offset = Long.parseLong(request.getParam("start"));
  } catch (NumberFormatException e) {
    // error handling...
  }

  long end = Long.MAX_VALUE;
  try {
    end = Long.parseLong(request.getParam("end"));
  } catch (NumberFormatException e) {
    // error handling...
  }

  request.response().sendFile("web/mybigfile.txt", offset, end);
}).listen(8080);

Вам не нужно указывать длину, если вы хотите отправить файл, начиная с смещения, до конца. В этом случае вы можете просто сделать так:

vertx.createHttpServer().requestHandler(request -> {
  long offset = 0;
  try {
    offset = Long.parseLong(request.getParam("start"));
  } catch (NumberFormatException e) {
    // error handling...
  }

  request.response().sendFile("web/mybigfile.txt", offset);
}).listen(8080);

Перенаправление ответов

Ответ сервера — это WriteStream, поэтому вы можете передать в него данные из любого ReadStream, например, AsyncFile, NetSocket, WebSocket или HttpServerRequest.

Вот пример, который повторяет тело запроса в ответе для любых методов PUT. Он использует передачу для тела, поэтому он будет работать даже если тело HTTP-запроса намного больше, чем может поместиться в памяти за один раз:

vertx.createHttpServer().requestHandler(request -> {
  HttpServerResponse response = request.response();
  if (request.method() == HttpMethod.PUT) {
    response.setChunked(true);
    request.pipeTo(response);
  } else {
    response.setStatusCode(400).end();
  }
}).listen(8080);

Вы также можете использовать метод send для отправки ReadStream.

Отправка потока — это операция перенаправления, однако, поскольку это метод HttpServerResponse, он также будет заботиться о разбиении ответа на части, когда content-length не задан.

vertx.createHttpServer().requestHandler(request -> {
  HttpServerResponse response = request.response();
  if (request.method() == HttpMethod.PUT) {
    response.send(request);
  } else {
    response.setStatusCode(400).end();
  }
}).listen(8080);

Написание кадров HTTP/2

HTTP/2 — это протокол с кадрами, имеющими различные кадры для модели запроса/ответа HTTP. Протокол позволяет отправлять и получать другие типы кадров.

Для отправки таких кадров можно использовать writeCustomFrame в ответе. Вот пример:

int frameType = 40;
int frameStatus = 10;
Buffer payload = Buffer.buffer("some data");

// Sending a frame to the client
response.writeCustomFrame(frameType, frameStatus, payload);

Эти кадры отправляются немедленно и не подлежат управлению потоком — при отправке такого кадра это может быть сделано до других DATA кадров.

Сброс потока

HTTP/1.x не позволяет выполнить чистый сброс потока запроса или ответа, например, когда клиент загружает ресурс, уже присутствующий на сервере, серверу необходимо принять весь ответ.

HTTP/2 поддерживает сброс потока в любое время во время запроса/ответа:

request.response().reset();

По умолчанию отправляется код ошибки NO_ERROR (0), вместо него можно отправить другой код:

request.response().reset(8);

Спецификация HTTP/2 определяет список кодов ошибок, которые можно использовать.

Обработчик запросов уведомляется о событиях сброса потока с помощью request handler и response handler:

request.response().exceptionHandler(err -> {
  if (err instanceof StreamResetException) {
    StreamResetException reset = (StreamResetException) err;
    System.out.println("Stream reset " + reset.getCode());
  }
});

Продвижение сервера

Продвижение сервера — это новая функция HTTP/2, которая позволяет отправлять несколько ответов параллельно для одного запроса клиента.

Когда сервер обрабатывает запрос, он может направить запрос/ответ клиенту:

HttpServerResponse response = request.response();

// Push main.js to the client
response
  .push(HttpMethod.GET, "/main.js")
  .onComplete(ar -> {

    if (ar.succeeded()) {

      // The server is ready to push the response
      HttpServerResponse pushedResponse = ar.result();

      // Send main.js response
      pushedResponse.
        putHeader("content-type", "application/json").
        end("alert(\"Push response hello\")");
    } else {
      System.out.println("Could not push client resource " + ar.cause());
    }
  });

// Send the requested resource
response.sendFile("<html><head><script src=\"/main.js\"></script></head><body></body></html>");

Когда сервер готов отправить ответ, вызывается обработчик ответа продвижения, и обработчик может отправить ответ.

Обработчик ответа продвижения может получить ошибку, например, клиент может отменить продвижение, потому что у него уже есть main.js в кэше и он больше не хочет его.

Метод push должен быть вызван до завершения начального ответа, однако ответ продвижения может быть записан позже.

Обработка исключений

Можно установить exceptionHandler для получения любых исключений, которые возникают до того, как соединение передается в requestHandler или webSocketHandler, например, во время рукопожатия TLS.

Обработка недопустимых запросов

Vert.x будет обрабатывать недопустимые запросы HTTP и предоставляет обработчик по умолчанию, который будет должным образом обрабатывать распространенный случай, например, он отвечает с REQUEST_HEADER_FIELDS_TOO_LARGE, когда заголовок запроса слишком длинный.

Можно задать собственный invalidRequestHandler для обработки недопустимых запросов. Ваша реализация может обрабатывать конкретные случаи и делегировать другие случаи в HttpServerRequest.DEFAULT_INVALID_REQUEST_HANDLER.

Сжатие HTTP

Vert.x предоставляет поддержку HTTP-сжатия прямо из коробки.

Это означает, что вы можете автоматически сжимать тело ответов перед отправкой клиенту.

Если клиент не поддерживает HTTP-сжатие, ответы отправляются без сжатия тела.

Это позволяет обрабатывать клиентов, которые поддерживают HTTP-сжатие, и тех, которые не поддерживают его, одновременно.

Для включения сжатия можно настроить его с помощью setCompressionSupported.

По умолчанию сжатие не включено.

При включенном HTTP-сжатии сервер проверит, включает ли клиент заголовок Accept-Encoding, который содержит поддерживаемые виды сжатия. Обычно используются deflate и gzip. Оба поддерживаются Vert.x.

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

В случае необходимости отправить ответ без сжатия, можно установить заголовок content-encoding в значение identity:

request.response()
  .putHeader(HttpHeaders.CONTENT_ENCODING, HttpHeaders.IDENTITY)
  .sendFile("/path/to/image.jpg");

Обратите внимание, что сжатие может уменьшить трафик сети, но при этом требует больше ресурсов процессора.

Для решения этой проблемы Vert.x позволяет настроить параметр «уровень сжатия», который является собственным параметром алгоритмов gzip/deflate, а также установить пороговое значение размера содержимого ответа для сжатия.

Уровень сжатия позволяет настроить алгоритмы gzip/deflate с точки зрения соотношения сжатия результирующих данных и вычислительных затрат на операцию сжатия/распаковки.

Уровень сжатия — это целое число от «1» до «9», где «1» означает меньшее соотношение сжатия, но самый быстрый алгоритм, а «9» означает максимальное соотношение сжатия, но более медленный алгоритм.

Использование уровней сжатия выше 1-2 обычно позволяет сэкономить только некоторые байты в размере — прирост нелинейный и зависит от конкретных данных, которые нужно сжать — но это влечёт существенные затраты на процессорные циклы сервера при генерации сжатых данных ответа ( Обратите внимание, что Vert.x в настоящее время не поддерживает кэширование сжатых данных ответа, даже для статических файлов, поэтому сжатие выполняется на лету при каждом формировании тела запроса ) и аналогично влияет на клиент(ы) при декодировании (распаковке) полученных ответов, операция, которая становится более ресурсоёмкой, чем выше уровень.

По умолчанию — если сжатие включено через setCompressionSupported — Vert.x будет использовать «6» в качестве уровня сжатия, но этот параметр может быть настроен для любых случаев с помощью setCompressionLevel.

Сжимать ответы, размер которых ниже определённых пороговых значений, может не иметь смысла, так как соотношение затрат процессора и сэкономленных байтов сети не выгодное. Пороговое значение минимального размера содержимого ответа для сжатия можно настроить с помощью setCompressionContentSizeThreshold. Например, если значение установлено в «100», ответы размером меньше 100 байт не будут сжиматься. По умолчанию это «0», что означает, что всё содержимое может быть сжато.

Алгоритмы сжатия HTTP

Vert.x поддерживает deflate и gzip по умолчанию.

Также можно использовать Brotli, snappy и zstandard.

new HttpServerOptions()
  .addCompressor(io.netty.handler.codec.compression.StandardCompressionOptions.gzip())
  .addCompressor(io.netty.handler.codec.compression.StandardCompressionOptions.deflate())
  .addCompressor(io.netty.handler.codec.compression.StandardCompressionOptions.brotli())
  .addCompressor(io.netty.handler.codec.compression.StandardCompressionOptions.zstd());

используйте StandardCompressionOptions статические методы для создания CompressionOptions

Для библиотек Brotli и zstandard необходимо добавить их в класспуть, snappy предоставляется по умолчанию.

Maven (в вашем файле pom.xml):

<dependency>
  <groupId>com.aayushatharva.brotli4j</groupId>
  <artifactId>brotli4j</artifactId>
  <version>${brotli4j.version}</version>
</dependency>
<dependency>
  <groupId>com.github.luben</groupId>
  <artifactId>zstd-jni</artifactId>
  <version>${zstd-jini.version}</version>
</dependency>

Gradle (в вашем файле build.gradle):

dependencies {
  implementation 'com.aayushatharva.brotli4j:brotli4j:${brotli4j.version}'
  runtimeOnly 'com.aayushatharva.brotli4j:native-$system-and-arch:${brotli4j.version}'
  implementation 'com.github.luben:zstd-jni:${zstd-jini.version}'
}

При использовании Gradle, вам необходимо вручную добавить родную библиотеку времени выполнения в зависимости от вашей операционной системы и архитектуры. Смотрите раздел Gradle в Brotli4j для получения дополнительных сведений.

Вы можете настроить сжатие в соответствии со своими потребностями.

GzipOptions gzip = StandardCompressionOptions.gzip(6, 15, 8);

Создание HTTP-клиента

Вы создаёте экземпляр HttpClient с параметрами по умолчанию следующим образом:

HttpClientAgent client = vertx.createHttpClient();

Если вы хотите настроить параметры для клиента, вы создаёте его следующим образом:

HttpClientOptions options = new HttpClientOptions().setKeepAlive(false);
HttpClientAgent client = vertx.createHttpClient(options);

Vert.x поддерживает HTTP/2 по TLS h2 и по TCP h2c.

По умолчанию HTTP-клиент выполняет запросы HTTP/1.1. Для выполнения запросов HTTP/2 параметр setProtocolVersion должен быть установлен в HTTP_2.

Для запросов h2, TLS должен быть включён с помощью Application-Layer Protocol Negotiation:

HttpClientOptions options = new HttpClientOptions().
    setProtocolVersion(HttpVersion.HTTP_2).
    setSsl(true).
    setUseAlpn(true).
    setTrustAll(true);

HttpClient client = vertx.createHttpClient(options);

Для запросов h2c, TLS должен быть отключён, клиент выполнит запросы HTTP/1.1 и попробует выполнить апгрейд до HTTP/2:

HttpClientOptions options = new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2);

HttpClient client = vertx.createHttpClient(options);

Установление соединений h2c также может быть выполнено непосредственно, т.е. соединение инициируется с предварительным знанием, когда параметр setHttp2ClearTextUpgrade установлен в false: после установления соединения, клиент отправит префикс соединения HTTP/2 и ожидает получения такого же префикса от сервера.

HTTP-сервер может не поддерживать HTTP/2, фактическую версию можно проверить с помощью version при получении ответа.

При подключении клиента к HTTP/2-серверу клиент отправляет серверу свои initial settings. Эти настройки определяют, как сервер может использовать соединение. Начальные настройки клиента по умолчанию — значения по умолчанию, определенные спецификацией HTTP/2 RFC.

Установление соединений с Unix-сокетных доменах

При работе с JDK 16+ или использовании родного транспорта клиент может подключаться к Unix-сокетных доменах:

HttpClient httpClient = vertx.createHttpClient();

// Only available when running on JDK16+, or using a native transport
SocketAddress addr = SocketAddress.domainSocketAddress("/var/tmp/myservice.sock");

// Send request to the server
httpClient.request(new RequestOptions()
  .setServer(addr)
  .setHost("localhost")
  .setPort(8080)
  .setURI("/"))
  .compose(request -> request.send().compose(HttpClientResponse::body))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // Process response
    } else {
      // Handle failure
    }
  });

Настройка пула

Для повышения производительности клиент использует пул подключений при взаимодействии с серверами HTTP/1.1. Пул создаёт до 5 подключений на сервер. Вы можете переопределить настройку пула следующим образом:

PoolOptions options = new PoolOptions().setHttp1MaxSize(10);
HttpClientAgent client = vertx.createHttpClient(options);

Вы можете настроить различные параметры пула следующим образом:

options#setHttp1MaxSize максимальное количество открытых подключений на сервер HTTP/1.x (по умолчанию 5)
options#setHttp2MaxSize максимальное количество открытых подключений на сервер HTTP/2 (по умолчанию 1), вы не должны изменять это значение, так как одно HTTP/2 подключение способно обеспечить тот же уровень производительности, что и несколько подключений HTTP/1.x
options#setCleanerPeriod период в миллисекундах, в течение которого пул проверяет истекшие подключения (по умолчанию 1 секунда)
options#setEventLoopSize устанавливает количество циклов обработки событий, которые использует пул (по умолчанию 0)
Значение 0 настраивает пул на использование цикла обработки событий вызывающего объекта.
Положительное значение настраивает пул на балансировку загрузки создания подключений по списку циклов обработки событий, определяемых значением.
options#setMaxWaitQueueSize максимальное количество запросов HTTP, ожидающих подключения, когда очередь полная, запрос отклоняется.

Ведение журнала активности сетевого клиента

Для отладки может быть залогирована сетевая активность.

HttpClientOptions options = new HttpClientOptions().setLogActivity(true);
HttpClientAgent client = vertx.createHttpClient(options);

См. главу о ведении журнала сетевой активности для получения подробных объяснений.

Расширенное создание HTTP клиента

Вы можете передать параметры createHttpClient методы для настройки HTTP клиента.

Альтернативно, вы можете создать клиент с помощью билдера API :

HttpClientAgent build = vertx
  .httpClientBuilder()
  .with(options)
  .build();

В дополнение к HttpClientOptions и PoolOptions, вы можете установить

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

Выполнение запросов

Клиент HTTP очень гибкий, и есть различные способы отправки запросов.

Первый шаг при отправке запроса — получение HTTP-соединения с удалённым сервером:

client
  .request(HttpMethod.GET, 8080, "myserver.mycompany.com", "/some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      // Connected to the server
    }
  });

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

По умолчанию хост и порт

Часто требуется отправлять много запросов на один и тот же хост/порт с помощью клиента HTTP. Чтобы избежать повторного указания хоста/порта каждый раз при отправке запроса, можно настроить клиента с хостом/портом по умолчанию:

HttpClientOptions options = new HttpClientOptions().setDefaultHost("wibble.com");

// Can also set default port if you want...
HttpClientAgent client = vertx.createHttpClient(options);
client
  .request(HttpMethod.GET, "/some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      HttpClientRequest request = ar1.result();
      request
        .send()
        .onComplete(ar2 -> {
          if (ar2.succeeded()) {
            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          }
        });
    }
  });

Запись заголовков запроса

Заголовки запроса можно записать, используя HttpHeaders следующим образом:

HttpClientAgent client = vertx.createHttpClient();

// Write some headers using the headers multi-map
MultiMap headers = HttpHeaders.set("content-type", "application/json").set("other-header", "foo");

client
  .request(HttpMethod.GET, "some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      if (ar1.succeeded()) {
        HttpClientRequest request = ar1.result();
        request.headers().addAll(headers);
        request
          .send()
          .onComplete(ar2 -> {
            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          });
      }
    }
  });

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

Заголовки также можно записать, используя putHeader

request.putHeader("content-type", "application/json")
       .putHeader("other-header", "foo");

Если необходимо записать заголовки запроса, необходимо сделать это до записи любой части тела запроса.

Отправка запроса и обработка ответа

Методы HttpClientRequest request подключаются к удалённому серверу или повторно используют существующее соединение. Полученный экземпляр запроса предварительно заполнен некоторыми данными, такими как хост или URI запроса, но вам нужно отправить этот запрос на сервер.

Можно вызвать send для отправки запроса, например, HTTP-GET, и обработать асинхронный HttpClientResponse.

client
  .request(HttpMethod.GET, 8080, "myserver.mycompany.com", "/some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      HttpClientRequest request = ar1.result();

      // Send the request and process the response
      request
        .send()
        .onComplete(ar -> {
          if (ar.succeeded()) {
            HttpClientResponse response = ar.result();
            System.out.println("Received response with status code " + response.statusCode());
          } else {
            System.out.println("Something went wrong " + ar.cause().getMessage());
          }
        });
    }
  });

Также можно отправить запрос с телом.

send со строкой, заголовок Content-Length будет задан для вас, если он ранее не был задан.

client
  .request(HttpMethod.GET, 8080, "myserver.mycompany.com", "/some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      HttpClientRequest request = ar1.result();

      // Send the request and process the response
      request
        .send("Hello World")
        .onComplete(ar -> {
          if (ar.succeeded()) {
            HttpClientResponse response = ar.result();
            System.out.println("Received response with status code " + response.statusCode());
          } else {
            System.out.println("Something went wrong " + ar.cause().getMessage());
          }
        });
    }
  });

send с буфером, заголовок Content-Length будет задан для вас, если он ранее не был задан.

request
  .send(Buffer.buffer("Hello World"))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      HttpClientResponse response = ar.result();
      System.out.println("Received response with status code " + response.statusCode());
    } else {
      System.out.println("Something went wrong " + ar.cause().getMessage());
    }
  });

send со потоком, если заголовок Content-Length не был ранее задан, запрос отправляется с раздробленным Content-Encoding.

request
  .putHeader(HttpHeaders.CONTENT_LENGTH, "1000")
  .send(stream)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      HttpClientResponse response = ar.result();
      System.out.println("Received response with status code " + response.statusCode());
    } else {
      System.out.println("Something went wrong " + ar.cause().getMessage());
    }
  });

Тело запроса потоковой передачи

Метод send отправляет запросы сразу.

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

Для формирования тела запроса можно использовать HttpClientRequest.

Ниже приведены примеры отправки запроса POST с телом:

HttpClientAgent client = vertx.createHttpClient();

client.request(HttpMethod.POST, "some-uri")
  .onSuccess(request -> {
    request.response().onSuccess(response -> {
      System.out.println("Received response with status code " + response.statusCode());
    });

    // Now do stuff with the request
    request.putHeader("content-length", "1000");
    request.putHeader("content-type", "text/plain");
    request.write(body);

    // Make sure the request is ended when you're done with it
    request.end();
});

Существуют методы для записи строк в кодировке UTF-8 и в любой другой кодировке, а также для записи буферов:

request.write("some data");

// Write string encoded in specific encoding
request.write("some other data", "UTF-16");

// Write a buffer
Buffer buffer = Buffer.buffer();
buffer.appendInt(123).appendLong(245l);
request.write(buffer);

Если вы просто записываете одну строку или буфер в HTTP-запрос, вы можете сделать это и завершить запрос в одном вызове функции end.

request.end("some simple data");

// Write buffer and end the request (send it) in a single call
Buffer buffer = Buffer.buffer().appendDouble(12.34d).appendLong(432l);
request.end(buffer);

При записи в запрос первый вызов write приведет к записи заголовков запроса в сеть.

Фактическая запись асинхронная и может произойти некоторое время после возврата вызова.

Неразбитые HTTP-запросы с телом запроса требуют наличия заголовка Content-Length.

Следовательно, если вы не используете разбитый HTTP, вы должны установить заголовок Content-Length перед записью в запрос, так как в противном случае будет слишком поздно.

Если вы вызываете один из методов end, принимающих строку или буфер, Vert.x автоматически рассчитает и установит заголовок Content-Length перед записью тела запроса.

Если вы используете разбитый HTTP, заголовок Content-Length не требуется, поэтому не нужно предварительно вычислять размер.

Завершение запросов потоковой передачи HTTP

После завершения работы с HTTP-запросом вы должны завершить его одной из операций end.

Завершение запроса приводит к записи любых заголовков, если они еще не были записаны, и к маркировке запроса как завершенного.

Запросы могут быть завершены несколькими способами. Без аргументов запрос просто завершается:

request.end();

Или строка или буфер могут быть предоставлены в вызове end. Это аналогично вызову write со строкой или буфером перед вызовом end без аргументов.

request.end("some-data");

// End it with a buffer
Buffer buffer = Buffer.buffer().appendFloat(12.3f).appendInt(321);
request.end(buffer);

Использование запроса как потока

Экземпляр HttpClientRequest также является экземпляром WriteStream.

Вы можете передавать в него данные из любого экземпляра ReadStream.

Например, вы можете передать файл на диске в тело HTTP-запроса следующим образом:

request.setChunked(true);
file.pipeTo(request);

Чанкированные HTTP-запросы

Vert.x поддерживает HTTP Chunked Transfer Encoding для запросов.

Это позволяет разбивать тело HTTP-запроса на куски и обычно используется, когда большое тело запроса передаётся на сервер потоком, размер которого заранее неизвестен.

Вы можете установить HTTP-запрос в режим чанков, используя setChunked.

В режиме чанков каждый вызов write приведет к записи нового куска в сеть. В режиме чанков нет необходимости задавать Content-Length запроса заранее.

request.setChunked(true);

// Write some chunks
for (int i = 0; i < 10; i++) {
  request.write("this-is-chunk-" + i);
}

request.end();

Отправка форм

Вы можете отправлять тела HTTP-запросов в формате формы с вариантом send.

ClientForm form = ClientForm.form();
form.attribute("firstName", "Dale");
form.attribute("lastName", "Cooper");

// Submit the form as a form URL encoded body
request
  .send(form)
  .onSuccess(res -> {
    // OK
  });

По умолчанию форма отправляется с заголовком типа контента application/x-www-form-urlencoded. Вы можете установить заголовок content-type на multipart/form-data вместо этого.

ClientForm form = ClientForm.form();
form.attribute("firstName", "Dale");
form.attribute("lastName", "Cooper");

// Submit the form as a multipart form body
request
  .putHeader("content-type", "multipart/form-data")
  .send(form)
  .onSuccess(res -> {
    // OK
  });

Если вы хотите загружать файлы и отправлять атрибуты, вы можете создать ClientMultipartForm.

ClientMultipartForm form = ClientMultipartForm.multipartForm()
  .attribute("imageDescription", "a very nice image")
  .binaryFileUpload(
    "imageFile",
    "image.jpg",
    "/path/to/image",
    "image/jpeg");

// Submit the form as a multipart form body
request
  .send(form)
  .onSuccess(res -> {
    // OK
  });

Таймауты запросов

Вы можете установить таймаут ожидания, чтобы предотвратить неотзывчивость вашего приложения к серверам, используя setIdleTimeout или idleTimeout. Если запрос не возвращает данные в течение заданного периода времени, произойдет исключение, и запрос будет сброшен.

Future<Buffer> fut = client
  .request(new RequestOptions()
    .setHost(host)
    .setPort(port)
    .setURI(uri)
    .setIdleTimeout(timeoutMS))
  .compose(request -> request.send().compose(HttpClientResponse::body));

таймаут начинается, когда доступен HttpClientRequest, подразумевая, что соединение получено из пула.

Вы можете установить таймаут подключения, чтобы предотвратить неотзывчивость вашего приложения к занятому пулу подключений к клиенту. Future<HttpClientRequest> срабатывает, если соединение не получено до истечения таймаута.

Опция таймаута подключения не связана с опцией TCP setConnectTimeout. Когда запрос отправляется к пуловому HTTP-клиенту, таймаут относится к времени получения соединения из пула для обработки запроса. Таймаут может сработать, потому что сервер не отвечает вовремя или пул слишком занят, чтобы обработать запрос.

Вы можете настроить оба таймаута, используя setTimeout

Future<Buffer> fut = client
  .request(new RequestOptions()
    .setHost(host)
    .setPort(port)
    .setURI(uri)
    .setTimeout(timeoutMS))
  .compose(request -> request.send().compose(HttpClientResponse::body));

Запись кадров HTTP/2

HTTP/2 — это протокол с кадрами, использующими различные кадры для модели HTTP-запроса/ответа. Протокол позволяет отправлять и получать другие типы кадров.

Для отправки таких кадров можно использовать write на запросе. Вот пример:

int frameType = 40;
int frameStatus = 10;
Buffer payload = Buffer.buffer("some data");

// Sending a frame to the server
request.writeCustomFrame(frameType, frameStatus, payload);

Сброс потока

HTTP/1.x не позволяет чисто сбросить поток запроса или ответа, например, при загрузке клиентом ресурса, уже присутствующего на сервере, серверу необходимо принять весь ответ.

HTTP/2 поддерживает сброс потока в любое время во время запроса/ответа:

request.reset();

По умолчанию отправляется код ошибки NO_ERROR (0), вместо него можно отправить другой код:

request.reset(8);

Спецификация HTTP/2 определяет список кодов ошибок, которые можно использовать.

Обработчики запросов уведомляются о событиях сброса потока с помощью request handler и response handler:

request.exceptionHandler(err -> {
  if (err instanceof StreamResetException) {
    StreamResetException reset = (StreamResetException) err;
    System.out.println("Stream reset " + reset.getCode());
  }
});

Защита от атак HTTP/2 RST-флудом

Сервер HTTP/2 защищен от атак RST-флудом DDOS (CVE-2023-44487): существует верхняя граница количества RST кадров, которые сервер может получить в течение определенного временного интервала. По умолчанию эта граница установлена в 200 для временного интервала в 30 секунды.

Вы можете использовать setHttp2RstFloodMaxRstFramePerWindow и setHttp2RstFloodWindowDuration для изменения этих настроек.

Обработка HTTP-ответов

Вы получаете экземпляр HttpClientResponse в обработчик, который вы указываете в методах запроса или задаёте непосредственно на объекте HttpClientRequest.

Вы можете получить код состояния и сообщение состояния ответа с помощью statusCode и statusMessage.

request
  .send()
  .onComplete(ar2 -> {
    if (ar2.succeeded()) {

      HttpClientResponse response = ar2.result();

      // the status code - e.g. 200 or 404
      System.out.println("Status code is " + response.statusCode());

      // the status message e.g. "OK" or "Not Found".
      System.out.println("Status message is " + response.statusMessage());
    }
  });

Использование ответа как потока

Экземпляр HttpClientResponse также является потоком ReadStream, что означает, что вы можете передавать его любому экземпляру WriteStream.

Заголовки и фрагменты ответа

HTTP-ответы могут содержать заголовки. Используйте headers для получения заголовков.

Возвращаемый объект является MultiMap, так как HTTP-заголовки могут содержать несколько значений для одного ключа.

String contentType = response.headers().get("content-type");
String contentLength = response.headers().get("content-lengh");

В фрагментированных HTTP-ответах также могут быть фрагменты - они отправляются в последнем фрагменте тела ответа.

Используйте trailers для получения фрагментов. Фрагменты также являются MultiMap.

Чтение тела ответа

Обработчик ответа вызывается, когда заголовки ответа были прочитаны из канала.

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

По мере поступления частей тела ответа, handler вызывается с Buffer, представляющим часть тела:

client
  .request(HttpMethod.GET, "some-uri")
  .onComplete(ar1 -> {

    if (ar1.succeeded()) {
      HttpClientRequest request = ar1.result();
      request
        .send()
        .onComplete(ar2 -> {
          HttpClientResponse response = ar2.result();
          response.handler(buffer -> {
            System.out.println("Received a part of the response body: " + buffer);
          });
        });
    }
  });

Если вы знаете, что тело ответа не очень большое и хотите агрегировать его в памяти до обработки, вы можете сделать это сами:

request
  .send()
  .onComplete(ar2 -> {

    if (ar2.succeeded()) {

      HttpClientResponse response = ar2.result();

      // Create an empty buffer
      Buffer totalBuffer = Buffer.buffer();

      response.handler(buffer -> {
        System.out.println("Received a part of the response body: " + buffer.length());

        totalBuffer.appendBuffer(buffer);
      });

      response.endHandler(v -> {
        // Now all the body has been read
        System.out.println("Total response body length is " + totalBuffer.length());
      });
    }
  });

Или вы можете использовать удобную функцию body, которая вызывается со всем телом, когда ответ полностью прочитан:

request
  .send()
  .onComplete(ar1 -> {

    if (ar1.succeeded()) {
      HttpClientResponse response = ar1.result();
      response
        .body()
        .onComplete(ar2 -> {

          if (ar2.succeeded()) {
            Buffer body = ar2.result();
            // Now all the body has been read
            System.out.println("Total response body length is " + body.length());
          }
        });
    }
  });

Обработчик завершения ответа

Обработчик завершения ответа endHandler вызывается, когда всё тело ответа было прочитано или сразу после того, как были прочитаны заголовки и вызван обработчик ответа, если тела нет.

Состав запроса и ответа

Интерфейс клиента очень прост и следует этому шаблону:

request подключение
send или write/end запрос на сервер
обработать начало HttpClientResponse
обработать события ответа

Вы можете использовать методы композиции Vert.x future, чтобы упростить свой код, однако API ориентирован на события, и вам необходимо это понимать, иначе вы можете столкнуться с потенциальными гонками данных (т.е. потеря событий, приводящих к повреждению данных).

Vert.x Web Client — это альтернативный API более высокого уровня (на самом деле он построен поверх этого клиента), который вы можете рассмотреть, если этот клиент слишком низкого уровня для ваших задач

API клиента намеренно не возвращает Future<HttpClientResponse>, потому что установка обработчика завершения на future может быть проблематичной, когда это выполняется вне цикла событий.

Возможная гонка при составлении запроса/ответа

Future<HttpClientResponse> get = client.get("some-uri");

// Assuming we have a client that returns a future response
// assuming this is *not* on the event-loop
// introduce a potential data race for the sake of this example
Thread.sleep(100);

get.onSuccess(response -> {

  // Response events might have happen already
  response
    .body()
    .onComplete(ar -> {

    });
});

Ограничение использования HttpClientRequest внутри вертикли — самое простое решение, так как вертикла гарантирует, что события обрабатываются последовательно, предотвращая гонки.

vertx.deployVerticle(() -> new AbstractVerticle() {
  @Override
  public void start() {

    HttpClient client = vertx.createHttpClient();

    Future<HttpClientRequest> future = client.request(HttpMethod.GET, "some-uri");
  }
}, new DeploymentOptions());

При взаимодействии с клиентом, возможно, вне вертикли, вы можете безопасно выполнять композицию, пока не задерживаете события ответа, например, обрабатывая ответ непосредственно в цикле событий.

Future<JsonObject> future = client
  .request(HttpMethod.GET, "some-uri")
  .compose(request -> request
    .send()
    .compose(response -> {
      // Process the response on the event-loop which guarantees no races
      if (response.statusCode() == 200 &&
          response.getHeader(HttpHeaders.CONTENT_TYPE).equals("application/json")) {
        return response
          .body()
          .map(buffer -> buffer.toJsonObject());
      } else {
        return Future.failedFuture("Incorrect HTTP response");
      }
    }));

// Listen to the composed final json result
future.onSuccess(json -> {
  System.out.println("Received json result " + json);
}).onFailure(err -> {
  System.out.println("Something went wrong " + err.getMessage());
});

Вы также можете защитить тело ответа с помощью ожиданий HTTP-ответов.

Future<JsonObject> future = client
  .request(HttpMethod.GET, "some-uri")
  .compose(request -> request
    .send()
    .expecting(HttpResponseExpectation.SC_OK.and(HttpResponseExpectation.JSON))
    .compose(response -> response
      .body()
      .map(buffer -> buffer.toJsonObject())));
// Listen to the composed final json result
future.onSuccess(json -> {
  System.out.println("Received json result " + json);
}).onFailure(err -> {
  System.out.println("Something went wrong " + err.getMessage());
});

Если вам нужно отложить обработку ответа, вам необходимо pause ответ или использовать pipe, это может потребоваться, когда задействована другая асинхронная операция.

Future<Void> future = client
  .request(HttpMethod.GET, "some-uri")
  .compose(request -> request
    .send()
    .compose(response -> {
      // Process the response on the event-loop which guarantees no races
      if (response.statusCode() == 200) {

        // Create a pipe, this pauses the response
        Pipe<Buffer> pipe = response.pipe();

        // Write the file on the disk
        return fileSystem
          .open("/some/large/file", new OpenOptions().setWrite(true))
          .onFailure(err -> pipe.close())
          .compose(file -> pipe.to(file));
      } else {
        return Future.failedFuture("Incorrect HTTP response");
      }
    }));

Ожидания ответа

Как показано выше, вы должны вручную выполнять проверки целостности после получения ответа.

Вы можете пожертвовать гибкостью ради ясности и краткости, используя ожидания ответа.

Response expectations могут защитить поток управления, когда ответ не соответствует критериям.

HTTP-клиент поставляется с набором готовых к использованию предикатов:

Future<Buffer> fut = client
  .request(options)
  .compose(request -> request
    .send()
    .expecting(HttpResponseExpectation.SC_SUCCESS)
    .compose(response -> response.body()));

Вы также можете создавать собственные предикаты, когда существующие предикаты не подходят для ваших потребностей:

HttpResponseExpectation methodsPredicate =
  resp -> {
    String methods = resp.getHeader("Access-Control-Allow-Methods");
    return methods != null && methods.contains("POST");
  };

// Send pre-flight CORS request
client
  .request(new RequestOptions()
    .setMethod(HttpMethod.OPTIONS)
    .setPort(8080)
    .setHost("myserver.mycompany.com")
    .setURI("/some-uri")
    .putHeader("Origin", "Server-b.com")
    .putHeader("Access-Control-Request-Method", "POST"))
  .compose(request -> request
    .send()
    .expecting(methodsPredicate))
  .onSuccess(res -> {
    // Process the POST request now
  })
  .onFailure(err ->
    System.out.println("Something went wrong " + err.getMessage()));

Предопределённые ожидания

Для удобства HTTP-клиент предоставляет несколько предикатов для распространённых случаев использования.

Для кодов состояния, например, HttpResponseExpectation.SC_SUCCESS, чтобы проверить, что ответ имеет код 2xx, вы также можете создать пользовательский:

client
  .request(options)
  .compose(request -> request
    .send()
    .expecting(HttpResponseExpectation.status(200, 202)))
  .onSuccess(res -> {
    // ....
  });

Для типов содержимого, например, HttpResponseExpectation.JSON, чтобы проверить, что тело ответа содержит данные JSON, вы также можете создать пользовательский:

client
  .request(options)
  .compose(request -> request
    .send()
    .expecting(HttpResponseExpectation.contentType("some/content-type")))
  .onSuccess(res -> {
    // ....
  });

Обратитесь к документации HttpResponseExpectation для получения полного списка предопределённых ожиданий.

Создание пользовательских ошибок

По умолчанию ожидания (включая предопределённые) передают простое сообщение об ошибке. Вы можете настроить класс исключения, изменив преобразователь ошибок:

Expectation<HttpResponseHead> expectation = HttpResponseExpectation.SC_SUCCESS
  .wrappingFailure((resp, err) -> new MyCustomException(resp.statusCode(), err.getMessage()));

Создание исключения на Java может иметь затраты на производительность, когда оно захватывает стек вызовов, поэтому вы можете захотеть создать исключения, которые не захватывают стек вызовов. По умолчанию исключения сообщаются с помощью исключения, которое не захватывает стек вызовов.

Чтение куки из ответа

Вы можете получить список куки из ответа, используя cookies.

В качестве альтернативы, вы можете просто самостоятельно разобрать заголовки Set-Cookie в ответе.

Обработка перенаправлений 30x

Клиент может быть настроен на следование HTTP-перенаправлениям, предоставленным заголовком ответа Location, когда клиент получает:

код состояния 301, 302, 307 или 308 вместе с методом HTTP GET или HEAD
код состояния 303, кроме того, направленный запрос выполняет метод HTTP GET

Вот пример:

client
  .request(HttpMethod.GET, "some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {

      HttpClientRequest request = ar1.result();
      request.setFollowRedirects(true);
      request
        .send()
        .onComplete(ar2 -> {
          if (ar2.succeeded()) {

            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          }
        });
    }
  });

Максимальное количество перенаправлений по умолчанию равно 16 и может быть изменено с помощью setMaxRedirects.

HttpClientAgent client = vertx.createHttpClient(
    new HttpClientOptions()
        .setMaxRedirects(32));

client
  .request(HttpMethod.GET, "some-uri")
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {

      HttpClientRequest request = ar1.result();
      request.setFollowRedirects(true);
      request
        .send()
        .onComplete(ar2 -> {
          if (ar2.succeeded()) {

            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          }
        });
    }
  });

Один размер не подходит всем, и политика перенаправления по умолчанию может не подойти для ваших потребностей.

Политика перенаправления по умолчанию может быть изменена с помощью пользовательской реализации:

HttpClientAgent client = vertx.httpClientBuilder()
  .withRedirectHandler(response -> {

    // Only follow 301 code
    if (response.statusCode() == 301 && response.getHeader("Location") != null) {

      // Compute the redirect URI
      String absoluteURI = resolveURI(response.request().absoluteURI(), response.getHeader("Location"));

      // Create a new ready to use request that the client will use
      return Future.succeededFuture(new RequestOptions().setAbsoluteURI(absoluteURI));
    }

    // We don't redirect
    return null;
  })
  .build();

Политика обрабатывает полученное исходное HttpClientResponse и возвращает либо null, либо Future<HttpClientRequest>.

если возвращается null, исходный ответ обрабатывается
если возвращается будущее, запрос будет отправлен по его успешному завершению
если возвращается будущее, обработчик исключений, установленный для запроса, вызывается при его сбое

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

Большинство исходных настроек запроса будут перенесены в новый запрос:

заголовки запроса, если вы не установили некоторые заголовки
тело запроса, если возвращаемый запрос не использует метод GET
обработчик ответа
обработчик исключений запроса
таймаут запроса

Обработка 100-Continue

Согласно спецификации HTTP 1.1, клиент может установить заголовок Expect: 100-Continue и отправить заголовок запроса перед отправкой остальной части тела запроса.

Затем сервер может ответить промежуточным кодом состояния Status: 100 (Continue), чтобы указать клиенту, что он может отправить остальную часть тела.

Идея в том, что это позволяет серверу авторизовать и принять/отклонить запрос, прежде чем будут отправлены большие объемы данных. Отправка больших объемов данных, если запрос может быть отклонен, является пустой тратой пропускной способности и связывает сервер чтением данных, которые он просто отбросит.

Vert.x позволяет установить continueHandler на объекте запроса клиента

Этот обработчик будет вызван, если сервер отправит ответ Status: 100 (Continue), чтобы указать, что он готов принять остальную часть запроса.

Это используется совместно с `sendHead` для отправки заголовка запроса.

Вот пример:

client.request(HttpMethod.PUT, "some-uri")
  .onSuccess(request -> {
    request.response().onSuccess(response -> {
      System.out.println("Received response with status code " + response.statusCode());
    });

    request.putHeader("Expect", "100-Continue");

    request.continueHandler(v -> {
      // OK to send rest of body
      request.write("Some data");
      request.write("Some more data");
      request.end();
    });

    request.sendHead();
});

На стороне сервера сервер Vert.x можно настроить на автоматическую отправку промежуточных ответов 100 Continue, когда он получает заголовок Expect: 100-Continue.

Это делается путем установки опции setHandle100ContinueAutomatically.

Если вы предпочитаете самостоятельно решать, отправлять ли ответы continue, то эта опция должна быть установлена в false (по умолчанию), после чего вы можете проверить заголовки и вызвать writeContinue, чтобы клиент продолжил отправку тела:

httpServer.requestHandler(request -> {
  if (request.getHeader("Expect").equalsIgnoreCase("100-Continue")) {

    // Send a 100 continue response
    request.response().writeContinue();

    // The client should send the body when it receives the 100 response
    request.bodyHandler(body -> {
      // Do something with body
    });

    request.endHandler(v -> {
      request.response().end();
    });
  }
});

Вы также можете отклонить запрос, отправив код состояния ошибки напрямую: в этом случае тело должно быть проигнорировано или соединение должно быть закрыто (100-Continue — подсказка производительности и не может быть логическим ограничением протокола):

httpServer.requestHandler(request -> {
  if (request.getHeader("Expect").equalsIgnoreCase("100-Continue")) {

    //
    boolean rejectAndClose = true;
    if (rejectAndClose) {

      // Reject with a failure code and close the connection
      // this is probably best with persistent connection
      request.response()
          .setStatusCode(405)
          .putHeader("Connection", "close")
          .end();
    } else {

      // Reject with a failure code and ignore the body
      // this may be appropriate if the body is small
      request.response()
          .setStatusCode(405)
          .end();
    }
  }
});

Создание HTTP-туннелей

HTTP-туннели можно создать с помощью connect:

client.request(HttpMethod.CONNECT, "some-uri")
  .onSuccess(request -> {

    // Connect to the server
    request
      .connect()
      .onComplete(ar -> {
        if (ar.succeeded()) {
          HttpClientResponse response = ar.result();

          if (response.statusCode() != 200) {
            // Connect failed for some reason
          } else {
            // Tunnel created, raw buffers are transmitted on the wire
            NetSocket socket = response.netSocket();
          }
        }
      });
});

Обработчик будет вызван после получения заголовка HTTP-ответа, сокет будет готов для туннелирования и будет отправлять и получать буферы.

connect работает как send, но перенастраивает транспорт для обмена сырыми буферами.

Клиентский push

Push на серверной стороне — это новая функция HTTP/2, которая позволяет отправлять несколько ответов параллельно для одного запроса клиента.

Обработчик push можно установить в запросе, чтобы получать запросы/ответы, которые были отправлены сервером:

client.request(HttpMethod.GET, "/index.html")
  .onSuccess(request -> {

    request
      .response().onComplete(response -> {
        // Process index.html response
      });

    // Set a push handler to be aware of any resource pushed by the server
    request.pushHandler(pushedRequest -> {

      // A resource is pushed for this request
      System.out.println("Server pushed " + pushedRequest.path());

      // Set an handler for the response
      pushedRequest.response().onComplete(pushedResponse -> {
        System.out.println("The response for the pushed request");
      });
    });

    // End the request
    request.end();
});

Если клиент не хочет получать отправленный запрос, он может сбросить поток:

request.pushHandler(pushedRequest -> {
  if (pushedRequest.path().equals("/main.js")) {
    pushedRequest.reset();
  } else {
    // Handle it
  }
});

Если обработчик не задан, любой отправленный поток будет автоматически отменён клиентом сбросом потока (код ошибки 8).

Получение пользовательских кадров HTTP/2

HTTP/2 — это протокол с кадрами, имеющими различные варианты для модели запроса/ответа HTTP. Протокол позволяет отправлять и получать другие типы кадров.

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

response.customFrameHandler(frame -> {

  System.out.println("Received a frame type=" + frame.type() +
      " payload" + frame.payload().toString());
});

Включение сжатия на стороне клиента

Клиент http поддерживает сжатие HTTP по умолчанию.

Это означает, что клиент может сообщить удалённому серверу http о поддержке сжатия и будет способен обрабатывать сжатые тела ответов.

Сервер http свободен либо сжать данные одним из поддерживаемых алгоритмов, либо отправить тело без сжатия. Поэтому это лишь подсказка для сервера Http, которую он может игнорировать.

Чтобы указать серверу http, какие алгоритмы сжатия поддерживает клиент, он включит заголовок Accept-Encoding со значением поддерживаемого алгоритма сжатия. Поддерживается несколько алгоритмов сжатия. В случае Vert.x это приведёт к добавлению следующего заголовка:

Accept-Encoding: gzip, deflate

Затем сервер выберет один из них. Вы можете определить, сжимал ли сервер тело, проверив заголовок Content-Encoding в ответе, отправленном от него.

Если тело ответа было сжато с помощью gzip, оно будет содержать, например, следующий заголовок:

Content-Encoding: gzip

Для включения сжатия установите setDecompressionSupported в параметрах, используемых при создании клиента.

По умолчанию сжатие отключено.

Сбалансированная загрузка на стороне клиента

По умолчанию, когда клиент разрешает имя хоста в список нескольких IP-адресов, клиент использует первый возвращённый IP-адрес.

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

HttpClientAgent client = vertx
  .httpClientBuilder()
  .withLoadBalancer(LoadBalancer.ROUND_ROBIN)
  .build();

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

Большинство политик сбалансированной загрузки достаточно очевидны.

Балансировку на основе хеширования можно получить с помощью политики LoadBalancer.CONSISTENT_HASHING.

HttpClientAgent client = vertx
  .httpClientBuilder()
  .withLoadBalancer(LoadBalancer.ROUND_ROBIN)
  .build();

Политика по умолчанию для согласованного хеширования использует 4 виртуальных узла на сервер и использует случайную политику в отсутствие ключа маршрутизации.

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

LoadBalancer loadBalancer = LoadBalancer.consistentHashing(10, LoadBalancer.POWER_OF_TWO_CHOICES);

Также можно использовать пользовательские политики сбалансированной загрузки.

LoadBalancer loadBalancer = endpoints -> {
  // Returns an endpoint selector for the given endpoints
  // a selector is a stateful view of the provided immutable list of endpoints
  return () -> indexOfEndpoint(endpoints);
};

HttpClientAgent client = vertx
  .httpClientBuilder()
  .withLoadBalancer(loadBalancer)
  .build();

Пулы соединений и keep-alive для HTTP/1.x

Keep-alive позволяет использовать HTTP-соединения для обработки более одного запроса. Это может быть более эффективным использованием соединений, когда вы отправляете несколько запросов одному и тому же серверу.

Для версий HTTP/1.x клиент поддерживает пулы соединений, позволяющие повторно использовать соединения между запросами.

Для работы пула соединений параметр keep alive должен быть true, используя setKeepAlive в опциях при настройке клиента. Значение по умолчанию — true.

Когда keep alive включен. Vert.x добавит заголовок Connection: Keep-Alive к каждому запросу HTTP/1.0. Когда keep alive отключен. Vert.x добавит заголовок Connection: Close к каждому запросу HTTP/1.1, чтобы указать, что соединение будет закрыто после обработки ответа.

Максимальное количество соединений в пуле для каждого сервера настраивается с помощью setHttp1MaxSize

При отправке запроса с включенным пулом соединений Vert.x создаст новое соединение, если создано меньше максимального количества соединений для данного сервера, в противном случае запрос будет помещен в очередь.

Keep-alive соединения будут автоматически закрываться клиентом после таймаута. Таймаут может быть задан сервером с помощью заголовка keep-alive:

keep-alive: timeout=30

Вы можете установить значение таймаута по умолчанию используя setKeepAliveTimeout - любые соединения, не используемые в течение этого таймаута, будут закрыты. Обратите внимание, что значение таймаута указано в секундах, а не миллисекундах.

Пипелинирование HTTP/1.1

Клиент также поддерживает пипелинирование запросов по одному соединению.

Пипелинирование подразумевает отправку следующего запроса по тому же соединению, прежде чем вернется ответ на предыдущий. Пипелинирование не подходит для всех запросов.

Чтобы включить пипелинирование, необходимо включить его используя setPipelining. По умолчанию пипелинирование отключено.

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

Количество пипелинированных запросов по одному соединению ограничено значением setPipeliningLimit. Этот параметр определяет максимальное количество HTTP-запросов, отправленных на сервер, ожидающих ответа. Это ограничение гарантирует справедливое распределение клиентских запросов по соединениям с одним и тем же сервером.

Мультиплексирование HTTP/2

HTTP/2 поддерживает использование одного соединения с сервером. По умолчанию HTTP-клиент использует одно соединение для каждого сервера, все потоки к одному серверу мультиплексируются по одному соединению.

Когда клиенту нужно использовать более одного соединения и использовать пулы соединений, следует использовать setHttp2MaxSize.

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

HttpClient client = vertx.createHttpClient(
  new HttpClientOptions().setHttp2MultiplexingLimit(10),
  new PoolOptions().setHttp2MaxSize(3)
);

Предел мультиплексирования для соединения — это параметр клиента, ограничивающий количество потоков одного соединения. Эффективное значение может быть даже ниже, если сервер установил более низкое ограничение с помощью параметра SETTINGS_MAX_CONCURRENT_STREAMS.

HTTP/2 соединения не закрываются клиентом автоматически. Для их закрытия можно вызвать close или закрыть экземпляр клиента.

Также можно установить таймаут простоя, используя setIdleTimeout. Соединения, не используемые в течение этого таймаута, будут закрыты. Обратите внимание, что значение таймаута указано в секундах, а не миллисекундах.

Непулированные клиентские соединения

Большинство взаимодействий с HTTP выполняются с помощью API запросов/ответов {@code HttpClientAgent}: клиент получает соединение из своего пула соединений для выполнения запроса.

В качестве альтернативы вы можете подключиться напрямую к серверу (обойдя пул соединений) и получить соединение с HTTP-клиентом.

HttpConnectOptions connectOptions = new HttpConnectOptions()
  .setHost("example.com")
  .setPort(80);

Future<HttpClientConnection> fut = client.connect(connectOptions);

HttpClientConnection может создавать HttpClientRequest:

connection
  .request()
  .onSuccess(request -> {
    request.setMethod(HttpMethod.GET);
    request.setURI("/some-uri");
    Future<HttpClientResponse> response = request.send();
  });

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

HTTP-соединения

Этот HttpConnection предоставляет API для работы с событиями HTTP-соединений, жизненным циклом и настройками.

HTTP/2 полностью реализует API HttpConnection.

HTTP/1.x частично реализует API HttpConnection: реализованы только операция закрытия, обработчик закрытия и обработчик исключений. Этот протокол не предоставляет семантику для других операций.

Соединения сервера

Метод connection возвращает соединение запроса на сервере:

HttpConnection connection = request.connection();

Обработчик соединения может быть установлен на сервере, чтобы получать уведомления о любом входящем соединении:

HttpServer server = vertx.createHttpServer(http2Options);

server.connectionHandler(connection -> {
  System.out.println("A client connected");
});

Клиентские соединения

Метод connection возвращает соединение запроса на клиенте:

HttpConnection connection = request.connection();

Обработчик соединения может быть установлен в конструкторе клиента, чтобы получать уведомления о создании соединения:

vertx
  .httpClientBuilder()
  .with(options)
  .withConnectHandler(connection -> {
    System.out.println("Connected to the server");
  })
  .build();

Настройки соединения

Конфигурация HTTP/2 настраивается объектом данных Http2Settings.

Каждая конечная точка должна учитывать настройки, отправленные другой стороной соединения.

При установлении соединения клиент и сервер обмениваются начальными настройками. Начальные настройки настраиваются с помощью setInitialSettings на клиенте и setInitialSettings на сервере.

Настройки могут быть изменены в любое время после установления соединения:

connection.updateSettings(new Http2Settings().setMaxConcurrentStreams(100));

Так как удаленная сторона должна подтвердить получение обновления настроек, можно указать обратный вызов, чтобы получать уведомления о подтверждении:

connection
  .updateSettings(new Http2Settings().setMaxConcurrentStreams(100))
  .onSuccess(v -> System.out.println("The settings update has been acknowledged "));

В свою очередь, remoteSettingsHandler получает уведомление при получении новых настроек удаленной стороны:

connection.remoteSettingsHandler(settings -> {
  System.out.println("Received new settings");
});

это относится только к протоколу HTTP/2

Пинг соединения

Пинг соединения HTTP/2 полезен для определения времени задержки соединения или проверки его работоспособности: ping отправляет кадр PING на удаленную конечную точку:

Buffer data = Buffer.buffer();
for (byte i = 0;i < 8;i++) {
  data.appendByte(i);
}
connection
  .ping(data)
  .onSuccess(pong -> System.out.println("Remote side replied"));

Vert.x автоматически отправит подтверждение при получении кадра PING. Обработчик может быть установлен для получения уведомлений о каждом полученном пинге:

connection.pingHandler(ping -> {
  System.out.println("Got pinged by remote side");
});

Обработчик просто получает уведомление, подтверждение отправляется в любом случае. Такая функция предназначена для реализации протоколов поверх HTTP/2.

это относится только к протоколу HTTP/2

Завершение соединения и уход

Вызов shutdown отправит кадр GOAWAY на удалённую сторону соединения, попросив её остановить создание потоков: клиент перестанет выполнять новые запросы, а сервер — отправлять новые ответы. После отправки кадра GOAWAY соединение ждёт некоторое время (по умолчанию 30 секунд), пока все текущие потоки не будут закрыты, и закрывает соединение:

connection.shutdown();

Метод shutdownHandler уведомляет, когда все потоки были закрыты, но соединение ещё не закрыто.

Можно просто отправить кадр GOAWAY. Основное отличие от завершения заключается в том, что он просто сообщит удалённой стороне соединения об остановке создания новых потоков без планирования закрытия соединения:

connection.goAway(0);

И наоборот, можно получать уведомления о получении GOAWAY:

connection.goAwayHandler(goAway -> {
  System.out.println("Received a go away frame");
});

Метод shutdownHandler будет вызван, когда все текущие потоки будут закрыты, и соединение можно будет закрыть:

connection.goAway(0);
connection.shutdownHandler(v -> {

  // All streams are closed, close the connection
  connection.close();
});

Это также относится к случаю получения GOAWAY.

это относится только к протоколу HTTP/2

Закрытие соединения

Закрытие соединения close закрывает соединение:

закрывает сокет для HTTP/1.x
для HTTP/2 завершение без задержки, кадр GOAWAY по-прежнему будет отправлен перед закрытием соединения.

Метод closeHandler уведомляет о закрытии соединения.

Плавное завершение

Сервер HTTP и клиент поддерживают плавное завершение.

Вы можете завершить server или client.

Вызов shutdown запускает фазу завершения, в ходе которой серверу или клиенту предоставляется возможность выполнить действия по очистке.

Самостоятельный сервер HTTP отвязывается
Удалённый сервер HTTP удаляется из набора принимающих серверов
Клиент HTTP отказывается отправлять новые запросы

После обработки всех запросов в незавершенных соединениях сервер или клиент закрываются.

Кроме того, соединения HTTP/2 отправляют кадр GOAWAY, чтобы сигнализировать удалённому конечной точке, что соединение больше нельзя использовать.

server
  .shutdown()
  .onSuccess(res -> {
    System.out.println("Server is now closed");
  });

Завершение ожидает, пока все сокеты не будут закрыты или не сработает таймаут завершения. Если таймаут срабатывает, все сокеты закрываются принудительно.

Каждое открытое соединение HTTP уведомляется о событии завершения, позволяя выполнить очистку перед фактическим закрытием соединения.

server.connectionHandler(conn -> {
  conn.shutdownHandler(v -> {
    // Perform clean-up
  });
});

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

server
  .shutdown(60, TimeUnit.SECONDS)
  .onSuccess(res -> {
    System.out.println("Server is now closed");
  });

Общий доступ к клиенту

Вы можете использовать HTTP-клиент совместно между несколькими вертиками или экземплярами одного вертика. Такой клиент должен быть создан вне вертика, иначе он будет закрыт при развёртывании вертика, его создавшего.

HttpClientAgent client = vertx.createHttpClient(new HttpClientOptions().setShared(true));
vertx.deployVerticle(() -> new AbstractVerticle() {
  @Override
  public void start() throws Exception {
    // Use the client
  }
}, new DeploymentOptions().setInstances(4));

Вы также можете создать общий HTTP-клиент в каждом вертике:

vertx.deployVerticle(() -> new AbstractVerticle() {
  HttpClientAgent client;
  @Override
  public void start() {
    // Get or create a shared client
    // this actually creates a lease to the client
    // when the verticle is undeployed, the lease will be released automaticaly
    client = vertx.createHttpClient(new HttpClientOptions().setShared(true).setName("my-client"));
  }
}, new DeploymentOptions().setInstances(4));

В первый раз, когда создаётся общий клиент, он создаётся и возвращается. Последующие вызовы будут повторно использовать этот клиент и создавать разрешение на него. Клиент закрывается после того, как все разрешения будут удалены.

По умолчанию, клиент повторно использует текущую очередь событий при необходимости создания TCP-соединения. Таким образом, HTTP-клиент будет случайным образом использовать очереди событий вертиков, используя их безопасным способом.

Вы можете назначить количество очередей событий, которые будет использовать клиент независимо от использующего его клиента

vertx.deployVerticle(() -> new AbstractVerticle() {
  HttpClientAgent client;
  @Override
  public void start() {
    // The client creates and use two event-loops for 4 instances
    client = vertx.createHttpClient(new HttpClientOptions().setShared(true).setName("my-client"), new PoolOptions().setEventLoopSize(2));
  }
}, new DeploymentOptions().setInstances(4));

Общий доступ к серверу

Когда несколько HTTP-серверов слушают на одном порту, vert.x организует обработку запросов с помощью стратегии циклического обхода.

Давайте рассмотрим вертик, создающий HTTP-сервер, такой как:

io.vertx.examples.http.sharing.HttpServerVerticle

vertx.createHttpServer().requestHandler(request -> {
  request.response().end("Hello from server " + this);
}).listen(8080);

Эта служба прослушивает порт 8080.

Итак, когда этот вертик создаётся несколько раз, как с: deploymentOptions.setInstances(2), что происходит? Если оба вертика связаны с одним и тем же портом, вы получите исключение сокета. К счастью, vert.x обрабатывает этот случай за вас. При развертывании другого сервера на том же хосте и порту, что и существующий сервер, он фактически не пытается создать новый сервер, прослушивающий тот же хост/порт. Он связывается с сокетом только один раз. При получении запроса он вызывает обработчики сервера, следуя стратегии циклического обхода.

Теперь давайте представим клиента, такого как:

vertx.setPeriodic(100, (l) -> {
  vertx
    .createHttpClient()
    .request(HttpMethod.GET, 8080, "localhost", "/")
    .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        HttpClientRequest request = ar1.result();
        request
          .send()
          .onComplete(ar2 -> {
            if (ar2.succeeded()) {
              HttpClientResponse resp = ar2.result();
              resp.bodyHandler(body -> {
                System.out.println(body.toString("ISO-8859-1"));
              });
            }
          });
      }
    });
});

Vert.x делегирует запросы одному из серверов последовательно:

Hello from i.v.e.h.s.HttpServerVerticle@1
Hello from i.v.e.h.s.HttpServerVerticle@2
Hello from i.v.e.h.s.HttpServerVerticle@1
Hello from i.v.e.h.s.HttpServerVerticle@2
...

Следовательно, серверы могут масштабироваться по доступным ядрам, в то время как каждый экземпляр вертика Vert.x остаётся строго однопоточным, и вам не нужно применять какие-либо специальные приёмы, такие как написание балансировщиков нагрузки, для масштабирования сервера на многоядерной машине.

Вы можете связаться с общим случайным портом, используя отрицательное значение порта; первый bind выберет порт случайным образом, последующие bind с тем же значением порта будут использовать этот случайный порт.

io.vertx.examples.http.sharing.HttpServerVerticle

vertx.createHttpServer().requestHandler(request -> {
  request.response().end("Hello from server " + this);
}).listen(-1);

Использование HTTPS с Vert.x

Серверы и клиенты Vert.x http могут быть настроены на использование HTTPS точно так же, как и net серверы.

Дополнительную информацию можно найти в разделе настройка net серверов для использования SSL.

SSL также можно включить/выключить для каждого запроса с помощью RequestOptions или при указании схемы с помощью метода setAbsoluteURI.

client
  .request(new RequestOptions()
    .setHost("localhost")
    .setPort(8080)
    .setURI("/")
    .setSsl(true))
  .onComplete(ar1 -> {
    if (ar1.succeeded()) {
      HttpClientRequest request = ar1.result();
      request
        .send()
        .onComplete(ar2 -> {
          if (ar2.succeeded()) {
            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          }
        });
    }
  });

Настройка setSsl действует как значение по умолчанию для клиента.

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

Установив значение на false, вы отключите SSL/TLS, даже если клиент настроен на использование SSL/TLS.
Установив значение на true, вы включите SSL/TLS, даже если клиент настроен на отказ от использования SSL/TLS. Фактическое состояние SSL/TLS клиента (такое как доверие, ключи/сертификаты, шифры, ALPN и т. д.) будет повторно использовано.

Аналогично, схема setAbsoluteURI также переопределяет значение по умолчанию для клиента.

Указание имени сервера (SNI)

Серверы Vert.x http могут быть настроены для использования SNI точно так же, как и {@linkplain io.vertx.core.net net серверы}.

Vert.x http клиент предоставит фактическое имя хоста как имя сервера во время рукопожатия TLS.

Веб-сокеты

Веб-сокеты — веб-технология, которая позволяет создавать двустороннюю сокет-подобную связь между серверами HTTP и клиентами HTTP (обычно браузерами).

Vert.x поддерживает веб-сокеты как на стороне клиента, так и на стороне сервера.

Веб-сокеты на сервере

Существует два способа обработки веб-сокетов на стороне сервера.

Обработчик веб-сокетов

Первый способ включает предоставление webSocketHandler на экземпляре сервера.

При установлении подключения веб-сокета к серверу вызывается обработчик, передавая экземпляр ServerWebSocket.

server.webSocketHandler(webSocket -> {
  System.out.println("Connected!");
});

Обработка рукопожатия веб-сокета сервером

По умолчанию сервер принимает любые входящие веб-сокеты.

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

Вы можете отклонить веб-сокет, вызвав accept или reject.

server.webSocketHandshakeHandler(handshake -> {
  authenticate(handshake.headers(), ar -> {
    if (ar.succeeded()) {
      if (ar.result()) {
        // Terminate the handshake with the status code 101 (Switching Protocol)
        handshake.accept();
      } else {
        // Reject the handshake with 401 (Unauthorized)
        handshake.reject(401);
      }
    } else {
      // Will send a 500 error
      handshake.reject(500);
    }
  });
});

веб-сокет будет автоматически принят после вызова обработчика, если рукопожатие веб-сокета не было установлено

Обновление до веб-сокета

Второй способ обработки веб-сокетов — обработать запрос HTTP Upgrade, отправленный клиентом, и вызвать toWebSocket на запросе сервера.

server.requestHandler(request -> {
  if (request.path().equals("/myapi")) {

    Future<ServerWebSocket> fut = request.toWebSocket();
    fut.onSuccess(ws -> {
      // Do something
    });

  } else {
    // Reject
    request.response().setStatusCode(400).end();
  }
});

Веб-сокет сервера

Экземпляр ServerWebSocket позволяет получить headers, path, query и URI запроса HTTP веб-сокетного рукопожатия.

Веб-сокеты на клиенте

e Vert.x WebSocketClient поддерживает веб-сокеты.

You can connect a WebSocket to a server using one of the `link:../../apidocs/io/vertx/core/http/WebSocketClient.html#connect-int-java.lang.String-java.lang.String-[connect]` operations.

The returned future will be completed with an instance of `link:../../apidocs/io/vertx/core/http/WebSocket.html[WebSocket]` when the connection has been made:

WebSocketClient client = vertx.createWebSocketClient();

client
  .connect(80, "example.com", "/some-uri")
  .onComplete(res -> {
    if (res.succeeded()) {
      WebSocket ws = res.result();
      ws.textMessageHandler(msg -> {
        // Handle msg
      });
      System.out.println("Connected!");
    }
  });

При подключении из потока, не являющегося Vert.x, вы можете создать ClientWebSocket, настроить его обработчики и затем подключиться к серверу:

[source,java]
----
WebSocketClient client = vertx.createWebSocketClient();

client .webSocket() .textMessageHandler(msg → { // Обработать msg }) .connect(80, "example.com", "/some-uri") .onComplete(res → { if (res.succeeded()) { WebSocket ws = res.result(); } }); ----

По умолчанию клиент устанавливает заголовок origin для хоста сервера, например http://www.example.com. Некоторые серверы могут отклонить такой запрос, вы можете настроить клиента, чтобы не устанавливать этот заголовок.

WebSocketConnectOptions options = new WebSocketConnectOptions()
  .setHost(host)
  .setPort(port)
  .setURI(requestUri)
  .setAllowOriginHeader(false);
client
  .connect(options)
  .onComplete(res -> {
    if (res.succeeded()) {
      WebSocket ws = res.result();
      System.out.println("Connected!");
    }
  });

Вы также можете установить другой заголовок:

WebSocketConnectOptions options = new WebSocketConnectOptions()
  .setHost(host)
  .setPort(port)
  .setURI(requestUri)
  .addHeader(HttpHeaders.ORIGIN, origin);
client
  .connect(options)
  .onComplete(res -> {
    if (res.succeeded()) {
      WebSocket ws = res.result();
      System.out.println("Connected!");
    }
  });

более старые версии протокола WebSocket используют sec-websocket-origin вместо

Отправка сообщений в WebSocket

Если вы хотите отправить одно сообщение в WebSocket, вы можете сделать это с помощью writeBinaryMessage или writeTextMessage:

Buffer buffer = Buffer.buffer().appendInt(123).appendFloat(1.23f);
webSocket.writeBinaryMessage(buffer);

// Write a simple text message
String message = "hello";
webSocket.writeTextMessage(message);

Если сообщение WebSocket больше, чем максимальный размер кадра WebSocket, как настроено с помощью setMaxFrameSize, Vert.x разделит его на несколько кадров WebSocket перед отправкой по сети.

Отправка кадров в WebSocket

Сообщение WebSocket может состоять из нескольких кадров. В этом случае первый кадр — это либо бинарный, либо текстовый кадр, за которым следуют ноль или более продолжающих кадров.

Последний кадр в сообщении помечен как конечный.

Чтобы отправить сообщение, состоящее из нескольких кадров, создайте кадры с помощью WebSocketFrame.binaryFrame, WebSocketFrame.textFrame или WebSocketFrame.continuationFrame и отправьте их в WebSocket с помощью writeFrame.

Вот пример для бинарных кадров:

WebSocketFrame frame1 = WebSocketFrame.binaryFrame(buffer1, false);
webSocket.writeFrame(frame1);

WebSocketFrame frame2 = WebSocketFrame.continuationFrame(buffer2, false);
webSocket.writeFrame(frame2);

// Write the final frame
WebSocketFrame frame3 = WebSocketFrame.continuationFrame(buffer2, true);
webSocket.writeFrame(frame3);

В многих случаях вам нужно просто отправить сообщение WebSocket, состоящее из одного конечного кадра, поэтому мы предоставляем несколько сокращенных методов для этого с помощью writeFinalBinaryFrame и writeFinalTextFrame.

Вот пример:

webSocket.writeFinalTextFrame("Geronimo!");

// Send a WebSocket message consisting of a single final binary frame:

Buffer buff = Buffer.buffer().appendInt(12).appendString("foo");

webSocket.writeFinalBinaryFrame(buff);

Чтение кадров из WebSocket

Чтобы прочитать кадры из WebSocket, используйте frameHandler.

Обработчик кадров будет вызываться с экземплярами WebSocketFrame при получении кадра, например:

webSocket.frameHandler(frame -> {
  System.out.println("Received a frame of size!");
});

Закрытие WebSocket

Используйте close для закрытия подключения WebSocket, когда вы закончили с ним.

Использование WebSocket с конвейерами

Экземпляр WebSocket также является ReadStream и WriteStream, поэтому он может быть использован с конвейерами.

При использовании WebSocket как потока записи или чтения, он может быть использован только с подключениями WebSocket, которые используются с бинарными кадрами, не разбитыми на несколько кадров.

Обработчики события автобуса

Каждый WebSocket может зарегистрировать два обработчика в автобусе событий, и когда любые данные получены в этих обработчиках, они записывают данные в себя. Это локальные подписки, недоступные для других узлов кластера.

Это позволяет вам записывать данные в WebSocket, который потенциально находится в совершенно другой вертике, отправляя данные на адрес этого обработчика.

Эта функция по умолчанию отключена, но вы можете включить ее с помощью setRegisterWebSocketWriteHandlers или setRegisterWriteHandlers.

Адреса обработчиков задаются с помощью binaryHandlerID и textHandlerID.

Использование прокси для подключений HTTP/HTTPS

Клиент http поддерживает доступ к URL-адресам http/https через HTTP-прокси (например, Squid) или прокси SOCKS4a или SOCKS5. Протокол CONNECT использует HTTP/1.x, но может подключаться к серверам HTTP/1.x и HTTP/2.

Подключение к h2c (незашифрованным серверам HTTP/2) вероятно не поддерживается http-прокси, так как они поддерживают только HTTP/1.1.

Прокси можно настроить в HttpClientOptions, задав объект ProxyOptions, содержащий тип прокси, имя хоста, порт, а также (необязательно) имя пользователя и пароль.

Вот пример использования HTTP-прокси:

HttpClientOptions options = new HttpClientOptions()
    .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP)
        .setHost("localhost").setPort(3128)
        .setUsername("username").setPassword("secret"));
HttpClientAgent client = vertx.createHttpClient(options);

Когда клиент подключается к http URL, он подключается к прокси-серверу и предоставляет полный URL-адрес в запросе HTTP ("GET http://www.somehost.com/path/file.html HTTP/1.1").

Когда клиент подключается к https URL, он просит прокси создать туннель к удалённому хосту с помощью метода CONNECT.

Для прокси SOCKS5:

HttpClientOptions options = new HttpClientOptions()
    .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5)
        .setHost("localhost").setPort(1080)
        .setUsername("username").setPassword("secret"));
HttpClientAgent client = vertx.createHttpClient(options);

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

Параметры прокси также можно установить для каждого запроса:

client.request(new RequestOptions()
  .setHost("example.com")
  .setProxyOptions(proxyOptions))
  .compose(request -> request
    .send()
    .compose(HttpClientResponse::body))
  .onSuccess(body -> {
    System.out.println("Received response");
  });

кэширование соединений клиента учитывает прокси (включая аутентификацию), следовательно, два запроса к одному хосту через разные прокси не используют одно и то же кэшированное соединение

Вы можете использовать setNonProxyHosts для настройки списка хостов, обходящих прокси. Списки принимают * подстановочный знак для сопоставления доменов:

HttpClientOptions options = new HttpClientOptions()
  .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5)
    .setHost("localhost").setPort(1080)
    .setUsername("username").setPassword("secret"))
  .addNonProxyHost("*.foo.com")
  .addNonProxyHost("localhost");
HttpClientAgent client = vertx.createHttpClient(options);

Обработка других протоколов

Реализация HTTP-прокси поддерживает получение ftp:// url, если прокси поддерживает это.

Когда URI HTTP-запроса содержит полный URL, клиент не будет вычислять полный URL HTTP, а вместо этого будет использовать полный URL, указанный в URI запроса:

HttpClientOptions options = new HttpClientOptions()
    .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP));
HttpClientAgent client = vertx.createHttpClient(options);
client
  .request(HttpMethod.GET, "ftp://ftp.gnu.org/gnu/")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      HttpClientRequest request = ar.result();
      request
        .send()
        .onComplete(ar2 -> {
          if (ar2.succeeded()) {
            HttpClientResponse response = ar2.result();
            System.out.println("Received response with status code " + response.statusCode());
          }
        });
    }
  });

Использование протокола HA PROXY

Протокол HA PROXY предоставляет удобный способ безопасной передачи информации о соединении, такой как адрес клиента, через несколько слоев NAT или TCP-прокси.

Протокол HA PROXY можно включить, установив параметр setUseProxyProtocol и добавив следующую зависимость в вашу classpath:

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-codec-haproxy</artifactId>
  <!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

HttpServerOptions options = new HttpServerOptions()
  .setUseProxyProtocol(true);

HttpServer server = vertx.createHttpServer(options);
server.requestHandler(request -> {
  // Print the actual client address provided by the HA proxy protocol instead of the proxy address
  System.out.println(request.remoteAddress());

  // Print the address of the proxy
  System.out.println(request.localAddress());
});

Автоматическое удаление в вертексах

Если вы создаёте http-серверы и клиенты внутри вертексов, эти серверы и клиенты будут автоматически закрыты при развёртывании вертекса.

Использование API SharedData

Как следует из названия, API SharedData позволяет безопасно обмениваться данными между:

разными частями вашего приложения, или
разными приложениями в одном экземпляре Vert.x, или
разными приложениями в кластере экземпляров Vert.x.

На практике, он предоставляет:

синхронные карты (только локальные)
асинхронные карты
асинхронные блокировки
асинхронные счетчики

Поведение распределённой структуры данных зависит от используемого вами менеджера кластера. Резервирование (репликация) и поведение при сетевом разделении определяются менеджером кластера и его конфигурацией. Обратитесь к документации менеджера кластера, а также к руководству по основам фреймворка.

Локальные карты

Local maps позволяют безопасно обмениваться данными между различными циклами событий (например, различными вертиклами) в одном экземпляре Vert.x.

Они позволяют использовать только определённые типы данных в качестве ключей и значений:

неизменяемые типы (например, строки, булевы значения и т. д.), или
типы, реализующие интерфейс Shareable (буферы, массивы JSON, объекты JSON или ваши собственные обмениваемые объекты).

В последнем случае ключ/значение будут скопированы перед добавлением в карту.

Таким образом, мы можем гарантировать отсутствие совместного доступа к изменяемому состоянию между различными потоками в вашем приложении Vert.x. И вам не придётся беспокоиться о защите этого состояния, синхронизируя доступ к нему.

Для удобства, объекты, реализующие интерфейсы ClusterSerializable или java.io.Serializable, также могут использоваться в качестве ключей и значений. В этом случае ключ/значение будут скопированы перед добавлением в карту путём сериализации/десериализации. Поэтому рекомендуется рассмотреть реализацию интерфейса Shareable для повышения производительности.

Вот пример использования общей локальной карты:

SharedData sharedData = vertx.sharedData();

LocalMap<String, String> map1 = sharedData.getLocalMap("mymap1");

map1.put("foo", "bar"); // Strings are immutable so no need to copy

LocalMap<String, Buffer> map2 = sharedData.getLocalMap("mymap2");

map2.put("eek", Buffer.buffer().appendInt(123)); // This buffer will be copied before adding to map

// Then... in another part of your application:

map1 = sharedData.getLocalMap("mymap1");

String val = map1.get("foo");

map2 = sharedData.getLocalMap("mymap2");

Buffer buff = map2.get("eek");

Асинхронные общие карты

Asynchronous shared maps позволяют помещать данные в карту и получать их локально или с любого другого узла.

Это делает их очень полезными для таких задач, как хранение состояния сеанса в ферме серверов, на которых работает веб-приложение Vert.x.

Они позволяют использовать только определённые типы данных в качестве ключей и значений:

неизменяемые типы (например, строки, булевы значения и т. д.), или
типы, реализующие интерфейс ClusterSerializable (буферы, JSON-массивы, JSON-объекты или ваши собственные сериализуемые в кластере объекты), или
типы, реализующие интерфейс java.io.Serializable.

Получение карты выполняется асинхронно, и результат возвращается в указанный вами обработчик. Вот пример:

SharedData sharedData = vertx.sharedData();

sharedData.
  <String, String>getAsyncMap("mymap")
  .onComplete(res -> {
    if (res.succeeded()) {
      AsyncMap<String, String> map = res.result();
    } else {
      // Something went wrong!
    }
  });

При кластеризации Vert.x данные, помещаемые в карту, доступны локально, а также на любом из других членов кластера.

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

Если ваше приложение не нуждается в том, чтобы данные были доступны каждому другому узлу, вы можете получить локальную карту:

SharedData sharedData = vertx.sharedData();

sharedData.
  <String, String>getLocalAsyncMap("mymap")
  .onComplete(res -> {
    if (res.succeeded()) {
      // Local-only async map
      AsyncMap<String, String> map = res.result();
    } else {
      // Something went wrong!
    }
  });

Помещение данных в карту

Вы помещаете данные в карту с помощью put.

Фактическое размещение выполняется асинхронно, и возвращаемое будущее уведомляется об окончании операции:

map
  .put("foo", "bar")
  .onComplete(resPut -> {
    if (resPut.succeeded()) {
      // Successfully put the value
    } else {
      // Something went wrong!
    }
  });

Получение данных из карты

Вы получаете данные из карты с помощью get.

Фактическое получение выполняется асинхронно, и возвращаемое будущее уведомляется о результате несколько позже:

map
  .get("foo")
  .onComplete(resGet -> {
    if (resGet.succeeded()) {
      // Successfully got the value
      Object val = resGet.result();
    } else {
      // Something went wrong!
    }
  });

Другие операции с картой

Вы также можете удалять записи из асинхронной карты, очищать её и получать размер.

См. API docs для подробного списка операций с картой.

Асинхронные блокировки

Asynchronous locks позволяют получить эксклюзивные блокировки локально или по всему кластеру. Это полезно, когда вы хотите выполнить действие или получить доступ к ресурсу только на одном узле кластера в любой момент времени.

Асинхронные блокировки имеют асинхронный API, в отличие от большинства API блокировок, которые блокируют поток вызова, пока блокировка не будет получена.

Для получения блокировки используйте getLock. Это не будет блокировать, но когда блокировка доступна, возвращаемое будущее завершается экземпляром Lock, сигнализируя, что вы теперь владеете блокировкой.

Пока вы владеете блокировкой, ни один другой вызывающий объект, локально или в кластере, не сможет получить блокировку.

После завершения работы с блокировкой, вызывайте release, чтобы освободить её, и другой вызывающий объект сможет её получить:

SharedData sharedData = vertx.sharedData();

sharedData
  .getLock("mylock")
  .onComplete(res -> {
    if (res.succeeded()) {
      // Got the lock!
      Lock lock = res.result();

      // 5 seconds later we release the lock so someone else can get it

      vertx.setTimer(5000, tid -> lock.release());

    } else {
      // Something went wrong
    }
  });

Вы также можете получить блокировку с таймаутом. Если блокировка не будет получена в течение таймаута, обработчик будет вызван с ошибкой:

SharedData sharedData = vertx.sharedData();

sharedData
  .getLockWithTimeout("mylock", 10000)
  .onComplete(res -> {
    if (res.succeeded()) {
      // Got the lock!
      Lock lock = res.result();

    } else {
      // Failed to get lock
    }
  });

См. API docs для подробного списка операций с блокировками.

В кластерном режиме асинхронные блокировки полагаются на распределённые структуры данных, предоставляемые менеджером кластера. Обратите внимание, что задержка по сравнению с асинхронными операциями с общими блокировками может быть значительно выше в кластерном режиме, чем в локальном.

Если вашему приложению не нужно, чтобы блокировка была доступна каждому узлу, вы можете получить локальную блокировку:

SharedData sharedData = vertx.sharedData();

sharedData
  .getLocalLock("mylock")
  .onComplete(res -> {
    if (res.succeeded()) {
      // Local-only lock
      Lock lock = res.result();

      // 5 seconds later we release the lock so someone else can get it

      vertx.setTimer(5000, tid -> lock.release());

    } else {
      // Something went wrong
    }
  });

Иногда вы используете API блокировки для получения асинхронного результата и применяете шаблон acquire/release вокруг асинхронного вызова. Vert.x предоставляет упрощённый API блокировки для упрощения этого шаблона.

SharedData sharedData = vertx.sharedData();

Future<String> res = sharedData.withLock("mylock", () -> {
  // Obtained the lock!
  Future<String> future = getAsyncString();
  // It will be released upon completion of this future
  return future;
});

Блокировка приобретается перед вызовом поставщика и освобождается при завершении будущего, возвращённого поставщиком.

Асинхронные счётчики

Часто полезно поддерживать атомный счётчик локально или по различным узлам вашего приложения.

Вы можете сделать это с помощью Counter.

Вы получаете экземпляр с помощью getCounter:

SharedData sharedData = vertx.sharedData();

sharedData
  .getCounter("mycounter")
  .onComplete(res -> {
    if (res.succeeded()) {
      Counter counter = res.result();
    } else {
      // Something went wrong!
    }
  });

После получения экземпляра, вы можете получить текущее значение, атомарно инкрементировать, декрементировать и добавлять значение с помощью различных методов.

См. API docs для подробного списка операций со счётчиками.

В кластерном режиме асинхронные счётчики полагаются на распределённые структуры данных, предоставляемые менеджером кластера. Обратите внимание, что задержка по сравнению с асинхронными операциями с общими счётчиками может быть значительно выше в кластерном режиме, чем в локальном.

Если вашему приложению не нужно, чтобы счётчик был доступен каждому другому узлу, вы можете получить локальный счётчик:

SharedData sharedData = vertx.sharedData();

sharedData
  .getLocalCounter("mycounter")
  .onComplete(res -> {
    if (res.succeeded()) {
      // Local-only counter
      Counter counter = res.result();
    } else {
      // Something went wrong!
    }
  });

Использование файловой системы с Vert.x

Объект Vert.x FileSystem предоставляет множество операций для работы с файловой системой.

Один объект файловой системы предоставляется на один экземпляр Vert.x, и вы получаете его с помощью fileSystem.

Для каждой операции предоставляются блокирующая и неблокирующая версии. Неблокирующие версии принимают обработчик, который вызывается при завершении операции или возникновении ошибки.

Вот пример асинхронной копии файла:

FileSystem fs = vertx.fileSystem();

// Copy file from foo.txt to bar.txt
fs.copy("foo.txt", "bar.txt")
  .onComplete(res -> {
    if (res.succeeded()) {
      // Copied ok!
    } else {
      // Something went wrong
    }
  });

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

Вот копия с использованием блокирующего API:

FileSystem fs = vertx.fileSystem();

// Copy file from foo.txt to bar.txt synchronously
fs.copyBlocking("foo.txt", "bar.txt");

Существует множество операций для копирования, перемещения, обрезки, изменения разрешений и многих других операций с файлами. Мы не будем перечислять их все здесь, пожалуйста, обратитесь к API docs для получения полного списка.

Давайте рассмотрим несколько примеров, использующих асинхронные методы:

vertx.fileSystem()
  .readFile("target/classes/readme.txt")
  .onComplete(result -> {
    if (result.succeeded()) {
      System.out.println(result.result());
    } else {
      System.err.println("Oh oh ..." + result.cause());
    }
  });

// Copy a file
vertx.fileSystem()
  .copy("target/classes/readme.txt", "target/classes/readme2.txt")
  .onComplete(result -> {
    if (result.succeeded()) {
      System.out.println("File copied");
    } else {
      System.err.println("Oh oh ..." + result.cause());
    }
  });

// Write a file
vertx.fileSystem()
  .writeFile("target/classes/hello.txt", Buffer.buffer("Hello"))
  .onComplete(result -> {
    if (result.succeeded()) {
      System.out.println("File written");
    } else {
      System.err.println("Oh oh ..." + result.cause());
    }
  });

// Check existence and delete
vertx.fileSystem()
  .exists("target/classes/junk.txt")
  .compose(exist -> {
    if (exist) {
      return vertx.fileSystem().delete("target/classes/junk.txt");
    } else {
      return Future.failedFuture("File does not exist");
    }
  }).onComplete(result -> {
    if (result.succeeded()) {
      System.out.println("File deleted");
    } else {
      System.err.println("Oh oh ... - cannot delete the file: " + result.cause().getMessage());
    }
  });

Асинхронные файлы

Vert.x предоставляет асинхронную абстракцию файлов, которая позволяет управлять файлом в файловой системе.

Вы открываете AsyncFile следующим образом:

OpenOptions options = new OpenOptions();
fileSystem
  .open("myfile.txt", options)
  .onComplete(res -> {
    if (res.succeeded()) {
      AsyncFile file = res.result();
    } else {
      // Something went wrong!
    }
  });

AsyncFile реализует ReadStream и WriteStream, поэтому вы можете перенаправлять файлы в другие объекты потоков, такие как сокеты сети, запросы и ответы HTTP и WebSockets.

Они также позволяют вам читать и записывать непосредственно в них.

Случайный доступ к записи

Чтобы использовать AsyncFile для записи с произвольным доступом, используйте метод write.

Параметры метода:

buffer: буфер для записи.
position: целое число — позиция в файле, где нужно записать буфер. Если позиция больше или равна размеру файла, размер файла будет увеличен для размещения смещения.

Вот пример случайного доступа к записи:

vertx.fileSystem()
  .open("target/classes/hello.txt", new OpenOptions())
  .onComplete(result -> {
    if (result.succeeded()) {
      AsyncFile file = result.result();
      Buffer buff = Buffer.buffer("foo");
      for (int i = 0; i < 5; i++) {
        file
          .write(buff, buff.length() * i)
          .onComplete(ar -> {
            if (ar.succeeded()) {
              System.out.println("Written ok!");
              // etc
            } else {
              System.err.println("Failed to write: " + ar.cause());
            }
          });
      }
    } else {
      System.err.println("Cannot open file " + result.cause());
    }
  });

Случайный доступ к чтению

Чтобы использовать AsyncFile для чтения с произвольным доступом, используйте метод read.

Параметры метода:

buffer: буфер, в который будут считываться данные.
offset: целое число — смещение в буфере, где будут размещены данные, считанные из файла.
position: позиция в файле, откуда нужно считать данные.
length: количество байтов данных для чтения.
handler: обработчик результата.

Вот пример случайного доступа к чтению:

vertx.fileSystem()
  .open("target/classes/les_miserables.txt", new OpenOptions())
  .onComplete(result -> {
    if (result.succeeded()) {
      AsyncFile file = result.result();
      Buffer buff = Buffer.buffer(1000);
      for (int i = 0; i < 10; i++) {
        file
          .read(buff, i * 100, i * 100, 100)
          .onComplete(ar -> {
            if (ar.succeeded()) {
              System.out.println("Read ok!");
            } else {
              System.err.println("Failed to write: " + ar.cause());
            }
          });
      }
    } else {
      System.err.println("Cannot open file " + result.cause());
    }
  });

Параметры открытия

При открытии AsyncFile вы передаете экземпляр OpenOptions. Эти параметры описывают поведение доступа к файлу. Например, вы можете настроить права доступа к файлу с помощью методов setRead, setWrite и setPerms.

Вы также можете настроить поведение, если открываемый файл уже существует, с помощью setCreateNew и setTruncateExisting.

Вы также можете пометить файл для удаления при закрытии или при завершении работы JVM с помощью setDeleteOnClose.

Выгрузка данных в основное хранилище

В OpenOptions вы можете включить/отключить автоматическую синхронизацию содержимого при каждой записи с помощью setDsync. В этом случае вы можете вручную очистить любые записи из кэша ОС, вызвав метод flush.

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

Использование AsyncFile в качестве потока чтения и записи

AsyncFile реализует ReadStream и WriteStream. Вы можете использовать их с pipe для передачи данных в другие потоки чтения и записи. Например, это скопирует содержимое в другой AsyncFile:

final AsyncFile output = vertx.fileSystem().openBlocking("target/classes/plagiary.txt", new OpenOptions());

vertx.fileSystem()
  .open("target/classes/les_miserables.txt", new OpenOptions())
  .compose(file -> file
    .pipeTo(output)
    .eventually(() -> file.close()))
  .onComplete(result -> {
    if (result.succeeded()) {
      System.out.println("Copy done");
    } else {
      System.err.println("Cannot copy file " + result.cause().getMessage());
    }
  });

Вы также можете использовать pipe для записи содержимого файла в ответы HTTP или, более обобщённо, в любые WriteStream.

Доступ к файлам из класса

Когда vert.x не находит файл в файловой системе, он пытается найти его в пути класса. Обратите внимание, что пути к ресурсам в классе никогда не начинаются с /.

Из-за того, что Java не предоставляет асинхронный доступ к ресурсам класса, файл копируется в файловую систему в потоке работ при первом доступе к ресурсу класса и обслуживается оттуда асинхронно. При повторном доступе к тому же ресурсу файл из файловой системы обслуживается непосредственно из неё. Исходное содержимое обслуживается, даже если ресурс класса меняется (например, в системе разработки).

Это поведение кэширования можно настроить с помощью параметра setFileCachingEnabled. Значение по умолчанию этого параметра - true, если системная переменная vertx.disableFileCaching не определена.

Путь, где кэшируются файлы, - /tmp/vertx-cache-UUID по умолчанию и может быть настроен настройкой системной переменной vertx.cacheDirBase. При использовании этой переменной обратите внимание, что она должна ссылаться на префикс директории в процессе читаемо/записываемой локации, например: -Dvertx.cacheDirBase=/tmp/my-vertx-cache (Обратите внимание, что UUID нет).

Каждый процесс vert.x добавит свой собственный UUID, чтобы сохранить кэши независимо от различных приложений, работающих на одном компьютере.

Всю функциональность поиска по пути класса можно отключить по всему системе, установив системную переменную vertx.disableFileCPResolving на true.

Эти системные переменные оцениваются один раз при загрузке класса io.vertx.core.file.FileSystemOptions, поэтому эти переменные должны быть установлены до загрузки этого класса или как системные переменные JVM при запуске.

Если вы хотите отключить поиск по пути класса для определённого приложения, но оставить его включённым по умолчанию по всей системе, вы можете сделать это с помощью параметра setClassPathResolvingEnabled.

Закрытие AsyncFile

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

Сокеты дейтаграмм (UDP)

Использование протокола User Datagram Protocol (UDP) с Vert.x — проще простого.

UDP — это протокол транспортного уровня без установления соединения, что означает отсутствие постоянного соединения с удалённым узлом.

Вместо этого вы можете отправлять и получать пакеты, при этом адрес удалённого узла содержится в каждом из них.

Помимо этого, UDP не такой надёжный, как TCP, что означает отсутствие гарантии получения отправленного пакета данных конечным получателем.

Единственная гарантия — это полное получение пакета или его полное отсутствие.

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

Однако будьте внимательны, даже если размер пакета меньше размера MTU, он всё равно может быть утерян.

Размер, при котором пакет будет утерян, зависит от операционной системы и т. д. Поэтому эмпирическое правило — отправлять небольшие пакеты.

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

Преимуществами являются меньшие накладные расходы по сравнению с TCP, с которыми могут справиться NetServer и NetClient (см. выше).

Создание сокета дейтаграмм

Чтобы использовать UDP, сначала необходимо создать DatagramSocket. Здесь не имеет значения, хотите ли вы только отправлять данные или отправлять и получать.

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());

Возвращённый DatagramSocket не будет привязан к определённому порту. Это не проблема, если вы хотите только отправлять данные (например, клиент), но подробнее об этом в следующем разделе.

Отправка пакетов дейтаграмм

Как уже упоминалось, протокол User Datagram Protocol (UDP) отправляет данные в пакетах удалённым узлам, но не подключён к ним постоянно.

Это означает, что каждый пакет может быть отправлен другому удалённому узлу.

Отправка пакетов выполняется так же просто, как показано здесь:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
Buffer buffer = Buffer.buffer("content");
// Send a Buffer
socket
  .send(buffer, 1234, "10.0.0.1")
  .onComplete(asyncResult -> System.out.println("Send succeeded? " + asyncResult.succeeded()));
// Send a String
socket
  .send("A string used as content", 1234, "10.0.0.1")
  .onComplete(asyncResult -> System.out.println("Send succeeded? " + asyncResult.succeeded()));

Приём пакетов дейтаграмм

Если вы хотите получать пакеты, вам нужно привязать DatagramSocket, вызвав listen(…) на нём.

Таким образом, вы сможете получать DatagramPacket`s that were sent to the address and port on which the `DatagramSocket прослушивают.

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

У DatagramPacket следующие методы:

sender: InetSocketAddress, представляющий отправителя пакета
data: Буфер, содержащий полученные данные.

Таким образом, для прослушивания по определённому адресу и порту вы можете сделать что-то вроде этого:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
socket
  .handler(packet -> {
    // Do something with the packet
  })
  .listen(1234, "0.0.0.0")
  .onComplete(asyncResult -> System.out.println("Send succeeded? " + asyncResult.succeeded()));
;

Обратите внимание, что даже если {code AsyncResult} успешен, это только означает, что данные могут быть записаны в сетевой стеке, но не гарантирует, что они вообще дойдут или дойдут до удалённого узла.

Если вам нужна такая гарантия, используйте TCP с логикой рукопожатия, построенной поверх него.

Мультипоток

Отправка пакетов мультипотока

Мультипоток позволяет нескольким сокетам получать одни и те же пакеты. Это работает, объединяя сокеты в одну и ту же группу мультипотока, в которую затем можно отправлять пакеты.

В следующем разделе мы рассмотрим, как присоединиться к группе мультипотока и получать пакеты.

Отправка пакетов мультипотока ничем не отличается от отправки обычных пакетов Datagram. Разница заключается в том, что вы передаете адрес группы мультипотока в метод отправки.

Это показано здесь:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
Buffer buffer = Buffer.buffer("content");
// Send a Buffer to a multicast address
socket
  .send(buffer, 1234, "230.0.0.1")
  .onComplete(asyncResult -> System.out.println("Send succeeded? " + asyncResult.succeeded()));

Все сокеты, присоединившиеся к группе мультипотока 230.0.0.1, получат пакет.

Получение пакетов мультипотока

Если вы хотите получать пакеты для определенной группы мультипотока, вам нужно связать DatagramSocket, вызвав listen(…) для присоединения к группе мультипотока.

Таким образом, вы будете получать DatagramPackets, которые были отправлены по адресу и порту, на котором слушает DatagramSocket, а также тем, которые были отправлены в группу мультипотока.

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

У DatagramPacket есть следующие методы:

sender(): InetSocketAddress, представляющий отправителя пакета
data(): Буфер, содержащий полученные данные.

Таким образом, чтобы прослушивать по определенному адресу и порту и также получать пакеты для группы мультипотока 230.0.0.1, вам нужно сделать что-то вроде показанного здесь:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
socket
  .handler(packet -> {
    // Do something with the packet
  })
  .listen(1234, "0.0.0.0")
  .compose(v -> socket.listenMulticastGroup("230.0.0.1")) // join the multicast group
  .onComplete(asyncResult -> System.out.println("Listen succeeded? " + asyncResult.succeeded()));

Отключение прослушивания / выход из группы мультипотока

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

В таких ситуациях вы можете сначала начать прослушивать их, а затем позже отключиться.

Это показано здесь:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
socket
  .handler(packet -> {
    // Do something with the packet
  })
  .listen(1234, "0.0.0.0")
  .compose(v -> socket.listenMulticastGroup("230.0.0.1")) // join the multicast group
  .onComplete(asyncResult -> {
    if (asyncResult.succeeded()) {
      // will now receive packets for group

      // do some work

      socket.unlistenMulticastGroup("230.0.0.1").onComplete(asyncResult2 -> {
        System.out.println("Unlisten succeeded? " + asyncResult2.succeeded());
      });
    } else {
      System.out.println("Listen failed" + asyncResult.cause());
    }
  });

Блокировка мультипотока

Помимо отключения прослушивания адреса мультипотока, можно также просто заблокировать мультипоток для определенного отправителя.

Обратите внимание, что это работает только на некоторых операционных системах и версиях ядра. Поэтому, пожалуйста, проверьте документацию операционной системы, если это поддерживается.

Это экспертная функция.

Чтобы заблокировать мультипоток от определенного адреса, вы можете вызвать blockMulticastGroup(…) на DatagramSocket, как показано здесь:

DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());

// Some code

// This would block packets which are send from 10.0.0.2
socket
  .blockMulticastGroup("230.0.0.1", "10.0.0.2")
  .onComplete(asyncResult -> System.out.println("block succeeded? " + asyncResult.succeeded()));

Свойства DatagramSocket

При создании объекта DatagramSocket можно установить несколько свойств, чтобы изменить его поведение с помощью объекта DatagramSocketOptions. Они перечислены здесь:

setSendBufferSize Устанавливает размер буфера отправки в байтах.
setReceiveBufferSize Устанавливает размер буфера приема TCP в байтах.
setReuseAddress Если значение true, то адреса в состоянии TIME_WAIT могут быть повторно использованы после их закрытия.
setTrafficClass
setBroadcast Устанавливает или сбрасывает опцию сокета SO_BROADCAST. При установке этой опции пакеты Datagram (UDP) могут отправляться на адрес локального широковещания.
setMulticastNetworkInterface Устанавливает или сбрасывает опцию сокета IP_MULTICAST_LOOP. При установке этой опции пакеты мультивещания также будут приниматься на локальный интерфейс.
setMulticastTimeToLive Устанавливает опцию сокета IP_MULTICAST_TTL. TTL означает «Время жизни», но в данном контексте он определяет количество IP-хопов, через которые пакет разрешено проходить, в частности для трафика мультивещания. Каждый маршрутизатор или шлюз, передающий пакет, уменьшает значение TTL. Если TTL уменьшается до 0 маршрутизатором, пакет не будет пересылаться.

Локальный адрес DatagramSocket

Вы можете узнать локальный адрес сокета (т.е. адрес этой стороны UDP-сокета), вызвав localAddress. Это вернет InetSocketAddress только если вы привязали DatagramSocket к listen(…) ранее, в противном случае вернется null.

Закрытие DatagramSocket

Вы можете закрыть сокет, вызвав метод close. Это закроет сокет и освободит все ресурсы.

Клиент DNS

Часто вы сталкиваетесь с ситуациями, когда вам нужно получить информацию DNS асинхронно. К сожалению, это невозможно с помощью API, поставляемого вместе с Java Virtual Machine. Поэтому Vert.x предлагает собственный API для разрешения DNS, который полностью асинхронный.

Чтобы получить экземпляр DnsClient, создайте новый через экземпляр Vertx.

DnsClient client = vertx.createDnsClient(53, "10.0.0.1");

Вы также можете создать клиента с параметрами и настроить тайм-аут запроса.

DnsClient client = vertx.createDnsClient(new DnsClientOptions()
  .setPort(53)
  .setHost("10.0.0.1")
  .setQueryTimeout(10000)
);

Создание клиента без аргументов или пропуск адреса сервера будет использовать адрес сервера, используемого внутри для асинхронного разрешения адресов.

DnsClient client1 = vertx.createDnsClient();

// Just the same but with a different query timeout
DnsClient client2 = vertx.createDnsClient(new DnsClientOptions().setQueryTimeout(10000));

Клиент использует единственную очередь событий для запросов, его безопасно использовать из любой потоковой нити, включая нити, не относящиеся к Vert.x.

lookup

Попробуйте найти запись A (ipv4) или AAAA (ipv6) для данного имени. Первая из них, которая будет возвращена, будет использована, поэтому она ведет себя так же, как вы привыкли при использовании «nslookup» в вашей операционной системе.

Чтобы найти запись A/AAAA для «vertx.io», вы обычно используете её так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .lookup("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println(ar.result());
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

lookup4

Попробуйте найти запись A (ipv4) для данного имени. Первая из них, которая будет возвращена, будет использована, поэтому она ведет себя так же, как вы привыкли при использовании «nslookup» в вашей операционной системе.

Чтобы найти запись A для «vertx.io», вы обычно используете её так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .lookup4("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println(ar.result());
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

lookup6

Попробуйте найти запись AAAA (ipv6) для данного имени. Первая из них, которая будет возвращена, будет использована, поэтому она ведет себя так же, как вы привыкли при использовании «nslookup» в вашей операционной системе.

Чтобы найти запись A для «vertx.io», вы обычно используете её так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .lookup6("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println(ar.result());
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveA

Попробуйте разрешить все записи A (ipv4) для данного имени. Это довольно похоже на использование «dig» в операционных системах типа Unix.

Чтобы найти все записи A для «vertx.io», вы обычно делаете так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveA("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<String> records = ar.result();
      for (String record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveAAAA

Попробуйте разрешить все записи AAAA (ipv6) для данного имени. Это довольно похоже на использование «dig» в операционных системах типа Unix.

Чтобы найти все записи AAAA для «vertx.io», вы обычно делаете так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveAAAA("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<String> records = ar.result();
      for (String record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveCNAME

Попробуйте разрешить все записи CNAME для данного имени. Это довольно похоже на использование «dig» в операционных системах типа Unix.

Чтобы найти все записи CNAME для «vertx.io», вы обычно делаете так:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveCNAME("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<String> records = ar.result();
      for (String record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveMX

Попытка разрешить все записи MX для данного имени. Записи MX используются для определения почтового сервера, принимающего электронную почту для данного домена.

Для поиска всех записей MX для "vertx.io" обычно выполняется:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveMX("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<MxRecord> records = ar.result();
      for (MxRecord record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

Обратите внимание, что список будет содержать MxRecord, отсортированные по приоритету, что означает, что записи MX с меньшим приоритетом будут в списке первыми.

MxRecord позволяет получить доступ к приоритету и имени записи MX, предлагая для этого методы, такие как:

record.priority();
record.name();

resolveTXT

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

Для разрешения всех записей TXT для "vertx.io" можно использовать что-то подобное:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveTXT("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<String> records = ar.result();
      for (String record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveNS

Попытка разрешить все записи NS для данного имени. Записи NS указывают, какой DNS-сервер хранит DNS-информацию для данного домена.

Для разрешения всех записей NS для "vertx.io" можно использовать что-то подобное:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveNS("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<String> records = ar.result();
      for (String record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

resolveSRV

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

Для поиска всех записей SRV для "vertx.io" обычно выполняется:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolveSRV("vertx.io")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      List<SrvRecord> records = ar.result();
      for (SrvRecord record : records) {
        System.out.println(record);
      }
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

Обратите внимание, что список будет содержать SrvRecords, отсортированные по приоритету, что означает, что SrvRecords с меньшим приоритетом будут в списке первыми.

SrvRecord позволяет получить доступ ко всей информации, содержащейся в самой записи SRV:

record.priority();
record.name();
record.weight();
record.port();
record.protocol();
record.service();
record.target();

Для точных деталей обратитесь к документации API.

resolvePTR

Попытка разрешить запись PTR для данного имени. Запись PTR сопоставляет IP-адрес с именем.

Для разрешения записи PTR для IP-адреса 10.0.0.1 используется запись PTR "1.0.0.10.in-addr.arpa"

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .resolvePTR("1.0.0.10.in-addr.arpa")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      String record = ar.result();
      System.out.println(record);
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

reverseLookup

Попытка обратного поиска IP-адреса. Это в основном то же самое, что разрешение записи PTR, но позволяет просто передать IP-адрес, а не строку запроса PTR.

Для обратного поиска IP-адреса 10.0.0.1 выполните что-то подобное:

DnsClient client = vertx.createDnsClient(53, "9.9.9.9");
client
  .reverseLookup("10.0.0.1")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      String record = ar.result();
      System.out.println(record);
    } else {
      System.out.println("Failed to resolve entry" + ar.cause());
    }
  });

Обработка ошибок

Как вы видели в предыдущих разделах, DnsClient позволяет вам передать обработчик, который будет уведомлён с AsyncResult после завершения запроса. В случае ошибки он будет уведомлён с DnsException, который будет содержать DnsResponseCode, указывающий причину неудачи разрешения. Этот код ответа DNS можно использовать для более подробного анализа причины.

Возможные коды ответа DNS:

NOERROR Запрашиваемый ресурс не найден.
FORMERROR Ошибка формата.
SERVFAIL Ошибка сервера.
NXDOMAIN Ошибка имени.
NOTIMPL Не реализовано на DNS-сервере.
REFUSED DNS-сервер отклонил запрос.
YXDOMAIN Доменное имя не должно существовать.
YXRRSET Ресурсный запис должен не существовать.
NXRRSET RRSET не существует.
NOTZONE Имя не в зоне.
BADVERS Неправильный механизм расширения для версии.
BADSIG Неправимая подпись.
BADKEY Неправитый ключ.
BADTIME Неправильное отметка времени.

Все эти ошибки «генерируются» самим DNS-сервером.

Вы можете получить DnsResponseCode из DnsException следующим образом:

DnsClient client = vertx.createDnsClient(53, "10.0.0.1");
client
  .lookup("nonexisting.vert.xio")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      String record = ar.result();
      System.out.println(record);
    } else {
      Throwable cause = ar.cause();
      if (cause instanceof DnsException) {
        DnsException exception = (DnsException) cause;
        DnsResponseCode code = exception.code();
        // ...
      } else {
        System.out.println("Failed to resolve entry" + ar.cause());
      }
    }
  });

Виртуальные потоки Vert.x

Используйте виртуальные потоки для написания кода Vert.x, который выглядит синхронным.

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

Введение

Неблокирующая природа Vert.x приводит к асинхронным API. Асинхронные API могут принимать различные формы, включая стиль обратного вызова, обещания и реактивные расширения.

В некоторых случаях программирование с использованием асинхронных API может быть сложнее, чем использование прямого синхронного стиля, особенно если у вас есть несколько операций, которые вы хотите выполнить последовательно. Кроме того, распространение ошибок часто сложнее при использовании асинхронных API.

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

Это достигается с помощью виртуальных потоков Java 21 virtual threads. Виртуальные потоки являются очень лёгкими потоками, которые не соответствуют базовым потокам ядра. Когда они блокируются, они не блокируют поток ядра.

Использование виртуальных потоков

Вы можете развернуть виртуальные нити вертиклов.

Виртуальный поток вертиклов способен ожидать результатов Vert.x, получая их синхронно.

Когда вертикл ожидает результат, вертикл по-прежнему может обрабатывать события, как вертикл цикла событий.

AbstractVerticle verticle = new AbstractVerticle() {
  @Override
  public void start() {
    HttpClient client = vertx.createHttpClient();
    HttpClientRequest req = client.request(
      HttpMethod.GET,
      8080,
      "localhost",
      "/").await();
    HttpClientResponse resp = req.send().await();
    int status = resp.statusCode();
    Buffer body = resp.body().await();
  }
};

// Run the verticle a on virtual thread
vertx.deployVerticle(verticle, new DeploymentOptions().setThreadingModel(ThreadingModel.VIRTUAL_THREAD));

Использование виртуальных потоков требует Java 21 или выше.

Блокировка внутри вертиклов с виртуальными потоками

Вы можете использовать await для приостановки текущего виртуального потока до тех пор, пока ожидаемый результат не станет доступен.

Виртуальный поток фактически блокируется, но приложение всё ещё может обрабатывать события.

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

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

Как и любой вертикл, одновременно выполняется не более одной задачи.

Вы можете ожидать результата Vert.x Future

Buffer body = response.body().await();

или на JDK CompletionStage

Buffer body = Future.fromCompletionStage(completionStage).await();

Вы также можете преобразовать Vert.x ReadStream в Java-потоковый поток:

server.requestHandler(request -> {
  Stream<Buffer> blockingStream = request.blockingStream();
  HttpServerResponse response = request.response();
  response.setChunked(true);
  blockingStream
    .map(buff -> "" + buff.length())
    .forEach(size -> response.write(size));
  response.end();
});

Видимость полей

Вертикл с виртуальным потоком может безопасно взаимодействовать с полями до вызова await. Однако, если вы читаете поле до вызова await и повторно используете значение после вызова, следует помнить, что это значение может измениться.

int value = counter;
value += getRemoteValue().await();
// the counter value might have changed
counter = value;

Следует читать/записывать поля перед вызовом await, чтобы избежать этого.

counter += getRemoteValue().await();

Это поведение идентично вертиклу цикла событий.

Ожидание нескольких результатов

Если нужно ожидать нескольких результатов, вы можете использовать Vert.x CompositeFuture:

Future<String> f1 = getRemoteString();
Future<Integer> f2 = getRemoteValue();
CompositeFuture res = Future.all(f1, f2).await();
String v1 = res.resultAt(0);
Integer v2 = res.resultAt(1);

Блокировка без ожидания

Если ваше приложение блокируется без использования await, например, используя ReentrantLock#lock, планировщик Vert.x не осведомлён об этом и не может планировать события в вертикле: он ведет себя как рабочий вертикл, но использует виртуальные потоки.

Этот случай не поощряется, но не запрещён; однако вертикл должен быть развернут с несколькими экземплярами для обеспечения необходимой конкуреции, как и рабочий вертикл.

Поддержка ThreadLocal

ThreadLocal надежны только в рамках выполнения задачи контекста.

ThreadLocal<String> local = new ThreadLocal();
local.set(userId);
HttpClientRequest req = client.request(HttpMethod.GET, 8080, "localhost", "/").await();
HttpClientResponse resp = req.send().await();

Потоки

Существует несколько объектов в Vert.x, которые позволяют читать и записывать данные.

В Vert.x вызовы записи возвращают значения немедленно, а сами записи очередируются внутри.

Несложно понять, что если вы записываете данные в объект быстрее, чем он может фактически записать их в свой базовый ресурс, то очередь записи может неограниченно расти, что в конечном итоге приведёт к исчерпанию памяти.

Для решения этой проблемы некоторые объекты в API Vert.x предоставляют простую возможность управления потоком (обратная загрузка).

Любой объект, который осознаёт управление потоком и в который можно записать, реализует WriteStream, а любой объект, который осознаёт управление потоком и из которого можно читать, считается реализующим ReadStream.

Давайте рассмотрим пример, где мы хотим читать данные из ReadStream, а затем записать их в WriteStream.

Очень простым примером было бы чтение данных из NetSocket и последующая запись обратно в тот же NetSocket, так как NetSocket реализует как ReadStream, так и WriteStream. Обратите внимание, что это работает между любыми объектами, совместимыми с ReadStream и WriteStream, включая HTTP-запросы, HTTP-ответы, асинхронные операции с файлами ввода-вывода, WebSockets и т.д.

Примитивный способ сделать это состоял бы в том, чтобы сразу взять данные, прочитанные из объекта, и немедленно записать их в NetSocket:

NetServer server = vertx.createNetServer(
    new NetServerOptions().setPort(1234).setHost("localhost")
);
server.connectHandler(sock -> {
  sock.handler(buffer -> {
    // Write the data straight back
    sock.write(buffer);
  });
}).listen();

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

Поскольку NetSocket реализует WriteStream, мы можем проверить, полна ли WriteStream, прежде чем записывать в неё:

NetServer server = vertx.createNetServer(
    new NetServerOptions().setPort(1234).setHost("localhost")
);
server.connectHandler(sock -> {
  sock.handler(buffer -> {
    if (!sock.writeQueueFull()) {
      sock.write(buffer);
    }
  });

}).listen();

В этом примере оперативная память не будет исчерпана, но мы потеряем данные, если очередь записи заполнится. На самом деле, нам нужно приостановить NetSocket, когда очередь записи заполнится:

NetServer server = vertx.createNetServer(
    new NetServerOptions().setPort(1234).setHost("localhost")
);
server.connectHandler(sock -> {
  sock.handler(buffer -> {
    sock.write(buffer);
    if (sock.writeQueueFull()) {
      sock.pause();
    }
  });
}).listen();

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

NetServer server = vertx.createNetServer(
    new NetServerOptions().setPort(1234).setHost("localhost")
);
server.connectHandler(sock -> {
  sock.handler(buffer -> {
    sock.write(buffer);
    if (sock.writeQueueFull()) {
      sock.pause();
      sock.drainHandler(done -> {
        sock.resume();
      });
    }
  });
}).listen();

И вот оно. Обработчик события drainHandler будет вызван, когда очередь записи готова принять больше данных, это возобновит NetSocket, что позволит прочитать больше данных.

Это довольно распространённая необходимость при написании приложений Vert.x, поэтому мы добавили метод pipeTo, который выполняет всю эту работу за вас. Просто передайте WriteStream и используйте его:

NetServer server = vertx.createNetServer(
  new NetServerOptions().setPort(1234).setHost("localhost")
);
server.connectHandler(sock -> {
  sock.pipeTo(sock);
}).listen();

Это делает ровно то же самое, что и более подробный пример, плюс он обрабатывает сбои и завершение потока: целевой WriteStream завершается, когда передача завершается успешно или с ошибкой.

Вы можете получить уведомление о завершении операции:

server.connectHandler(sock -> {

  // Pipe the socket providing an handler to be notified of the result
  sock
    .pipeTo(sock)
    .onComplete(ar -> {
      if (ar.succeeded()) {
        System.out.println("Pipe succeeded");
      } else {
        System.out.println("Pipe failed");
      }
    });
}).listen();

Когда вы имеете дело с асинхронным назначением, вы можете создать экземпляр Pipe, который приостанавливает источник и возобновляет его, когда источник направляется в назначение:

server.connectHandler(sock -> {

  // Create a pipe to use asynchronously
  Pipe<Buffer> pipe = sock.pipe();

  // Open a destination file
  fs.open("/path/to/file", new OpenOptions())
    .onComplete(ar -> {
      if (ar.succeeded()) {
        AsyncFile file = ar.result();

        // Pipe the socket to the file and close the file at the end
        pipe.to(file);
      } else {
        sock.close();
      }
  });
}).listen();

Если вам нужно прервать передачу, вам нужно её закрыть:

vertx.createHttpServer()
  .requestHandler(request -> {

    // Create a pipe that to use asynchronously
    Pipe<Buffer> pipe = request.pipe();

    // Open a destination file
    fs.open("/path/to/file", new OpenOptions())
      .onComplete(ar -> {
        if (ar.succeeded()) {
          AsyncFile file = ar.result();

          // Pipe the socket to the file and close the file at the end
          pipe.to(file);
        } else {
          // Close the pipe and resume the request, the body buffers will be discarded
          pipe.close();

          // Send an error response
          request.response().setStatusCode(500).end();
        }
      });
  }).listen(8080);

Когда труба закрыта, обработчики потоков сбрасываются, и ReadStream возобновляется.

Как видно выше, по умолчанию пункт назначения всегда завершается при завершении потока. Вы можете контролировать это поведение на объекте трубы:

endOnFailure управляет поведением при возникновении ошибки
endOnSuccess управляет поведением при завершении потока чтения
endOnComplete управляет поведением во всех случаях

Вот короткий пример:

src.pipe()
  .endOnSuccess(false)
  .to(dst)
  .onComplete(rs -> {
    // Append some text and close the file
    dst.end(Buffer.buffer("done"));
  });

Теперь давайте подробнее рассмотрим методы ReadStream и WriteStream:

Поток чтения (ReadStream)

ReadStream реализован в HttpClientResponse, DatagramSocket, HttpClientRequest, HttpServerFileUpload, HttpServerRequest, MessageConsumer, NetSocket, WebSocket и AsyncFile.

handler: устанавливает обработчик, который будет получать элементы из потока чтения.
pause: приостанавливает поток. При приостановке элементы не будут поступать в обработчик.
fetch: извлекает указанное количество элементов из потока. Обработчик будет вызываться, если появятся какие-либо элементы. Вызовы `fetch` накапливаются.
resume: возобновляет поток. Обработчик будет вызываться, если появятся какие-либо элементы. Возобновление равносильно вызову `fetch` для Long.MAX_VALUE элементов.
exceptionHandler: вызывается при возникновении исключения в потоке чтения.
endHandler: вызывается при достижении конца потока. Это может произойти при достижении конца файла (EOF), если поток чтения представляет файл, при достижении конца запроса, если это HTTP-запрос, или при закрытии соединения, если это TCP-сокет.

Поток чтения находится в режиме потока или вызова

по умолчанию поток находится в режиме потока
в режиме потока элементы передаются в обработчик
в режиме вызова в обработчик передается только запрошенное количество элементов

pause, resume и fetch изменяют режим

resume() устанавливает режим потока
pause() устанавливает режим вызова и сбрасывает запрос до 0
fetch(long) запрашивает определённое количество элементов и добавляет его к фактическому запросу

WriteStream

WriteStream реализуется с помощью HttpClientRequest, HttpServerResponse WebSocket, NetSocket и AsyncFile.

Функции:

write: записывает объект в WriteStream. Этот метод никогда не блокируется. Записи помещаются в очередь и асинхронно записываются в базовом ресурсе.
setWriteQueueMaxSize: устанавливает количество объектов, при котором очередь записи считается полной, и метод writeQueueFull возвращает true. Обратите внимание, что когда очередь записи считается полной, вызов метода write все равно примет и поместит данные в очередь. Фактическое количество зависит от реализации потока, для Buffer размер представляет фактическое количество записанных байтов, а не количество буферов.
writeQueueFull: возвращает true, если очередь записи считается полной.
exceptionHandler: вызывается, если в WriteStream произошла ошибка.
drainHandler: обработчик вызывается, если очередь WriteStream больше не считается полной.

Сведение потоков

Коллекторы Java могут свести ReadStream к результату аналогичным способом, как это делает java.util.Stream, но асинхронно.

Future<Long> result = stream.collect(Collectors.counting());

result.onSuccess(count -> System.out.println("Stream emitted " + count + " elements"));

Обратите внимание, что collect переопределяет любой ранее установленный обработчик в потоке.

Парсер записей

Парсер записей позволяет легко анализировать протоколы, которые разделяются последовательностью байтов или записями фиксированного размера. Он преобразует последовательность входных буферов в последовательность буферов, структурированных в соответствии с конфигурацией (либо фиксированный размер, либо записи, разделенные разделителями).

Например, если у вас есть простой текстовый ASCII-протокол, ограниченный символом '\n', и входные данные следующие:

buffer1:HELLO\nHOW ARE Y
buffer2:OU?\nI AM
buffer3: DOING OK
buffer4:\n

Парсер записей создаст

buffer1:HELLO
buffer2:HOW ARE YOU?
buffer3:I AM DOING OK

Посмотрим на соответствующий код:

final RecordParser parser = RecordParser.newDelimited("\n", h -> {
  System.out.println(h.toString());
});

parser.handle(Buffer.buffer("HELLO\nHOW ARE Y"));
parser.handle(Buffer.buffer("OU?\nI AM"));
parser.handle(Buffer.buffer("DOING OK"));
parser.handle(Buffer.buffer("\n"));

Также можно получить куски фиксированного размера следующим образом:

RecordParser.newFixed(4, h -> {
  System.out.println(h.toString());
});

Для получения дополнительной информации см. класс RecordParser.

Парсер JSON

Вы легко можете парсить структуры JSON, но для этого нужно сразу предоставить содержимое JSON, что может быть неудобно, когда необходимо парсить очень большие структуры.

Неблокирующий парсер JSON — это управляемый событиями парсер, способный обрабатывать очень большие структуры. Он преобразует последовательность входных буферов в последовательность событий разбора JSON.

JsonParser parser = JsonParser.newParser();

// Set handlers for various events
parser.handler(event -> {
  switch (event.type()) {
    case START_OBJECT:
      // Start an objet
      break;
    case END_OBJECT:
      // End an objet
      break;
    case START_ARRAY:
      // Start an array
      break;
    case END_ARRAY:
      // End an array
      break;
    case VALUE:
      // Handle a value
      String field = event.fieldName();
      if (field != null) {
        // In an object
      } else {
        // In an array or top level
        if (event.isString()) {

        } else {
          // ...
        }
      }
      break;
  }
});

Парсер неблокирующий, и события, которые он генерирует, управляются входными буферами.

JsonParser parser = JsonParser.newParser();

// start array event
// start object event
// "firstName":"Bob" event
parser.handle(Buffer.buffer("[{\"firstName\":\"Bob\","));

// "lastName":"Morane" event
// end object event
parser.handle(Buffer.buffer("\"lastName\":\"Morane\"},"));

// start object event
// "firstName":"Luke" event
// "lastName":"Lucky" event
// end object event
parser.handle(Buffer.buffer("{\"firstName\":\"Luke\",\"lastName\":\"Lucky\"}"));

// end array event
parser.handle(Buffer.buffer("]"));

// Always call end
parser.end();

Обработка событий обеспечивает больший контроль, но сопряжена с необходимостью обработки мелкозернистых событий, что иногда может быть неудобно. Парсер JSON позволяет обрабатывать структуры JSON как значения, когда это желательно:

JsonParser parser = JsonParser.newParser();

parser.objectValueMode();

parser.handler(event -> {
  switch (event.type()) {
    case START_ARRAY:
      // Start the array
      break;
    case END_ARRAY:
      // End the array
      break;
    case VALUE:
      // Handle each object
      break;
  }
});

parser.handle(Buffer.buffer("[{\"firstName\":\"Bob\"},\"lastName\":\"Morane\"),...]"));
parser.end();

Режим значений можно устанавливать и сбрасывать во время разбора, что позволяет переключаться между мелкозернистыми событиями или событиями значений объектов JSON.

JsonParser parser = JsonParser.newParser();

parser.handler(event -> {
  // Start the object

  switch (event.type()) {
    case START_OBJECT:
      // Set object value mode to handle each entry, from now on the parser won't emit start object events
      parser.objectValueMode();
      break;
    case VALUE:
      // Handle each object
      // Get the field in which this object was parsed
      String id = event.fieldName();
      System.out.println("User with id " + id + " : " + event.value());
      break;
    case END_OBJECT:
      // Set the object event mode so the parser emits start/end object events again
      parser.objectEventMode();
      break;
  }
});

parser.handle(Buffer.buffer("{\"39877483847\":{\"firstName\":\"Bob\"},\"lastName\":\"Morane\"),...}"));
parser.end();

То же самое можно сделать и с массивами.

JsonParser parser = JsonParser.newParser();

parser.handler(event -> {
  // Start the object

  switch (event.type()) {
    case START_OBJECT:
      // Set array value mode to handle each entry, from now on the parser won't emit start array events
      parser.arrayValueMode();
      break;
    case VALUE:
      // Handle each array
      // Get the field in which this object was parsed
      System.out.println("Value : " + event.value());
      break;
    case END_OBJECT:
      // Set the array event mode so the parser emits start/end object events again
      parser.arrayEventMode();
      break;
  }
});

parser.handle(Buffer.buffer("[0,1,2,3,4,...]"));
parser.end();

Вы также можете декодировать POJO.

parser.handler(event -> {
  // Handle each object
  // Get the field in which this object was parsed
  String id = event.fieldName();
  User user = event.mapTo(User.class);
  System.out.println("User with id " + id + " : " + user.firstName + " " + user.lastName);
});

В случае, если парсер не может обработать буфер, будет выброшено исключение, если вы не установили обработчик исключений:

JsonParser parser = JsonParser.newParser();

parser.exceptionHandler(err -> {
  // Catch any parsing or decoding error
});

Парсер также парсит потоки JSON:

конкатенированные потоки JSON: {"temperature":30}{"temperature":50}
потоки JSON с разделителями строк: {"an":"object"}\r\n3\r\n"a string"\r\nnull

Для получения более подробной информации см. класс JsonParser.

Безопасность потоков

Большинство объектов Vert.x можно безопасно использовать из разных потоков. Однако производительность оптимизирована, когда к ним обращаются из того же контекста, в котором они были созданы.

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

Если вы придерживаетесь стандартной модели развертывания Vert.x вертикалей и избегаете совместного использования объектов между вертикалями, то это должно выполняться без необходимости о ней задумываться.

Выполнение блокирующего кода

В идеальном мире не будет войн или голода, все API будут написаны асинхронно, а зайцы и ягнята будут весело скакать по зелёным лугам.

Но… реальный мир не такой. (Вы смотрели новости?)

Факт в том, что многие, если не большинство библиотек, особенно в экосистеме JVM, имеют синхронные API, и многие методы, вероятно, будут блокирующими. Хороший пример — API JDBC. Он по своей природе синхронный, и как бы Vert.x ни старался, он не сможет добавить к нему магическую пыльцу, чтобы сделать его асинхронным.

Мы не будем переписывать всё заново на асинхронный режим в одночасье, поэтому нам нужно предоставить способ безопасного использования «традиционных» блокирующих API в приложении Vert.x.

Как обсуждалось ранее, вы не можете вызывать блокирующие операции непосредственно из цикла событий, так как это помешает ему выполнять другую полезную работу. Итак, как это сделать?

Это делается с помощью вызова executeBlocking с блокирующим кодом для выполнения, в результате вы получаете будущее, завершённое результатом выполнения блокирующего кода.

vertx.executeBlocking(() -> {
  // Call some blocking API that takes a significant amount of time to return
  return someAPI.blockingMethod("hello");
}).onComplete(res -> {
  System.out.println("The result is: " + res.result());
});

Блокирующий код должен блокироваться на разумный срок (т.е. не более нескольких секунд). Длительные блокирующие операции или операции опроса (т.е. поток, который вращается в цикле, опрашивая события блокирующим образом) запрещены. Если блокирующая операция длится более 10 секунд, заблокированный проверяющий поток выведет сообщение в консоль. Длительные блокирующие операции должны использовать выделенный поток, управляемый приложением, который может взаимодействовать с вертиксом, используя шину событий или runOnContext

По умолчанию, если executeBlocking вызывается несколько раз из одного контекста (например, одного и того же экземпляра вертикса), то разные вызовы executeBlocking выполняются последовательно (т.е. один за другим).

Если порядок не важен, вы можете вызвать executeBlocking, указав false в качестве аргумента для ordered. В этом случае любой вызов executeBlocking может выполняться параллельно в пуле рабочих потоков.

Альтернативный способ выполнения блокирующего кода — использование рабочего вертикса

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

По умолчанию блокирующий код выполняется в пуле рабочих потоков Vert.x, настроенном с помощью setWorkerPoolSize.

Для разных целей могут быть созданы дополнительные пулы:

WorkerExecutor executor = vertx.createSharedWorkerExecutor("my-worker-pool");
executor.executeBlocking(() -> {
  // Call some blocking API that takes a significant amount of time to return
  return someAPI.blockingMethod("hello");
}).onComplete(res -> {
  System.out.println("The result is: " + res.result());
});

Рабочий исполнитель должен быть закрыт, когда он больше не нужен:

executor.close();

Когда несколько рабочих создаются с одинаковым именем, они будут использовать один и тот же пул. Пул рабочих потоков уничтожается, когда все рабочие исполнители, использующие его, закрываются.

Когда исполнитель создаётся в вертиксе, Vert.x автоматически закроет его, когда вертикс будет развёрнут.

Рабочие исполнители могут быть настроены при создании:

int poolSize = 10;

// 2 minutes
long maxExecuteTime = 2;
TimeUnit maxExecuteTimeUnit = TimeUnit.MINUTES;

WorkerExecutor executor = vertx.createSharedWorkerExecutor("my-worker-pool", poolSize, maxExecuteTime, maxExecuteTimeUnit);

Настройка задаётся при создании пула рабочих потоков

Vert.x SPI

У экземпляра Vert.x есть несколько точек расширения, известных как SPI (интерфейс поставщика услуг).

Такие SPI часто загружаются из каталога классов с помощью механизма Java ServiceLoader.

SPI метрики и отслеживания

По умолчанию Vert.x не записывает метрики и не выполняет отслеживание. Вместо этого он предоставляет SPI для реализации другими компонентами, которые можно добавить в каталог классов. SPI метрик — это функция, позволяющая реализаторам собирать события из Vert.x для сбора и отчётности о метриках, аналогично SPI отслеживанию для следов.

Для получения дополнительной информации см. https://vertx.io/docs/#monitoring

SPI менеджера кластера

В Vert.x используется менеджер кластера для различных функций, включая:

Обнаружение и членство группы узлов Vert.x в кластере
Поддержание списков подписчиков тем в кластере (чтобы мы знали, какие узлы заинтересованы в каких адресах канала событий)
Поддержка распределённых карт
Распределённые блокировки
Распределённые счётчики

Менеджеры кластера не обрабатывают межузловую транспортную часть канала событий, это делается напрямую Vert.x с помощью TCP-соединений.

По умолчанию в дистрибутивах Vert.x используется менеджер кластера, который использует Hazelcast, но его легко заменить другой реализацией, поскольку менеджеры кластеров Vert.x являются подключаемыми.

Менеджер кластера должен реализовывать интерфейс ClusterManager. Vert.x находит менеджеры кластера во время выполнения, используя функциональность Java Service Loader для поиска экземпляров ClusterManager в каталоге классов.

Если вы используете Vert.x в командной строке и хотите использовать кластеризацию, убедитесь, что каталог lib вашей установки Vert.x содержит jar-файл вашего менеджера кластера.

Если вы используете Vert.x из проекта Maven или Gradle, просто добавьте jar-файл менеджера кластера как зависимость вашего проекта.

Для получения дополнительной информации см. https://vertx.io/docs/#clustering

Создатель Vert.x

Статические методы Vertx.vertx и Vertx.clusteredVertx — это самый простой способ получить экземпляр Vertx.

Вы также можете использовать шаблон строителя для создания экземпляра Vertx. Шаблон строителя позволяет программно настроить несколько поставщиков (SPI), которые обычно загружаются с помощью VertxOptions и/или плагинов Java Service Loader.

Нативный транспорт
Менеджер кластера
Отслеживание
Экземпляр метрик

Vertx vertx = Vertx.builder()
  .with(options)
  .withTracer(tracerFactory)
  .withMetrics(metricsFactory)
  .build();

Также можно создавать кластеризованные экземпляры.

Future<Vertx> vertx = Vertx.builder()
  .with(options)
  .withClusterManager(clusterManager)
  .buildClustered();

Ведение журнала

Vert.x использует свою внутреннюю API ведения журнала и поддерживает различные бэкенды ведения журнала.

Бэкенд ведения журнала выбирается следующим образом:

бэкенд, обозначенный свойством системы vertx.logger-delegate-factory-class-name, если оно присутствует, или,
журналирование JDK, когда файл vertx-default-jul-logging.properties находится в пути к классам, или,
бэкенд, присутствующий в пути к классам, в следующем порядке предпочтения:
1. SLF4J
2. Log4J2

В противном случае Vert.x по умолчанию использует журналирование JDK.

Настройка с помощью системного свойства

Установите системное свойство vertx.logger-delegate-factory-class-name в:

io.vertx.core.logging.SLF4JLogDelegateFactory для SLF4J или,
io.vertx.core.logging.Log4j2LogDelegateFactory для Log4J2 или,
io.vertx.core.logging.JULLogDelegateFactory для журналирования JDK

Автоматическая настройка

Когда системное свойство vertx.logger-delegate-factory-class-name не установлено, Vert.x будет пытаться найти наиболее подходящий логгер:

используйте SLF4J, если он доступен в пути к классам с фактической реализацией (т. е. LoggerFactory.getILoggerFactory() не является экземпляром NOPLoggerFactory)
в противном случае используйте Log4j2, если он доступен в пути к классам
в противном случае используйте JUL

Настройка журналирования JUL

Файл конфигурации журналирования JUL можно указать обычным для JUL способом, предоставив системное свойство с именем java.util.logging.config.file, значением которого является ваш конфигурационный файл. Для получения дополнительной информации об этом и структуре файла конфигурации JUL, пожалуйста, обратитесь к документации по журналированию JDK.

Vert.x также предоставляет немного более удобный способ указать конфигурационный файл без необходимости устанавливать системное свойство. Просто предоставьте файл конфигурации JUL с именем vertx-default-jul-logging.properties в пути к классам (например, внутри вашего fatjar), и Vert.x будет использовать его для настройки JUL.

Журналирование Netty

Netty не полагается на внешнюю конфигурацию ведения журнала (например, системные свойства). Вместо этого он реализует конфигурацию ведения журнала на основе библиотек ведения журнала, видимых из классов Netty:

используйте библиотеку SLF4J, если она видна
в противном случае используйте Log4j, если она видна
в противном случае используйте Log4j2, если она видна
в противном случае используйте java.util.logging по умолчанию

Внимательные из вас могли заметить, что Vert.x следует тому же порядку предпочтения.

Реализация логгера может быть принудительно установлена на определенную реализацию, путем непосредственной установки внутренней реализации логгера Netty на io.netty.util.internal.logging.InternalLoggerFactory:

// Force logging to Log4j 2
InternalLoggerFactory.setDefaultFactory(Log4J2LoggerFactory.INSTANCE);

Устранение неполадок

Предупреждение SLF4J при запуске

Если при запуске приложения вы видите следующее сообщение:

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

Это означает, что в вашей classpath присутствует SLF4J-API, но нет фактической привязки. Сообщения, записанные с помощью SLF4J, будут пропущены. Вам необходимо добавить привязку в вашу classpath. Обратитесь к https://www.slf4j.org/manual.html#swapping, чтобы выбрать привязку и настроить её.

Обратите внимание, что Netty ищет JAR SLF4-API и использует его по умолчанию.

Подключение прервано удалённым узлом

Если в ваших логах присутствует множество:

io.vertx.core.net.impl.ConnectionBase
SEVERE: java.io.IOException: Connection reset by peer

Это означает, что клиент прерывает HTTP-соединение вместо его закрытия. Это сообщение также указывает на то, что, возможно, вы не обработали весь payload (соединение было прервано до того, как вы смогли его обработать).

Разрешение имени хоста

Vert.x использует разрешитель адресов для преобразования имени хоста в IP-адреса вместо встроенного блокирующего разрешителя JVM.

Имя хоста преобразуется в IP-адрес с помощью:

файла hosts операционной системы
в противном случае запросов DNS к списку серверов

По умолчанию он будет использовать список адресов DNS-серверов системы из среды; если этот список получить невозможно, он будет использовать публичные DNS-серверы Google "8.8.8.8" и "8.8.4.4".

DNS-серверы также можно настроить при создании экземпляра Vertx:

Vertx vertx = Vertx.vertx(new VertxOptions().
    setAddressResolverOptions(
        new AddressResolverOptions().
            addServer("192.168.0.1").
            addServer("192.168.0.2:40000"))
);

Порт DNS-сервера по умолчанию равен 53; если сервер использует другой порт, этот порт можно указать с помощью разделителя двоеточие: 192.168.0.2:40000.

иногда может быть желательно использовать встроенный разрешитель JVM; системная переменная среды JVM -Dvertx.disableDnsResolver=true активирует это поведение

Резервирование

Когда сервер не отвечает в разумные сроки, разрешитель попробует следующий сервер из списка; поиск ограничен setMaxQueries (значение по умолчанию — 4 запросов).

Запрос DNS считается неудачным, если разрешитель не получил правильный ответ в течение getQueryTimeout миллисекунд (значение по умолчанию — 5 секунд).

Циклическое переключение списка серверов

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

Вы можете настроить setRotateServers на true, чтобы разрешитель выполнял циклический выбор вместо этого. Это распределит нагрузку запросов между серверами и предотвратит все обращение к первому серверу в списке.

Резервирование по-прежнему применимо и будет использовать следующий сервер в списке.

Сопоставление хостов

Файл hosts операционной системы используется для поиска IP-адреса по имени хоста.

Можно использовать альтернативный файл hosts:

Vertx vertx = Vertx.vertx(new VertxOptions().
    setAddressResolverOptions(
        new AddressResolverOptions().
            setHostsPath("/path/to/hosts"))
);

Домены поиска

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

Vertx vertx = Vertx.vertx(new VertxOptions().
    setAddressResolverOptions(
        new AddressResolverOptions().addSearchDomain("foo.com").addSearchDomain("bar.com"))
);

Когда используется список доменов поиска, порог количества точек равен 1 или загружается из /etc/resolv.conf в Linux; можно настроить его на конкретное значение с помощью setNdots.

Настройка MacOS

MacOS имеет специальное встроенное расширение для получения конфигурации сервера имен системы на основе открытого исходного кода mDNSResponder от Apple. Когда это расширение отсутствует, Netty выводит следующее предупреждение.

[main] WARN io.netty.resolver.dns.DnsServerAddressStreamProviders - Can not find io.netty.resolver.dns.macos.MacOSDnsServerAddressStreamProvider in the classpath, fallback to system defaults. This may result in incorrect DNS resolutions on MacOS.

Это расширение не является обязательным, так как его отсутствие не препятствует выполнению Vert.x, но рекомендуется.

Вы можете добавить его в свой класс, чтобы улучшить интеграцию и устранить предупреждение.

Mac с процессорами Intel

<profile>
  <id>mac-intel</id>
  <activation>
    <os>
      <family>mac</family>
      <arch>x86_64</arch>
    </os>
  </activation>
  <dependencies>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-resolver-dns-native-macos</artifactId>
      <classifier>osx-x86_64</classifier>
      <!--<version>Should align with netty version that Vert.x uses</version>-->
    </dependency>
  </dependencies>
</profile>

Mac с чипами M1/M2

<profile>
  <id>mac-silicon</id>
  <activation>
    <os>
      <family>mac</family>
      <arch>aarch64</arch>
    </os>
  </activation>
  <dependencies>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-resolver-dns-native-macos</artifactId>
      <classifier>osx-aarch_64</classifier>
      <!--<version>Should align with netty version that Vert.x uses</version>-->
    </dependency>
  </dependencies>
</profile>

Встроенные транспортные средства

Vert.x может работать с встроенными транспортными средствами (если доступны) на BSD (OSX) и Linux:

Vertx vertx = Vertx.vertx(new VertxOptions().
  setPreferNativeTransport(true)
);

// True when native is available
boolean usingNative = vertx.isNativeTransportEnabled();
System.out.println("Running with native: " + usingNative);

Использование встроенного транспорта не помешает запуску приложения (например, может отсутствовать встроенная зависимость). Если ваше приложение требует встроенного транспорта, вам нужно проверить isNativeTransportEnabled.

Вы также можете явно настроить транспорт для использования:

Transport transport = Transport.nativeTransport();

// Or use a very specific transport
transport = Transport.EPOLL;

Vertx vertx = Vertx.builder()
  .withTransport(transport)
  .build();

Встроенный epoll

Встроенный epoll на Linux предоставляет дополнительные сетевые возможности:

SO_REUSEPORT
TCP_QUICKACK
TCP_CORK
TCP_FASTOPEN
TCP_USER_TIMEOUT

Вам нужно добавить следующую зависимость в свой класс:

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-transport-native-epoll</artifactId>
  <classifier>linux-x86_64</classifier>
  <!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

Встроенный io_uring

Вам нужно добавить следующую зависимость в свой класс:

<dependency>
  <groupId>io.netty</groupId>
  <classifier>linux-x86_64</classifier>
  <artifactId>netty-transport-native-io_uring</artifactId>
  <!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

Встроенный kqueue

Вам нужно добавить следующую зависимость в свой класс:

Mac с процессорами Intel

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-transport-native-kqueue</artifactId>
  <classifier>osx-x86_64</classifier>
  <!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

Mac с чипами M1/M2

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-kqueue</artifactId>
<classifier>osx-aarch_64</classifier>
<!--<version>Should align with netty version that Vert.x uses</version>-->
</dependency>

Поддерживаются MacOS Sierra и выше.

Встроенный BSD предоставляет дополнительные сетевые возможности:

SO_REUSEPORT

vertx.createHttpServer(new HttpServerOptions().setReusePort(reusePort));

Примечания по безопасности

Vert.x — это набор инструментов, а не настроенный фреймворк, где мы заставляем вас делать вещи определённым способом. Это даёт вам большую силу как разработчику, но с этим приходит большая ответственность.

Как и с любым набором инструментов, возможно создание небезопасных приложений, поэтому следует проявлять осторожность при разработке приложения, особенно если оно доступно публично (например, через интернет).

Веб-приложения

При разработке веб-приложения настоятельно рекомендуется использовать Vert.x-Web вместо Vert.x core для предоставления ресурсов и обработки загрузки файлов.

Vert.x-Web нормализует путь в запросах, чтобы предотвратить злонамеренным клиентам создание URL-адресов для доступа к ресурсам за пределами корневого каталога веб-сервера.

Аналогично, для загрузки файлов Vert.x-Web предоставляет функциональность для загрузки в известное место на диске и не полагается на имя файла, предоставленное клиентом в загрузке, которое может быть сгенерировано для загрузки в другое место на диске.

Сам Vert.x core не предоставляет таких проверок, поэтому вам, как разработчику, придётся реализовать их самостоятельно.

Сеансы обмена сообщениями кластера

При кластеризации обмена сообщениями между разными узлами Vert.x в сети трафик отправляется без шифрования по сети, поэтому не используйте этот способ, если у вас есть конфиденциальные данные для отправки, и ваши узлы Vert.x не находятся в надёжной сети.

Стандартные рекомендации по безопасности

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

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

Настройка кэша Vert.x

Когда Vert.x нуждается в чтении файла из classpath (встроенного в fat jar, в jar из classpath или файла, который находится в classpath), он копирует его в каталог кэша. Причина проста: чтение файла из jar или из потока ввода-вывода является блокирующей операцией. Чтобы избежать оплаты за это каждый раз, Vert.x копирует файл в каталог своего кэша и читает его оттуда при каждом последующем чтении. Это поведение можно настроить.

По умолчанию Vert.x использует $CWD/.vertx в качестве каталога кэша. Создается уникальный каталог внутри него, чтобы избежать конфликтов. Это расположение можно настроить, используя системную переменную vertx.cacheDirBase. Например, если текущая рабочая директория не доступна для записи (например, в контексте неизменяемого контейнера), запустите приложение с:

java -jar my-fat.jar vertx.cacheDirBase=/tmp/vertx-cache

Каталог должен быть доступен для записи.

При редактировании ресурсов, таких как HTML, CSS или JavaScript, этот механизм кэширования может быть раздражающим, так как он отображает только первую версию файла (и поэтому вы не увидите свои изменения при перезагрузке страницы). Чтобы избежать этого поведения, запустите приложение с -Dvertx.disableFileCaching=true. С этим параметром Vert.x по-прежнему использует кэш, но всегда обновляет версию, хранящуюся в кэше, исходным источником. Таким образом, если вы отредактируете файл, предоставляемый из classpath, и обновите свой браузер, Vert.x прочитает его из classpath, скопирует его в каталог кэша и предоставит его оттуда. Не используйте этот параметр в производстве, он может убить производительность.

Наконец, вы можете полностью отключить кэш, используя -Dvertx.disableFileCPResolving=true. Этот параметр имеет последствия. Vert.x не сможет читать файлы из classpath (только из файловой системы). Будьте очень осторожны при использовании этого параметра.

© 2025 Eclipse Vert.x™Eclipse Vert.x™ is open source and dual-licensed under the Eclipse Public License 2.0 and the Apache License 2.0.Website design by Michel Krämer.
https://vertx.io/docs/vertx-core/java/