Высокопроизводительный сервер NIO TCP с высокой нагрузкой

В рамках моего исследования я пишу эхо-сервер с высокой нагрузкой TCP/IP в Java. Я хочу обслуживать около 3-4 тыс. Клиентов и просматривать максимально возможные сообщения в секунду, которые я могу выжать из него. Размер сообщения довольно мал - до 100 байт. Эта работа не имеет практической цели - только исследования.

В соответствии с многочисленными презентациями, которые я видел (тесты HornetQ, переговоры LMAX Disruptor и т.д.), системы высокой нагрузки реального мира имеют тенденцию обслуживать миллионы транзакций в секунду (я считаю, что Disruptor упомянул около 6 мил и Hornet - 8.5). Например, этот пост утверждает, что можно достичь до 40M MPS. Поэтому я воспринял это как приблизительную оценку того, на что способен современное оборудование.

Я написал простейший однопоточный NIO-сервер и запустил тест нагрузки. Я был немного удивлен, что могу получить только около 100 тыс. MPS на localhost и 25 тыс. С реальной сетью. Номера выглядят довольно маленькими. Я тестировал Win7 x64, ядро ​​i7. Глядя на загрузку процессора - занято только одно ядро ​​(что ожидается в однопоточном приложении), а остальные сидят без дела. Однако, даже если я загружу все 8 ядер (включая виртуальные), у меня будет не более 800k MPS - даже не до 40 миллионов:)

Мой вопрос: что такое типичный шаблон для подачи огромного количества сообщений клиентам? Должен ли я распределять сетевую нагрузку на несколько разных сокетов внутри одной JVM и использовать какой-то балансировщик нагрузки, например HAProxy, для распределения нагрузки на несколько ядер? Или я должен смотреть на использование нескольких селекторов в моем коде NIO? Или, может быть, даже распределить нагрузку между несколькими JVM и использовать Chronicle для создания межпроцессного взаимодействия между ними? Будет ли тестирование на правильной серверной ОС, такой как CentOS, иметь большое значение (может быть, Windows замедляет работу)?

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

