java - 使用 Netty.io 将文件从服务器发送到客户端

Question

我正在尝试发送一个文件，一个从服务器请求的客户端，从服务器到客户端。客户端在FileRequestProtocol中指定文件，将其发送给服务器，服务器将文件大小添加到文件FileRequestProtocol中并返回给客户端。

客户端向其管道添加一个具有正确文件大小的新 FileChunkReqWriteHandler。

服务器使用上下文和所需文件创建一个新的 ChunkedFileServerHandler 并尝试发送它，但 FileChunkReqWriteHandler 从未从 channel 读取字节。

我在这里做错了什么？

日志

INFO  ProtocolHeadHandler:48 - Client send ProtocolHead [version=1, jobType=FILEREQUEST]
INFO  ProtocolHeadServerHandler:36 - Server receive ProtocolHead [version=1, jobType=FILEREQUEST]
INFO  ProtocolHeadHandler:57 - Client ProtocolHead equals, Send Protocol FileRequestProtocol [filePath=test.jpg, fileSize=0]
INFO  FileRequestServerHandler:42 - Server new FileRequest FileRequestProtocol [filePath=test.jpg, fileSize=0]
INFO  FileRequestHandler:41 - Client receives FileRequestProtocol [filePath=test.jpg, fileSize=174878]
INFO  ChunkedFileServerHandler:39 - New ChunkedFileServerHandler
INFO  FileChunkReqWriteHandler:20 - New ChunkedFile Handler FileRequestProtocol [filePath=test.jpg, fileSize=174878]

客户端

文件请求处理程序.java

public class FileRequestHandler extends
    SimpleChannelInboundHandler<FileRequestProtocol> {

private Logger logger = Logger.getLogger(this.getClass());

public FileRequestHandler() {
}

@Override
public void channelRead0(ChannelHandlerContext ctx, FileRequestProtocol msg) {
    logger.info("Client receives " + msg);
    ReferenceCountUtil.release(msg);
    ctx.channel().pipeline().addLast(new FileChunkReqWriteHandler(msg));
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
    logger.info("Client read complete");
    ctx.flush();
}

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

FileChunkReqWriteHandler.java

public class FileChunkReqWriteHandler extends SimpleChannelInboundHandler<ChunkedFile> {

FileRequestProtocol fileRequestProtocol;
private Logger logger = Logger.getLogger(this.getClass());


public FileChunkReqWriteHandler(FileRequestProtocol msg) {
    this.fileRequestProtocol = msg;
    logger.info("New ChunkedFile Handler " + msg);
}

@Override
public void channelActive(ChannelHandlerContext ctx) {
    logger.info("in channel active method");
}

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

    if (ctx.channel().isActive()) {
        ctx.writeAndFlush("ERR: " +
                cause.getClass().getSimpleName() + ": " +
                cause.getMessage() + '\n').addListener(ChannelFutureListener.CLOSE);
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, ChunkedFile msg)
        throws Exception {
    logger.info("in channelRead0");

}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    logger.info("channelRead");
    ByteBuf buf = (ByteBuf) msg;
    byte[] bytes = new byte[buf.readableBytes()];
    buf.readBytes(bytes);
    if(buf.readableBytes() >= this.fileRequestProtocol.getFileSize())
    {
        logger.info("received all data");
    }
}
}

服务器

FileRequestServerHandler.java

public class FileRequestServerHandler extends
    SimpleChannelInboundHandler<FileRequestProtocol> {

private File f;
private Logger logger = Logger.getLogger(this.getClass());

@Override
public void channelRead0(ChannelHandlerContext ctx, FileRequestProtocol fileRequest) {
    logger.info("Server new FileRequest " + fileRequest);
    f = new File(fileRequest.getFilePath());
    fileRequest.setFileSize(f.length());
    ctx.writeAndFlush(fileRequest);

    new ChunkedFileServerHandler(ctx,f);
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
    logger.info("Server read complete");

}

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

ChunkedFileServerHandler.java

public class ChunkedFileServerHandler extends ChunkedWriteHandler {

private Logger logger = Logger.getLogger(this.getClass());

private File file;
public ChunkedFileServerHandler(ChannelHandlerContext ctx, File file) {
    this.file = file;

    logger.info("New ChunkedFileServerHandler");
    ChunkedFile chunkedFile;
    try {
        chunkedFile = new ChunkedFile(this.file);
        ctx.writeAndFlush(chunkedFile);
        ctx.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}


@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    super.channelActive(ctx);
    logger.info("FILE WRITE GETS ACTIVE");
}

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

更新

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

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

    p.addLast("encoder", new ObjectEncoder());
    p.addLast("decoder",
            new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
    p.addLast("protocolhead", new ProtocolHeadServerHandler());
    p.addLast("filerequestserverhandler", new FileRequestServerHandler());
    p.addLast("chunkedfileserver", new ChunkedFileServerHandler());

}

}

服务器启动

public void startUp()
{
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ServerInitializer());

            // Bind and start to accept incoming connections.
            b.bind(this.port).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
}

Answer 1

我可以看到两个部分:

1) 你不应该在你自己的处理程序中创建一个新的处理程序，而是直接创建 ChunkedFile 并写入它:

public class FileRequestServerHandler extends
    SimpleChannelInboundHandler<FileRequestProtocol> {

    private File f;
    private Logger logger = Logger.getLogger(this.getClass());

    @Override
    public void channelRead0(ChannelHandlerContext ctx, FileRequestProtocol fileRequest) {
        logger.info("Server new FileRequest " + fileRequest);
        f = new File(fileRequest.getFilePath());
        fileRequest.setFileSize(f.length());
        ctx.writeAndFlush(fileRequest);

        // directly make your chunkedFile there instead of creating a sub handler
        chunkedFile = new ChunkedFile(this.file);
        ctx.writeAndFlush(chunkedFile);// need a specific handler
        // Don't create such an handler: new ChunkedFileServerHandler(ctx,f);
}

2) 由于您使用 ChunkedInput(此处为 ChunkedFile)编写，因此您的管道中必须在处理程序之前有一个 ChunkedWriteHandler，因此您的 Initializer 可能如下所示:

public class ServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();

        p.addLast("encoder", new ObjectEncoder());
        p.addLast("decoder",
            new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
        p.addLast("chunkedWriteHandler", new ChunkedWriteHandler());// added
        p.addLast("protocolhead", new ProtocolHeadServerHandler());
        p.addLast("filerequestserverhandler", new FileRequestServerHandler());
        // removed: p.addLast("chunkedfileserver", new ChunkedFileServerHandler());
    }
}

ChunkedWriteHandler 的位置可以更改，但始终在您编写 ChunkedFile 的处理程序之前。

3) 最后注意事项:查看并处理您的编码器/解码器 (ObjectEncoder/ObjectDecoder)，因为我不能 100% 确定他们是否可以与文件中的 ByteBuf 的这种写入/读取协作。它可能有效，也可能无效...