无法使用Netty向服务器发送字节[]

我对荨麻很陌生。我目前正在编写一个非常简单的Echo客户端和服务器,但我尝试发送 byte[] 从客户端到服务器。我成功地用字符串消息编写了同一个客户机\服务器。

在相同的代码中,我只是将“解码器\编码器”更改为消费 字节[] .

我在下面发布我的代码。

EchoClient.java -启动客户端的主类

public class EchoClient {

    private final String host;
    private final int port;
    private final int firstMessageSize;

    public EchoClient(String host, int port, int firstMessageSize) {
        this.host = host;
        this.port = port;
        this.firstMessageSize = firstMessageSize;
    }

    public void run() throws Exception {
        // Configure the client.
        String firstTestMessage = "Avinash \r\n";
        byte[] bytes = firstTestMessage.getBytes("UTF-8");

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new EchoClientByteHandler(bytes));

            // Start the client.
            ChannelFuture f = b.connect(host, port).sync();

            // Wait until the connection is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down the event loop to terminate all threads.
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        // Print usage if no argument is specified.

        args = new String[2];
        args[0] = "localhost";
        args[1] = "9090";
        if (args.length < 2 || args.length > 3) {
            System.err.println("Usage: " + EchoClient.class.getSimpleName()
                    + " <host> <port> [<first message size>]");
            return;
        }

        // Parse options.
        final String host = args[0];
        final int port = Integer.parseInt(args[1]);
        final int firstMessageSize;
        if (args.length == 3) {
            firstMessageSize = Integer.parseInt(args[2]);
        } else {
            firstMessageSize = 5;
        }

        new EchoClient(host, port, firstMessageSize).run();
    }
}

EchoClientByteInitialzer -定义管道

public class EchoClientByteInitialzer extends ChannelInitializer<SocketChannel>{

    private byte[] mBytes = null;

    public EchoClientByteInitialzer(byte[] bytes)
    {
        this.mBytes = bytes;
    }

    @Override
    protected void initChannel(SocketChannel ch) throws Exception
    {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
        pipeline.addLast("bytesDecoder", new ByteArrayDecoder());

        // Encoder
        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
        pipeline.addLast("bytesEncoder", new ByteArrayEncoder());

        pipeline.addLast("handler", new EchoClientByteHandler(mBytes));
    }

}

EchoClientByteHandler -主客户端处理程序

public class EchoClientByteHandler extends ChannelInboundHandlerAdapter {
    private byte[] mBytes = null;

    public EchoClientByteHandler(byte[] bytes) {
        this.mBytes = bytes;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(mBytes);
        System.out.println(mBytes);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        ctx.write(msg);
        System.out.println(msg);
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

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

EchoServer -要启动服务器的主类

public class EchoServer {

    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {
        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new EchoServerByteInitialzer());

            // Start the server.
            ChannelFuture f = b.bind(port).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 9090;
        }
        new EchoServer(port).run();
    }
}

EchoServerInitializer -服务器管道

public class EchoServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel arg0) throws Exception {
        ChannelPipeline ch = arg0.pipeline();

        ch.addLast(
                "farmer",
                new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));

        ch.addLast("decoder", new StringDecoder());

        ch.addLast("encoder", new StringEncoder());

        ch.addLast("handler",new EchoServerHandler());

    }

}

EchoServerHandler -主服务器处理程序

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    private static final Logger logger = Logger
            .getLogger(EchoServerHandler.class.getName());

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {

        try {
            ctx.write(msg);
            System.out.println(msg);

        } finally {

        }
    }

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

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        logger.log(Level.WARNING, "Unexpected exception from downstream.",
                cause);
        ctx.close();
    }
}

请告诉我这里缺少什么。