编辑:创建了Github仓库:https://github.com/istiillyes/client-server-netty
我使用netty 4.0.15.Final创建了一个客户端-服务器,并使用OIO和NIO进行了一些测试。
我发送了一些字符串,大小不同 [1KB, 10KB, 100KB]。
我需要服务器和客户端能够并行发送消息,因此测试看起来像这样:
- 启动服务器(创建用于接受连接的通道)
- 启动客户端(创建连接到服务器的通道)
- 在通道变为活动状态时,从客户端向服务器发送100条消息(反之亦然)。
使用NIO,消息被传输,一切正常。
但是,在使用OIO时,无论是服务器还是客户端都会在一段时间后被阻塞在java.net.SocketOutputStream.write(byte[])中,永远不会返回。
有任何想法为什么会发生这种情况吗?我在使用netty方面做错了什么吗?
我使用普通的Java套接字进行了相同的测试,而且它可以工作。因此,我猜测要么我没有正确使用netty,要么这是一个bug。
我在这里添加了创建通道和通道处理程序的代码。
这是从客户端捕获的YourKit堆栈跟踪:
pool-1-thread-1 [RUNNABLE, IN_NATIVE]
java.net.SocketOutputStream.write(byte[])
io.netty.buffer.UnpooledUnsafeDirectByteBuf.getBytes(int, OutputStream, int)
io.netty.buffer.AbstractByteBuf.readBytes(OutputStream, int)
io.netty.channel.oio.OioByteStreamChannel.doWriteBytes(ByteBuf)
io.netty.channel.oio.AbstractOioByteChannel.doWrite(ChannelOutboundBuffer)
io.netty.channel.AbstractChannel$AbstractUnsafe.flush0()
io.netty.channel.AbstractChannel$AbstractUnsafe.flush()
io.netty.channel.DefaultChannelPipeline$HeadHandler.flush(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeFlush()
io.netty.channel.DefaultChannelHandlerContext.flush()
io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeFlush()
io.netty.channel.DefaultChannelHandlerContext.flush()
io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeFlush()
io.netty.channel.DefaultChannelHandlerContext.flush()
io.netty.handler.logging.LoggingHandler.flush(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeFlush()
io.netty.channel.DefaultChannelHandlerContext.write(Object, boolean, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.writeAndFlush(Object, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.writeAndFlush(Object)
io.netty.channel.DefaultChannelPipeline.writeAndFlush(Object)
io.netty.channel.AbstractChannel.writeAndFlush(Object)
client.ClientHandler.channelActive(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeChannelActive()
io.netty.channel.DefaultChannelHandlerContext.fireChannelActive()
io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelHandlerContext)
io.netty.handler.logging.LoggingHandler.channelActive(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeChannelActive()
io.netty.channel.DefaultChannelHandlerContext.fireChannelActive()
io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeChannelActive()
io.netty.channel.DefaultChannelHandlerContext.fireChannelActive()
io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelHandlerContext)
io.netty.channel.DefaultChannelHandlerContext.invokeChannelActive()
io.netty.channel.DefaultChannelHandlerContext.fireChannelActive()
io.netty.channel.DefaultChannelPipeline.fireChannelActive()
io.netty.channel.oio.AbstractOioChannel$DefaultOioUnsafe.connect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelPipeline$HeadHandler.connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.invokeConnect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.ChannelOutboundHandlerAdapter.connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.invokeConnect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.ChannelOutboundHandlerAdapter.connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.invokeConnect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.ChannelDuplexHandler.connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)
io.netty.handler.logging.LoggingHandler.connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.invokeConnect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelHandlerContext.connect(SocketAddress, ChannelPromise)
io.netty.channel.DefaultChannelPipeline.connect(SocketAddress, ChannelPromise)
io.netty.channel.AbstractChannel.connect(SocketAddress, ChannelPromise)
io.netty.bootstrap.Bootstrap$2.run()
io.netty.channel.ThreadPerChannelEventLoop.run()
io.netty.util.concurrent.SingleThreadEventExecutor$2.run()
java.lang.Thread.run()
创建接收器通道的代码:
final class ServerChannelFactory {
private static final Logger LOGGER = Logger.getLogger(ServerChannelFactory.class);
protected static Channel createAcceptorChannel(
final ChannelType channelType,
final InetSocketAddress localAddress,
final ServerHandler serverHandler
) {
final ServerBootstrap serverBootstrap = ServerBootstrapFactory.createServerBootstrap(channelType);
serverBootstrap
.childHandler(new ServerChannelInitializer(serverHandler))
.option(ChannelOption.SO_BACKLOG, Resources.SO_BACKLOG);
try {
ChannelFuture channelFuture = serverBootstrap.bind(
new InetSocketAddress(localAddress.getPort())).sync();
channelFuture.awaitUninterruptibly();
if (channelFuture.isSuccess()) {
return channelFuture.channel();
} else {
LOGGER.warn(String.format("Failed to open socket! Cannot bind to port: %d!",
localAddress.getPort()));
}
} catch (InterruptedException e) {
LOGGER.error("Failed to create acceptor socket.", e);
}
return null;
}
private static class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
private ChannelHandler serverHandler;
private ServerChannelInitializer(ChannelHandler serverHandler) {
this.serverHandler = serverHandler;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// Encoders
ch.pipeline().addLast(Resources.STRING_ENCODER_NAME, new StringEncoder(CharsetUtil.UTF_8));
ch.pipeline().addBefore(Resources.STRING_ENCODER_NAME, Resources.FRAME_ENCODER_NAME,
new LengthFieldPrepender(Resources.FRAME_LENGTH_FIELD_SIZE));
// Decoders
ch.pipeline().addLast(Resources.STRING_DECODER_NAME, new StringDecoder(CharsetUtil.UTF_8));
ch.pipeline().addBefore(Resources.STRING_DECODER_NAME, Resources.FRAME_DECODER_NAME,
new LengthFieldBasedFrameDecoder(Resources.MAX_FRAME_LENGTH,
Resources.FRAME_LENGTH_FIELD_OFFSET, Resources.FRAME_LENGTH_FIELD_SIZE,
Resources.FRAME_LENGTH_ADJUSTMENT, Resources.FRAME_LENGTH_FIELD_SIZE));
// Handlers
ch.pipeline().addLast(Resources.LOGGING_HANDLER_NAME, new LoggingHandler());
ch.pipeline().addLast(Resources.SERVER_HANDLER_NAME, serverHandler);
}
}
}
服务器处理程序:
final class ServerHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOGGER = Logger.getLogger(ServerHandler.class);
int noMessagesReceived = 0;
@Override
public void channelActive(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
for(int i=0; i< Resources.NO_MESSAGES_TO_SEND; i++) {
ctx.channel().writeAndFlush(MessageStore.getMessage(i));
}
}
@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
noMessagesReceived++;
if(noMessagesReceived == Resources.NO_MESSAGES_TO_SEND) {
ctx.channel().writeAndFlush(MessageStore.getMessage(0));
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
FileUtils.write(Resources.SERVER_OUTPUT, String.format("Received %d messages", noMessagesReceived));
}
@Override
public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
LOGGER.error(String.format("Exception in %s", this.getClass().getName()), cause);
}
}
服务器引导工厂:
public class ServerBootstrapFactory {
private ServerBootstrapFactory() {
}
public static ServerBootstrap createServerBootstrap(final ChannelType channelType) throws UnsupportedOperationException {
ServerBootstrap serverBootstrap = new ServerBootstrap();
switch (channelType) {
case NIO:
serverBootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup());
serverBootstrap.channel(NioServerSocketChannel.class);
return serverBootstrap;
case OIO:
serverBootstrap.group(new OioEventLoopGroup(), new OioEventLoopGroup());
serverBootstrap.channel(OioServerSocketChannel.class);
return serverBootstrap;
default:
throw new UnsupportedOperationException("Failed to create ServerBootstrap, " + channelType + " not supported!");
}
}
}
创建连接器通道的代码:
final class ClientChannelFactory {
private static final Logger LOGGER = Logger.getLogger(ClientChannelFactory.class);
protected static Channel createConnectorChannel(
ChannelType channelType,
final ClientHandler clientHandler,
InetSocketAddress remoteAddress
) {
final Bootstrap bootstrap = BootstrapFactory.createBootstrap(channelType);
bootstrap.handler(new ClientChannelInitializer(clientHandler));
try {
final ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(remoteAddress.getAddress(), remoteAddress.getPort()))
.sync();
channelFuture.awaitUninterruptibly();
if (channelFuture.isSuccess()) {
return channelFuture.channel();
} else {
LOGGER.warn(String.format(
"Failed to open socket! Cannot connect to ip: %s port: %d!",
remoteAddress.getAddress(), remoteAddress.getPort())
);
}
} catch (InterruptedException e) {
LOGGER.error("Failed to open socket!", e);
}
return null;
}
private static class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
private ChannelHandler clientHandler;
private ClientChannelInitializer(ChannelHandler clientHandler) {
this.clientHandler = clientHandler;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// Encoders
ch.pipeline().addLast(Resources.STRING_ENCODER_NAME, new StringEncoder(CharsetUtil.UTF_8));
ch.pipeline().addBefore(Resources.STRING_ENCODER_NAME, Resources.FRAME_ENCODER_NAME,
new LengthFieldPrepender(Resources.FRAME_LENGTH_FIELD_SIZE));
// Decoders
ch.pipeline().addLast(Resources.STRING_DECODER_NAME, new StringDecoder(CharsetUtil.UTF_8));
ch.pipeline().addBefore(Resources.STRING_DECODER_NAME, Resources.FRAME_DECODER_NAME,
new LengthFieldBasedFrameDecoder(Resources.MAX_FRAME_LENGTH,
Resources.FRAME_LENGTH_FIELD_OFFSET, Resources.FRAME_LENGTH_FIELD_SIZE,
Resources.FRAME_LENGTH_ADJUSTMENT, Resources.FRAME_LENGTH_FIELD_SIZE));
// Handlers
ch.pipeline().addLast(Resources.LOGGING_HANDLER_NAME, new LoggingHandler());
ch.pipeline().addLast(Resources.CLIENT_HANDLER_NAME, clientHandler);
}
}
}
客户端处理程序:
public final class ClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOGGER = Logger.getLogger(ClientHandler.class);
private int noMessagesReceived = 0;
@Override
public void channelActive(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
for(int i=0; i< Resources.NO_MESSAGES_TO_SEND; i++) {
ctx.channel().writeAndFlush(MessageStore.getMessage(i));
}
}
@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
noMessagesReceived++;
if (noMessagesReceived > Resources.NO_MESSAGES_TO_SEND) {
ctx.channel().close();
}
}
@Override
public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
FileUtils.write(Resources.CLIENT_OUTPUT, String.format("Received %d messages", noMessagesReceived));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error(String.format("Exception in %s", this.getClass().getName()), cause);
}
}
Bootstrap 工厂:
public class BootstrapFactory {
private BootstrapFactory() {
}
public static Bootstrap createBootstrap(final ChannelType channelType) throws UnsupportedOperationException {
Bootstrap bootstrap = new Bootstrap();
switch (channelType) {
case NIO:
bootstrap.group(new NioEventLoopGroup());
bootstrap.channel(NioSocketChannel.class);
return bootstrap;
case OIO:
bootstrap.group(new OioEventLoopGroup());
bootstrap.channel(OioSocketChannel.class);
return bootstrap;
default:
throw new UnsupportedOperationException("Failed to create Bootstrap, " + channelType + " not supported!");
}
}
}
频道类型:
public enum ChannelType {
// New IO - non-blocking
NIO,
// Old IO - blocking
OIO;
}