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

Question

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

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

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

我在这里做错了什么？

记录

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]

客户

FileRequestHandler.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读/写协作。它可能起作用，或者不起作用…