添加 decoder/encoder 时 Netty 管道损坏

Question

我最近开始在我的一个项目中使用 netty。我实现了 HexDumpProxy 示例以了解 netty 的工作原理。当我在通道管道中添加 StringDecoder 和 StringEncoder 导致管道损坏时，我遇到了一个问题。如果 decoder/encoder 不存在，程序将正常运行。有人可以向我解释为什么会这样吗？非常感谢任何帮助！

下面我正在添加代码。

主要class:

public final class HexDumpProxy {

public static void main(String[] args) throws Exception {

    // Configure the bootstrap.
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .handler(new LoggingHandler(LogLevel.INFO))
         .childHandler(new HexDumpProxyInitializer("127.0.0.1", 9000))
         .childOption(ChannelOption.AUTO_READ, false)
         .bind(8000).sync().channel().closeFuture().sync();
    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        }
    }
}

初始化器class:

public class HexDumpProxyInitializer extends ChannelInitializer<SocketChannel> {

private final String remoteHost;
private final int remotePort;

public HexDumpProxyInitializer (String remoteHost, int remotePort) {
    this.remoteHost = remoteHost;
    this.remotePort = remotePort;
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast("frameDecoder", new LineBasedFrameDecoder(80));
    ch.pipeline().addLast("decoder", new StringDecoder());
    ch.pipeline().addLast("encoder", new StringEncoder());
    //ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
    ch.pipeline().addLast(new HexDumpProxyFrontendHandler(remoteHost, remotePort));
    }

}

前端处理程序class:

public class HexDumpProxyFrontendHandler extends ChannelInboundHandlerAdapter {

private final String remoteHost;
private final int remotePort;

private Channel serverChannel;

public HexDumpProxyFrontendHandler(String remoteHost, int remotePort) {
    this.remoteHost = remoteHost;
    this.remotePort = remotePort;
}

@Override
public void channelActive(ChannelHandlerContext ctx) {
    final Channel clientChannel = ctx.channel();

    Bootstrap b = new Bootstrap();
    b.group(clientChannel.eventLoop())
     .channel(ctx.channel().getClass())
     .handler(new HexDumpProxyBackendHandler(clientChannel))
     .option(ChannelOption.AUTO_READ, false);

    ChannelFuture f = b.connect(remoteHost, remotePort);
    serverChannel = f.channel();
    f.addListener(new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                clientChannel.read();
            }
            else
                clientChannel.close();
            }

        });
}

@Override
public void channelRead (final ChannelHandlerContext ctx, Object msg) {
    if (serverChannel.isActive()) {
        System.out.println("************" + msg);
        serverChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) {
                if (future.isSuccess()) {
                    ctx.channel().read();
                    System.out.println("Read from server channel. - channelRead");
                } else {
                    future.channel().close();
                    System.out.println("Close server channel.");
                }
            }
        });
    }
}

@Override
public void channelInactive (ChannelHandlerContext ctx) {
    if (serverChannel != null) {
        closeOnFlush(serverChannel);
        System.out.println("close on flush - server channel");
    }
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    ctx.close();
}

static void closeOnFlush (Channel ch) {
    if (ch.isActive())
        ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ctx.flush();
    ctx.channel().read();
    }
}

后端处理程序class:

public class HexDumpProxyBackendHandler extends ChannelInboundHandlerAdapter {

private final Channel clientChannel;

public HexDumpProxyBackendHandler (Channel clientChannel) {
    this.clientChannel = clientChannel;
}

@Override
public void channelActive (ChannelHandlerContext ctx) {
    ctx.channel().read();
    System.out.println("Read from client channel. - channelActive");
}

@Override
public void channelRead (final ChannelHandlerContext ctx, Object msg) {
    System.out.println("~~~~~~~~~~~" + msg);
    clientChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                clientChannel.read();
                System.out.println("Read from client channel. - channelRead");
            } else {
                clientChannel.close();
                System.out.println("Close client channel.");
                }
            }
        });
}

@Override
public void channelInactive (ChannelHandlerContext ctx) {
    HexDumpProxyFrontendHandler.closeOnFlush(clientChannel);
    System.out.println("close on flush - client channel");
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    ctx.close();
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ctx.flush();
    ctx.channel().read();
    }
}

Answer 1

对于以后遇到类似问题的大家，我找到了问题所在。

在 class HexDumpProxyFrontendHandler 中而不是在 bootstrap 期间使用新的 HexDumpProxyBackendHandler(clientChannel) 作为处理程序，创建一个新的 class 例如HexDumpProxyBackendInitializer(clientChannel) 和 class 以与 class HexDumpProxyInitializer.

相同的方式初始化管道

希望这可以帮助到别人！

添加 decoder/encoder 时 Netty 管道损坏

Netty pipeline broken when adding decoder/encoder

java

netty