java - 提高JAVA中Netty服务器的性能

Question

我遇到这样的情况:我的 Netty 服务器将以极快的速度从客户端获取数据。我认为客户端正在使用某种 PUSH 机制来实现该速度。我不知道PUSH - POP机制到底是什么，但我确实觉得客户端正在使用某种机制以非常高的速度发送数据。现在我的要求是，我编写了一个简单的TCP Netty服务器，它从客户端接收数据客户端并只是添加到使用 ArrayBlockingQueue 实现的 BlockingQueue 中。现在，由于 Netty 是基于事件的，接受数据并将其存储在队列中所花费的时间更多，这在客户端引发了 Netty 服务器未运行的异常。但我的服务器运行完美，但接受单个数据并将其存储到队列中的时间较多。我怎样才能防止这种情况发生？对于这种情况，有没有最快的队列？我使用 BlockingQueue 作为另一个线程将从队列中获取数据并处理它。所以我需要一个同步队列。如何提高服务器的性能或者有什么方法可以非常高的速度插入数据？我关心的只是延迟。延迟需要尽可能低。

我的服务器代码:

public class Server implements Runnable {
private final int port;

static String message;
Channel channel;
ChannelFuture channelFuture;
int rcvBuf, sndBuf, lowWaterMark, highWaterMark;

public Server(int port) {
    this.port = port;
    rcvBuf = 2048;
    sndBuf =   2048;
    lowWaterMark = 1024;
    highWaterMark = 2048;
}

@Override
public void run() {
    try {
        startServer();

    } catch (Exception ex) {
       System.err.println("Error in  Server : "+ex);
       Logger.error(ex.getMessage());
    }
}

public void startServer()  {
   // System.out.println("8888 Server started");
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(group)
         .channel(NioServerSocketChannel.class)
         .localAddress(new InetSocketAddress(port))
         .childOption(ChannelOption.SO_RCVBUF, rcvBuf * 2048)
         .childOption(ChannelOption.SO_SNDBUF, sndBuf * 2048)
         .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark * 2048, highWaterMark * 2048))
         .childOption(ChannelOption.TCP_NODELAY, true)

             .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        channel = ch;
                        System.err.println("OMS connected : " + ch.localAddress());
                        ch.pipeline().addLast(new ReceiveFromOMSDecoder());
                    }
                });
        channelFuture = b.bind(port).sync();
        this.channel = channelFuture.channel();
        channelFuture.channel().closeFuture().sync();

    } catch (InterruptedException ex) {
        System.err.println("Exception raised in SendToOMS class"+ex);
    } finally {
        group.shutdownGracefully();
    }
}

}

我的ServerHandler代码:

   @Sharable
public class ReceiveFromOMSDecoder extends MessageToMessageDecoder<ByteBuf> {


    private  Charset charset;
    public ReceiveFromOMSDecoder() {
        this(Charset.defaultCharset());
    }

    /**
     * Creates a new instance with the specified character set.
     */
    public ReceiveFromOMSDecoder(Charset charset) {
        if (charset == null) {
            throw new NullPointerException("charset");
        }
        this.charset = charset;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {

       String buffer = msg.toString(charset);
       if(buffer!=null){
        Server.sq.insertStringIntoSendingQueue(buffer);  //inserting into queue
      }
        else{
            Logger.error("Null string received"+buffer);
        }
    }
       @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
      //  Logger.error(cause.getMessage());
        System.err.println(cause);
    }
}

Answer 1

三个快速:

1. 您似乎没有发送回复。你可能应该这样做。
2. 不要阻塞 IO 线程。使用EventExecutorGroup调度传入有效负载的处理。即类似 ChannelPipeline.addLast(EventExecutorGroup group, String name, ChannelHandler handler) .
3. 一般情况下不要屏蔽即可。放弃你的 ArrayBlockingQueue 并看看 JCTools或其他一些实现来查找非阻塞模拟。