Есть ли способ контролировать количество байтов, прочитанных в TcpClient Reactor Netty? - PullRequest
Новая
       28

Есть ли способ контролировать количество байтов, прочитанных в TcpClient Reactor Netty?

0 голосов
/ 07 мая 2020

Я использую TcpClient для подключения к простому эхо-серверу TCP. Сообщения состоят из размера сообщения в 4 байта, за которым следует само сообщение. Например, чтобы отправить сообщение «привет», сервер будет ожидать «0005hello» и ответить «0005hello». При тестировании под нагрузкой (примерно 300+ одновременных пользователей) смежные запросы иногда приводят к «скоплению» ответов, например, отправка «0004good» с последующим «0003day» может привести к тому, что клиент получит «0004good0003», а затем «day».

В обычном TCP-клиенте, не основанном на WebFlux, обычно можно прочитать первые 4 байта из сокета в буфер, определить длину сообщения N, а затем прочитать следующие N байтов из сокета в буфер перед возвратом ответа. Возможно ли достичь такого детального управления, возможно, используя базовый канал TcpClient?

Я также рассмотрел подход накопления ответов в некоторой структуре данных (Queue, StringBuffer, et c.) И наличия демон анализирует результат, но на практике это не дает желаемой производительности.

1 Ответ

0 голосов
/ 13 мая 2020

Я решил это, добавив в соединение обработчик типа LengthFieldBasedFrameDecoder: 

TcpClient.create()
        .host(ADDRESS)
        .port(PORT)
        .doOnConnected((connection) -> {
            connection.addHandler("parseLengthFromFirstFourBytes", new LengthFieldBasedFrameDecoder(9999, 0, 4) {
                @Override
                protected long getUnadjustedFrameLength(ByteBuf buf, int offset, int length, ByteOrder order) {
                    ByteBuf lengthBuffer = buf.copy(0, 4);
                    byte[] messageLengthBytes = new byte[4];
                    lengthBuffer.readBytes(messageLengthBytes);
                    String messageLengthString = new String(messageLengthBytes);
                    return Long.parseLong(messageLengthString);
                }
            });
        })
        .connect()
        .subscribe();

Это решает проблему с предупреждением о том, что ответы по-прежнему «накапливаются» (как описано в вопросе), когда приложение подвергается достаточной нагрузке.

...