public class EchoServer {

private static final int BUFFER_SIZE = 1024;
private final static int DEFAULT_PORT = 9090;

// The buffer into which we'll read data when it available
private ByteBuffer readBuffer = ByteBuffer.allocate(BUFFER_SIZE);

private InetAddress hostAddress = null;

private int port;
private Selector selector;

private long loopTime;
private long numMessages = 0;

public EchoServer() throws IOException {
    this(DEFAULT_PORT);
}

public EchoServer(int port) throws IOException {
    this.port = port;
    selector = initSelector();
    loop();
}

private void loop() {
    while (true) {
        try{
            selector.select();
            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();

                if (!key.isValid()) {
                    continue;
                }

                // Check what event is available and deal with it
                if (key.isAcceptable()) {
                    accept(key);
                } else if (key.isReadable()) {
                    read(key);
                } else if (key.isWritable()) {
                    write(key);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

private void accept(SelectionKey key) throws IOException {
    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

    SocketChannel socketChannel = serverSocketChannel.accept();
    socketChannel.configureBlocking(false);
    socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
    socketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
    socketChannel.register(selector, SelectionKey.OP_READ);

    System.out.println("Client is connected");
}

private void read(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Clear out our read buffer so it ready for new data
    readBuffer.clear();

    // Attempt to read off the channel
    int numRead;
    try {
        numRead = socketChannel.read(readBuffer);
    } catch (IOException e) {
        key.cancel();
        socketChannel.close();

        System.out.println("Forceful shutdown");
        return;
    }

    if (numRead == -1) {
        System.out.println("Graceful shutdown");
        key.channel().close();
        key.cancel();

        return;
    }

    socketChannel.register(selector, SelectionKey.OP_WRITE);

    numMessages++;
    if (numMessages%100000 == 0) {
        long elapsed = System.currentTimeMillis() - loopTime;
        loopTime = System.currentTimeMillis();
        System.out.println(elapsed);
    }
}

private void write(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    ByteBuffer dummyResponse = ByteBuffer.wrap("ok".getBytes("UTF-8"));

    socketChannel.write(dummyResponse);
    if (dummyResponse.remaining() > 0) {
        System.err.print("Filled UP");
    }

    key.interestOps(SelectionKey.OP_READ);
}

private Selector initSelector() throws IOException {
    Selector socketSelector = SelectorProvider.provider().openSelector();

    ServerSocketChannel serverChannel = ServerSocketChannel.open();
    serverChannel.configureBlocking(false);

    InetSocketAddress isa = new InetSocketAddress(hostAddress, port);
    serverChannel.socket().bind(isa);
    serverChannel.register(socketSelector, SelectionKey.OP_ACCEPT);
    return socketSelector;
}

public static void main(String[] args) throws IOException {
    System.out.println("Starting echo server");
    new EchoServer();
}
}

Ответы

Ответ 1

what is a typical pattern for serving massive amounts of messages to clients?

Существует много возможных шаблонов: Легкий способ использования всех ядер без прохождения нескольких jvms:

  • Попросите один поток принять соединения и прочитать с помощью селектора.
  • Как только у вас будет достаточно байтов для создания одного сообщения, передайте его другому ядру, используя конструкцию, такую ​​как кольцевой буфер. Рамка Disruptor Java подходит для этого. Это хороший шаблон, если обработка, необходимая для того, чтобы знать, что является полным сообщением, является легким. Например, если у вас есть префикс длины, вы можете подождать, пока не получите ожидаемое количество байтов, а затем отправьте его в другой поток. Если синтаксический анализ протокола очень тяжелый, вы можете подавить этот единственный поток, не позволяя ему принимать соединения или считывать байты в сети.
  • В рабочем потоке (-ах), который принимает данные из кольцевого буфера, выполните фактическую обработку.
  • Вы пишете ответы либо на своих рабочих потоках, либо через какой-то другой поток агрегатора.

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

  • Приложение с тяжелым статусом CPU говорит о приложении обработки изображений. Объем работы CPU/GPU для каждого запроса, вероятно, будет значительно выше, чем накладные расходы, генерируемые очень наивным межпоточным коммуникационным решением. В этом случае легкое решение - это куча рабочих потоков, вытягивающих работу из одной очереди. Обратите внимание, что это одна очередь, а не одна очередь для каждого рабочего. Преимущество заключается в том, что по своей сути сбалансированность нагрузки. Каждый работник завершает работу, а затем просто проверяет однопроцессорную очередь с несколькими потребителями. Несмотря на то, что это источник споров, работа по обработке изображений (секунды?) Должна быть намного дороже, чем любая альтернатива синхронизации.
  • Приложение чистого ввода-вывода, например. сервер статистики, который просто увеличивает количество счетчиков для запроса: здесь вы почти не работаете с ЦП. Большая часть работы - это просто чтение байтов и запись байтов. Многопоточное приложение может не принести вам существенного преимущества. На самом деле это может даже замедлить работу, если время, затрачиваемое на очередь, больше, чем время, затрачиваемое на их обработку. Однопоточный Java-сервер должен быть способен насыщать 1G-ссылку.

  • Учетные заявления, требующие умеренных объемов обработки, например. типичное деловое приложение: здесь каждый клиент имеет определенное состояние, которое определяет, как обрабатывается каждый запрос. Предполагая, что мы идем многопоточными, поскольку обработка является нетривиальной, мы могли бы аффинировать клиентов определенным потокам. Это вариант архитектуры актера:

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

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

    iii) Рабочий поток может писать запросы. Всегда пытайтесь просто написать write(). Если все ваши данные не могут быть записаны только тогда, вы регистрируетесь для OP_WRITE. Рабочий поток должен только делать выборные вызовы, если есть что-то выдающееся. Большинство писем должны просто преуспеть, делая это ненужным. Трюк здесь заключается в балансировке между выборами и опросе буфера звонка для большего количества запросов. Вы также можете использовать один поток писем, единственная ответственность которых заключается в том, чтобы писать запросы. Каждый рабочий поток может помещать ответы в кольцевой буфер, соединяющий его с этим единственным потоком записи. Обходной поток однопользовательского потока проверяет каждый входящий кольцевой буфер и записывает данные клиентам. Опять же, оговорка о попытке записи перед выбором применяется, как и трюк о балансировке между несколькими буферами звонков и выбора вызовов.

Как вы указываете, есть много других опций:

Should I distribute networking load over several different sockets inside a single JVM and use some sort of load balancer like HAProxy to distribute load to multiple cores?

Вы можете сделать это, но IMHO это не лучшее использование для балансировки нагрузки. Это купит вам независимые JVM, которые могут выйти из строя самостоятельно, но, вероятно, будут медленнее, чем писать одно JVM-приложение, которое многопоточно. Само приложение может быть проще записать, поскольку оно будет однопоточным.

Or I should look towards using multiple Selectors in my NIO code?

Вы тоже можете это сделать. Посмотрите на архитектуру Ngnix для некоторых подсказок о том, как это сделать.

Or maybe even distribute the load between multiple JVMs and use Chronicle to build an inter-process communication between them? Это также вариант. Хроника дает вам преимущество в том, что файлы с отображением памяти более устойчивы к процессу, выходящему посередине. Вы по-прежнему получаете много производительности, так как все общение осуществляется через разделяемую память.

Will testing on a proper serverside OS like CentOS make a big difference (maybe it is Windows that slows things down)?

Я не знаю об этом. Вряд ли. Если Java использует встроенные API Windows в полной мере, это не имеет большого значения. Я очень сомневаюсь в 40 миллионах транзакций/сек (без сетевого пространства пользователя + UDP), но перечисленные мной архитектуры должны делать очень хорошо.

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

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

Конструкция протокола также сильно влияет на производительность. Я предпочитаю длинные префиксные протоколы, потому что они позволяют вам распределять буферы правильных размеров, избегая списков буферов и/или слияния буфера. Длинные префиксные протоколы также позволяют легко решить, когда передать запрос - просто отметьте num bytes == expected. Фактический синтаксический анализ может выполняться рабочей нитью. Сериализация и десериализация выходят за пределы протоколов с префиксом длины. Здесь можно использовать шаблоны, такие как мухи-паттерны над буферами вместо распределения. Посмотрите на SBE для некоторых из этих принципов.

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

Ответ 2

Ваша логика при написании ошибочна. Вы должны попытаться написать сразу, что у вас есть данные для записи. Если write() возвращает ноль, тогда нужно зарегистрироваться для OP_WRITE, повторить запись, когда канал станет доступен для записи, и отменить регистрацию для OP_WRITE, когда запись выполнена. Здесь вы добавляете огромное количество латентности. Вы добавляете еще больше латентности, отменя регистрацию на OP_READ, пока вы делаете все это.

Ответ 3

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

Наилучший подход, как правило, зависит от того, связаны ли вы с привязкой io-bound или cpu-bound.

Для загрузок с привязкой к io (с высокой задержкой) вам необходимо выполнить асинхронную работу со многими потоками. Для достижения максимальной производительности вы должны стараться как можно меньше аннулировать передачу обслуживания между потоками. Таким образом, наличие выделенного потока селектора и другого потока потока для обработки выполняется медленнее, чем наличие пула потоков, в котором каждый поток выполняет либо выбор, либо обработку, так что этот запрос обрабатывается одним потоком в лучшем случае (если io сразу же доступен). Этот тип настройки сложнее кодировать, но быстро, и я не считаю, что любая асинхронная веб-структура полностью использует это.

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