本文共 12645 字，大约阅读时间需要 42 分钟。

本节我们使用netty来实现文件下载，netty的文件下载并不是通过建立长连接来传输下载的，而是通过分片chunked模式下载。

import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.http.HttpObjectAggregator;import io.netty.handler.codec.http.HttpRequestDecoder;import io.netty.handler.codec.http.HttpResponseEncoder;import io.netty.handler.stream.ChunkedWriteHandler;public class HttpFileServer {    private static final String DEFAULT_URL = "/sources/";    public void run(final int port, final String url) throws Exception {    	EventLoopGroup bossGroup = new NioEventLoopGroup();    	EventLoopGroup workerGroup = new NioEventLoopGroup();		try {		    ServerBootstrap b = new ServerBootstrap();		    b.group(bossGroup, workerGroup)			    .channel(NioServerSocketChannel.class)			    .childHandler(new ChannelInitializer
   
    () {				@Override				protected void initChannel(SocketChannel ch)					throws Exception {					// 加入http的解码器				    ch.pipeline().addLast("http-decoder", new HttpRequestDecoder());				    // 加入ObjectAggregator解码器，作用是他会把多个消息转换为单一的FullHttpRequest或者FullHttpResponse				    ch.pipeline().addLast("http-aggregator", new HttpObjectAggregator(65536));				    // 加入http的编码器				    ch.pipeline().addLast("http-encoder", new HttpResponseEncoder());				    // 加入chunked 主要作用是支持异步发送的码流（大文件传输），但不专用过多的内存，防止java内存溢出				    ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());				    // 加入自定义处理文件服务器的业务逻辑handler				    ch.pipeline().addLast("fileServerHandler",					    new HttpFileServerHandler(url));				}			    });		    ChannelFuture future = b.bind("192.168.109.1", port).sync();		    System.out.println("HTTP文件目录服务器启动，网址是 : " + "http://192.168.109.1:"  + port + url);		    future.channel().closeFuture().sync();		} finally {		    bossGroup.shutdownGracefully();		    workerGroup.shutdownGracefully();		}    }    public static void main(String[] args) throws Exception {		int port = 8765;		String url = DEFAULT_URL;		new HttpFileServer().run(port, url);    }}
   

HttpObjectAggregator：

参数(int maxContentLength)：每次最大上传或者下载分片的数据大小。

它负责把多个HttpMessage组装成一个完整的Http请求或者响应。到底是组装成请求还是响应，则取决于它所处理的内容是请求的内容，还是响应的内容。这其实可以通过Inbound和Outbound来判断，对于Server端而言，在Inbound 端接收请求，在Outbound端返回响应。

如果Server向Client返回的数据指定的传输编码是 chunked。则，Server不需要知道发送给Client的数据总长度是多少，它是通过分块发送的。

注意，HttpObjectAggregator通道处理器必须放到HttpRequestDecoder或者HttpRequestEncoder后面。

ChunkedWriteHandler：

该通道处理器主要是为了处理大文件传输的情形。大文件传输时，需要复杂的状态管理，而ChunkedWriteHandler实现这个功能。

我们来看下具体的HttpFileServerHandler，文件下载服务处理器：

import static io.netty.handler.codec.http.HttpHeaderNames.*;import static io.netty.handler.codec.http.HttpMethod.*;import static io.netty.handler.codec.http.HttpResponseStatus.*;import static io.netty.handler.codec.http.HttpVersion.*;import static io.netty.handler.codec.http.HttpMethod.GET;import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;import static io.netty.handler.codec.http.HttpResponseStatus.FOUND;import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;import static io.netty.handler.codec.http.HttpResponseStatus.OK;import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelProgressiveFuture;import io.netty.channel.ChannelProgressiveFutureListener;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.codec.http.DefaultFullHttpResponse;import io.netty.handler.codec.http.DefaultHttpResponse;import io.netty.handler.codec.http.FullHttpRequest;import io.netty.handler.codec.http.FullHttpResponse;import io.netty.handler.codec.http.HttpHeaderUtil;import io.netty.handler.codec.http.HttpHeaderValues;import io.netty.handler.codec.http.HttpHeaders;import io.netty.handler.codec.http.HttpResponse;import io.netty.handler.codec.http.HttpResponseStatus;import io.netty.handler.codec.http.LastHttpContent;import io.netty.handler.stream.ChunkedFile;import io.netty.util.CharsetUtil;import java.io.File;import java.io.FileNotFoundException;import java.io.RandomAccessFile;import java.io.UnsupportedEncodingException;import java.net.URLDecoder;import java.util.regex.Pattern;import javax.activation.MimetypesFileTypeMap;public class HttpFileServerHandler extends SimpleChannelInboundHandler
   
     {    	private final String url;    public HttpFileServerHandler(String url) {    	this.url = url;    }    @Override    public void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {    	//对请求的解码结果进行判断：		if (!request.decoderResult().isSuccess()) {			// 400		    sendError(ctx, BAD_REQUEST);		    return;		}		//对请求方式进行判断：如果不是get方式（如post方式）则返回异常		if (request.method() != GET) {			// 405		    sendError(ctx, METHOD_NOT_ALLOWED);		    return;		}		//获取请求uri路径		final String uri = request.uri();		//对url进行分析，返回本地系统		final String path = sanitizeUri(uri);		//如果 路径构造不合法，则path为null		if (path == null) {			//403		    sendError(ctx, FORBIDDEN);		    return;		}		// 创建file对象		File file = new File(path);		// 判断文件是否为隐藏或者不存在		if (file.isHidden() || !file.exists()) {			// 404 		    sendError(ctx, NOT_FOUND);		    return;		}		// 如果为文件夹		if (file.isDirectory()) {		    if (uri.endsWith("/")) {		    	//如果以正常"/"结束 说明是访问的一个文件目录：则进行展示文件列表（web服务端则可以跳转一个Controller，遍历文件并跳转到一个页面）		    	sendListing(ctx, file);		    } else {		    	//如果非"/"结束 则重定向，补全"/" 再次请求		    	sendRedirect(ctx, uri + '/');		    }		    return;		}		// 如果所创建的file对象不是文件类型		if (!file.isFile()) {			// 403		    sendError(ctx, FORBIDDEN);		    return;		}				//随机文件读写类		RandomAccessFile randomAccessFile = null;		try {		    randomAccessFile = new RandomAccessFile(file, "r");// 以只读的方式打开文件		} catch (FileNotFoundException fnfe) {			// 404		    sendError(ctx, NOT_FOUND);		    return;		}				//获取文件长度		long fileLength = randomAccessFile.length();		//建立响应对象		HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);		//设置响应信息		HttpHeaderUtil.setContentLength(response, fileLength);		//设置响应头		setContentTypeHeader(response, file);		//如果一直保持连接则设置响应头信息为：HttpHeaders.Values.KEEP_ALIVE		if (HttpHeaderUtil.isKeepAlive(request)) {		    response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);		}		//进行写出		ctx.write(response);				//构造发送文件线程，将文件写入到Chunked缓冲区中		ChannelFuture sendFileFuture;		//写出ChunkedFile		sendFileFuture = ctx.write(new ChunkedFile(randomAccessFile, 0, fileLength, 8192), ctx.newProgressivePromise());		//添加传输监听		sendFileFuture.addListener(new ChannelProgressiveFutureListener() {		    @Override		    public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {				if (total < 0) { // total unknown				    System.err.println("Transfer progress: " + progress);				} else {				    System.err.println("Transfer progress: " + progress + " / " + total);				}		    }		    @Override		    public void operationComplete(ChannelProgressiveFuture future) throws Exception {		    	System.out.println("Transfer complete.");		    }		});				//如果使用Chunked编码，最后则需要发送一个编码结束的看空消息体，进行标记，表示所有消息体已经成功发送完成。		ChannelFuture lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);		//如果当前连接请求非Keep-Alive ，最后一包消息发送完成后 服务器主动关闭连接		if (!HttpHeaderUtil.isKeepAlive(request)) {		    lastContentFuture.addListener(ChannelFutureListener.CLOSE);		}    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {    	//cause.printStackTrace();		if (ctx.channel().isActive()) {		    sendError(ctx, INTERNAL_SERVER_ERROR);		    ctx.close();		}    }    //非法URI正则    private static final Pattern INSECURE_URI = Pattern.compile(".*[<>&\"].*");    /**     * 
    方法名称：解析URI
    
     * 
    概要说明：对URI进行分析
    
     * @param uri netty包装后的字符串对象     * @return path 解析结果     */    private String sanitizeUri(String uri) {		try {			//使用UTF-8字符集		    uri = URLDecoder.decode(uri, "UTF-8");		} catch (UnsupportedEncodingException e) {		    try {		    	//尝试ISO-8859-1		    	uri = URLDecoder.decode(uri, "ISO-8859-1");		    } catch (UnsupportedEncodingException e1) {		    	//抛出预想外异常信息		    	throw new Error();		    }		}		// 对uri进行细粒度判断：4步验证操作		// step 1 基础验证		if (!uri.startsWith(url)) {		    return null;		}		// step 2 基础验证		if (!uri.startsWith("/")) {		    return null;		}		// step 3 将文件分隔符替换为本地操作系统的文件路径分隔符		uri = uri.replace('/', File.separatorChar);		// step 4 二次验证合法性		if (uri.contains(File.separator + '.')			|| uri.contains('.' + File.separator) || uri.startsWith(".")			|| uri.endsWith(".") || INSECURE_URI.matcher(uri).matches()) {		    return null;		}		//当前工程所在目录 + URI构造绝对路径进行返回 		return System.getProperty("user.dir") + File.separator + uri;    }        //文件是否被允许访问下载验证    private static final Pattern ALLOWED_FILE_NAME = Pattern.compile("[A-Za-z0-9][-_A-Za-z0-9\\.]*");    private static void sendListing(ChannelHandlerContext ctx, File dir) {    	// 设置响应对象		FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK);		// 响应头		response.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");		// 追加文本内容		StringBuilder ret = new StringBuilder();		String dirPath = dir.getPath();		ret.append("\r\n");		ret.append("
    \r\n");		ret.append("
    
");		ret.append(dirPath).append(" 目录：");		ret.append("
\r\n");		ret.append("
    
     ");		ret.append("
     
链接：..
\r\n");				// 遍历文件 添加超链接		for (File f : dir.listFiles()) {			//step 1: 跳过隐藏或不可读文件 		    if (f.isHidden() || !f.canRead()) {		    	continue;		    }		    String name = f.getName();		    //step 2: 如果不被允许，则跳过此文件		    if (!ALLOWED_FILE_NAME.matcher(name).matches()) {		    	continue;		    }		    //拼接超链接即可		    ret.append("
     
链接：");		    ret.append(name);		    ret.append("
\r\n");		}		ret.append("
    
\r\n");		//构造结构，写入缓冲区		ByteBuf buffer = Unpooled.copiedBuffer(ret, CharsetUtil.UTF_8);		//进行写出操作		response.content().writeBytes(buffer);		//重置写出区域		buffer.release();		//使用ctx对象写出并且刷新到SocketChannel中去 并主动关闭连接(这里是指关闭处理发送数据的线程连接)		ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);    }    //重定向操作    private static void sendRedirect(ChannelHandlerContext ctx, String newUri) {    	//建立响应对象		FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, FOUND);		//设置新的请求地址放入响应对象中去		response.headers().set(LOCATION, newUri);		//使用ctx对象写出并且刷新到SocketChannel中去 并主动关闭连接(这里是指关闭处理发送数据的线程连接)		ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);    }    //错误信息    private static void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {    	//建立响应对象    	FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status.toString()+ "\r\n", CharsetUtil.UTF_8));    	//设置响应头信息    	response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");    	//使用ctx对象写出并且刷新到SocketChannel中去 并主动关闭连接(这里是指关闭处理发送数据的线程连接)    	ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);    }    private static void setContentTypeHeader(HttpResponse response, File file) {    	//使用mime对象获取文件类型    	MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();    	response.headers().set(CONTENT_TYPE, mimeTypesMap.getContentType(file.getPath()));    }}
   

分析：通过一系列校验处理，把目录下的文件封装成html代码输出。

当点击链接后，向netty文件服务器发起下载请求，分多次下载，每次下载8192的字节。

//构造发送文件线程，将文件写入到Chunked缓冲区中		ChannelFuture sendFileFuture;		//写出ChunkedFile		sendFileFuture = ctx.write(new ChunkedFile(randomAccessFile, 0, fileLength, 8192), ctx.newProgressivePromise());		//添加传输监听		sendFileFuture.addListener(new ChannelProgressiveFutureListener() {		    @Override		    public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {				if (total < 0) { // total unknown				    System.err.println("Transfer progress: " + progress);				} else {				    System.err.println("Transfer progress: " + progress + " / " + total);				}		    }		    @Override		    public void operationComplete(ChannelProgressiveFuture future) throws Exception {		    	System.out.println("Transfer complete.");		    }		});				//如果使用Chunked编码，最后则需要发送一个编码结束的看空消息体，进行标记，表示所有消息体已经成功发送完成。		ChannelFuture lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);

转载地址：https://jeffsheng.blog.csdn.net/article/details/80849575 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！