java - 使用 netty 的二进制聊天

Question

我正在尝试修改 netty 的安全聊天示例以发送字节(字节数组)而不是字符串。但是我无法向服务器发送任何字节。我究竟做错了什么？如果使用 ObjectDecoder/Encoder，它工作得很好，但我需要通过线路发送原始字节。

如果取而代之的是一个 byte[]，一个 ByteBuffer 也足够了，只要流量只包含缓冲区中的那些字节。

有人可以帮忙吗？

客户端.java

public class Client {

private final String host;
private final int port;

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

public void run() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(NioSocketChannel.class)
         .handler(new ClientInitializer());
        Channel ch = b.connect(host, port).sync().channel();
        ChannelFuture lastWriteFuture = null;
        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
        for (;;) {
            String line = in.readLine();
            if (line == null) {
                break;
            }

            lastWriteFuture = ch.write(new byte[]{1,2,3,4,5,6,7,8});

        }
        if (lastWriteFuture != null) {
            lastWriteFuture.sync();
        }
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    new Client("localhost",6666).run();
}
}

客户端处理程序

public class ClientHandler extends SimpleChannelInboundHandler<Byte[]> {

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

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    logger.log(
            Level.WARNING,
            "Unexpected exception from downstream.", cause);
    ctx.close();
}

@Override
protected void channelRead0(ChannelHandlerContext chc, Byte[] i) throws Exception {
   System.out.println(i[3]);
}
}

客户端初始化器

public class ClientInitializer extends ChannelInitializer<SocketChannel> {

@Override
public 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());

    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
    pipeline.addLast("bytesEncoder", new ByteArrayEncoder());
    pipeline.addLast("handler", new ClientHandler());
}
}

服务器

public class Server {
private final int port;
public Server(int port) {
    this.port = port;
}
public void run() throws InterruptedException {
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .childHandler(new ServerInitializer());

        b.bind(port).sync().channel().closeFuture().sync();
    } finally {
        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 = 6666;
    }
    new Server(port).run();
}
}

服务器处理器

public class ServerHandler extends SimpleChannelInboundHandler<Byte[]> {
private static final Logger logger = Logger.getLogger(
        ServerHandler.class.getName());

static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
            channels.add(ctx.channel());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    logger.log(
            Level.WARNING,
            "Unexpected exception from downstream.", cause);
    ctx.close();
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Byte[] i) throws Exception {
    for (Channel c: channels) {
        if (c != ctx.channel()) {
            c.writeAndFlush(i);
        } 
    }

}
}

服务器初始化

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

@Override
public 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());
    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
    pipeline.addLast("bytesEncoder", new ByteArrayEncoder());
    pipeline.addLast("handler", new ServerHandler());
 }
}

Answer 1

您忘记在 channel.write(new byte[] { 1, 2, 3, 4, 5, 6, 7, 8 之后在客户端中调用 channel.flush() })。因为它从未被刷新，所以服务器永远不会收到它。或者，您可以使用 writeAndFlush()，它是 write(...) 和 flush() 的快捷方式。