Проблемы с нагрузкой на загрузку Java Netty

Question

Я написал сервер, который принимает сообщения о подключении и бомбардировках (~ 100 байт) с использованием текстового протокола, и моя реализация может отправлять сообщения о шлейфе 400K/sec с клиентским интерфейсом 3rt. Я выбрал Netty для этой задачи, SUSE 11 RealTime, JRockit RTS. Но когда я начал разрабатывать собственный клиент на основе Netty, я столкнулся с резким сокращением пропускной способности (от 400 К до 1,3 КБ/сек). Код клиента довольно прост. Не могли бы вы дать совет или показать примеры, как писать гораздо более эффективный клиент. Я, на самом деле, больше заботился о латентности, но начал с тестов пропускной способности, и я не думаю, что нормально иметь 1.5Kmsg/sec на loopback. P.S. клиентская цель - получать сообщения только от сервера и очень редко посылать сердечные сокращения.

Client.java 

public class Client { 

private static ClientBootstrap bootstrap; 
private static Channel connector; 
public static boolean start() 
{ 
    ChannelFactory factory = 
     new NioClientSocketChannelFactory(
       Executors.newCachedThreadPool(), 
       Executors.newCachedThreadPool()); 
    ExecutionHandler executionHandler = new ExecutionHandler(new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576)); 

    bootstrap = new ClientBootstrap(factory); 

    bootstrap.setPipelineFactory(new ClientPipelineFactory()); 

    bootstrap.setOption("tcpNoDelay", true); 
    bootstrap.setOption("keepAlive", true); 
    bootstrap.setOption("receiveBufferSize", 1048576); 
    ChannelFuture future = bootstrap 
      .connect(new InetSocketAddress("localhost", 9013)); 
    if (!future.awaitUninterruptibly().isSuccess()) { 
     System.out.println("--- CLIENT - Failed to connect to server at " + 
          "localhost:9013."); 
     bootstrap.releaseExternalResources(); 
     return false; 
    } 

    connector = future.getChannel(); 

    return connector.isConnected(); 
} 
public static void main(String[] args) 
{ 
    boolean started = start(); 
    if (started) 
     System.out.println("Client connected to the server"); 
} 

} 

ClientPipelineFactory.java 

public class ClientPipelineFactory implements ChannelPipelineFactory{ 

private final ExecutionHandler executionHandler; 
public ClientPipelineFactory(ExecutionHandler executionHandle) 
{ 
    this.executionHandler = executionHandle; 
} 
@Override 
public ChannelPipeline getPipeline() throws Exception { 
    ChannelPipeline pipeline = pipeline(); 
    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(
       1024, Delimiters.lineDelimiter())); 
    pipeline.addLast("executor", executionHandler); 
    pipeline.addLast("handler", new MessageHandler()); 

    return pipeline; 
} 

} 

MessageHandler.java 
public class MessageHandler extends SimpleChannelHandler{ 

long max_msg = 10000; 
long cur_msg = 0; 
long startTime = System.nanoTime(); 
@Override 
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) { 
    cur_msg++; 

    if (cur_msg == max_msg) 
    { 
     System.out.println("Throughput (msg/sec) : " + max_msg* NANOS_IN_SEC/( System.nanoTime() - startTime) ); 
     cur_msg = 0; 
     startTime = System.nanoTime(); 
    } 
} 

@Override 
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { 
    e.getCause().printStackTrace(); 
    e.getChannel().close(); 
} 

} 

Обновление. На стороне сервера существует периодический поток, который записывается на принятый клиентский канал. И канал скоро станет неприемлемым. Обновление N2. Добавлен OrderedMemoryAwareExecutor в конвейере, но все еще есть очень низкая пропускная способность (около 4k msg/sec)

Исправлено. Я поставил исполнитель перед целым стеком трубопроводов, и он сработал!

Answer 1

Если сервер отправляет сообщения с фиксированным размером (~ 100 байт), вы можете установить ReceiveBufferSizePredictor клиентского загрузчик, это позволит оптимизировать для чтения

bootstrap.setOption("receiveBufferSizePredictorFactory", 
      new AdaptiveReceiveBufferSizePredictorFactory(MIN_PACKET_SIZE, INITIAL_PACKET_SIZE, MAX_PACKET_SIZE)); 

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

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

package org.jboss.netty.channel.socket.nio; 

import org.jboss.netty.channel.Channel; 

public final class NioChannelUtil { 
    public static long getWriteTaskQueueCount(Channel channel) { 
    NioSocketChannel nioChannel = (NioSocketChannel) channel; 
    return nioChannel.writeBufferSize.get(); 
    } 
}