Как эффективно читать из сокета с помощью Java NIO

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

Я начал с самого простого прототипа java nio, подобного этому

ByteBuffer buf = ByteBuffer.allocateDirect(BUFFER_SIZE);
try {
       buf.clear();
       int numBytesRead = socketChannel.read(buf);

       if (numBytesRead == -1) {  
           socketChannel.close();
       } else {
           buf.flip();
           byte[] byteArrived = new byte[buf.remaining];
           buf.get(byteArrived,0,byteArrived.length);
           // here we send byteArrived to the parser
       }
   } catch (IOException e) {    
}

Я думаю, что создавать массив byte[] каждый раз неудобно, но из-за отсутствия знаний я не знаю, как анализировать ByteBuffer (потому что мне нужно разобрать байтовый протокол в сообщения и передать их в бизнес-логику). Можете ли вы порекомендовать, как избежать массового создания мусора?

Также я хотел бы спросить о лучших практиках, как организовать чтение сокетов с низкой задержкой и высокой пропускной способностью? Я читал о LMAX и инфраструктуре Disruptor, и они достигли 6 миллионов транзакций в одном потоке.


java networking nio
person Egor Lakomkin    schedule 29.11.2011    source источник

Ответы (2)


arrow_upward
4
arrow_downward

Вы можете достичь большего, чем с Disruptor и другими методами. Многое зависит от размера и сложности сообщения (а также от того, что вы делаете с сообщением!!)

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

Эта презентация может показаться вам интересной http://vanillajava.blogspot.com/2011/11/low-latency-slides.html

person Peter Lawrey    schedule 29.11.2011
comment
Сообщения довольно простые и имеют длину 12 или 27 байт. - person Egor Lakomkin; 30.11.2011
comment
В этом случае вы можете получить от 250Kmsg/s до 40M msg/s в зависимости от того, хотите ли вы самую низкую задержку (всего 5 мкс) или максимальную пропускную способность (но с задержкой 25 мкс). Конечно, все это должно откуда-то прийти и уйти. куда-то. Нет смысла беспокоиться об этом, если у вас нет спроса и ваши потребители не могут обрабатывать сообщения так быстро. - person Peter Lawrey; 30.11.2011
comment
В соответствии с рекомендациями - укажите длину в качестве первого поля и имейте в виду, что одно чтение из сокета может доставлять более одного сообщения, а также частичное сообщение. Другие вещи, которые следует учитывать: а) создать экземпляр буфера один раз б) создать пул объектов, обертывающих byte[] и предоставляющих методы доступа/другие методы в стиле bean (тогда сериализация/десериализация так же проста, как копирование байтов между буфером ввода-вывода и внутренним массивом объекта) - person Andrey Nudko; 30.11.2011
comment
спасибо за ваши советы, но я не могу указать длину сообщения в теле, потому что протокол уже определен поставщиком данных (я должен полагаться на то, что существует 3 типа двоичных сообщений, все они имеют фиксированную длину). Итак, вы рекомендуете создать ByteBuffer один раз, но я не получил пул объектов, обертывающих byte[]. Также нет способа получитьByte из ByteBuffer, потому что мне нужно, например, получить один байт и разобрать его, например, на младшие биты/более высокие ноты. Я сделал наивный способ, но я хочу улучшить производительность, задержку и не создавать столько мусора при разборе сообщений. - person Egor Lakomkin; 30.11.2011
comment
Если вы хотите get как байт, вы можете просто использовать get(). Вы должны иметь возможность выполнять весь синтаксический анализ, не создавая мусора. ;) - person Peter Lawrey; 30.11.2011

arrow_upward
1
arrow_downward

Предполагая, что вы можете адаптировать свой API парсера для приема (byte[] buffer, int offset, int length) в качестве аргументов, вы можете просто передать (bb.array(), 0, bb.limit()) в качестве параметров и вообще не создавать new byte[] для чтения. Однако это вряд ли будет шагом, определяющим скорость.

person user207421    schedule 30.11.2011