Netty 基于http的文件上传

80 阅读 0 评论 53 点赞

我是靠谱客的博主土豪奇异果，最近开发中收集的这篇文章主要介绍Netty 基于http的文件上传，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

上传Handler如下

package com.netty.server;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.multipart.*;
import io.netty.util.CharsetUtil;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.Random;
public class DemoHttpUploadServerHandler extends ChannelInboundHandlerAdapter {
private HttpRequest request;
private HttpPostRequestDecoder decoder;
// 测试：本地测试请求路径,打开测试页面模拟上传
private static final String testUploadPage = "/test_upload_page";
// 上传地址：上传文件form表单提交地址
private static final String uploadFileAddress = "/post_multipart";
//TODO
保存的本地文件路径 可配置，及以上地址需整理到配置文件中
private static final String saveFilePath = "D:/nettyFile/";
private static final HttpDataFactory factory =
new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE); // Disk if size exceed
static {
DiskFileUpload.deleteOnExitTemporaryFile = true; // should delete file
DiskFileUpload.baseDirectory = null; // system temp directory
DiskAttribute.deleteOnExitTemporaryFile = true; // should delete file on
DiskAttribute.baseDirectory = null; // system temp directory
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
if (decoder != null) {
decoder.cleanFiles();
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object httpRequest) throws Exception {
FullHttpRequest msg = ((HttpJsonRequest)httpRequest).getRequest();
try{
if (msg instanceof HttpRequest) {
this.request = (HttpRequest) msg;
URI uri = new URI(request.uri());
System.out.println(uri);
urlForwarding(ctx, uri.getPath());
}
if (decoder != null) {
if (msg instanceof HttpContent) {
// 接收一个新的请求体
decoder.offer((HttpContent) msg);
// 将内存中的数据序列化本地
readHttpDataChunkByLoacl();
}
}
} catch (NullPointerException e){
e.printStackTrace();
}catch (HttpPostRequestDecoder.EndOfDataDecoderException ex) {
//ignore
// ex.printStackTrace();
}catch (Exception ex) {
ex.printStackTrace();
} finally {
if (msg instanceof LastHttpContent) {
reset();
writeResponseJson(ctx,"上传成功");
}
}
}
// url路由转发
private void urlForwarding(ChannelHandlerContext ctx, String uri) {
StringBuilder urlResponse = new StringBuilder();
// 访问文件上传页面
if (uri.startsWith(testUploadPage)) {
// 1. 测试访问页面
urlResponse.append(getUploadResponse());
writeResponseHtml(ctx, urlResponse.toString());
} else if (uri.startsWith(uploadFileAddress)) {
// 2. 解码器将解码POST BODY 请求数据
decoder = new HttpPostRequestDecoder(factory, request);
return;
}
}
/**
* 响应html页面
* @param ctx
* @param context
*/
private void writeResponseHtml(ChannelHandlerContext ctx, String context) {
ByteBuf buf = Unpooled.copiedBuffer(context, CharsetUtil.UTF_8);
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=utf-8");
//设置短连接 addListener 写完马上关闭连接
ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/**
* 返回json/text数据
* @param ctx
* @param status
* @param box
*/
private void writeResponseJson(ChannelHandlerContext ctx, String content) {
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,status, Unpooled.copiedBuffer(content, CharsetUtil.UTF_8));
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/**
* 将数据序列化到本地
* @throws IOException
*/
private void readHttpDataChunkByLoacl() throws IOException {
while (null !=decoder && decoder.hasNext()) {
InterfaceHttpData data = decoder.next();
if (data != null) {
if (data.getHttpDataType() == InterfaceHttpData.HttpDataType.FileUpload) {
FileUpload fileUpload = (FileUpload) data;
if (fileUpload.isCompleted()) {
fileUpload.isInMemory();
// 获取文件
//
File file = fileUpload.getFile();
//
FileInputStream fileInputStream = new FileInputStream(file);
String filename = fileUpload.getFilename();
// 截取后缀名
String suffixName = "";
if(filename.contains(".")) {
suffixName = filename.substring(filename.lastIndexOf("."));
} else {
suffixName = filename;
}
//TODO 重命名文件名称
fileUpload.setFilename(new Random().nextInt(9999)+"" + suffixName);
// 将重命名后的文件存放于 saveFilePath 路径下
fileUpload.renameTo(new File(saveFilePath + File.separator + fileUpload.getFilename()));
decoder.removeHttpDataFromClean(fileUpload);
}
}
}
}
}
private void reset() {
request = null;
decoder.destroy();
decoder = null;
}
/**
* 测试页面，错误页
* @return
*/
private String getHomeResponse() {
return " <h1> welcome home </h1> ";
}
/**
* 测试页面，上传页
* @return
*/
private String getUploadResponse() {
return "<!DOCTYPE html>n" +
"<html lang="en">n" +
"<head>n" +
"
<meta charset="UTF-8">n" +
"
<title>Title</title>n" +
"</head>n" +
"<body>n" +
"n" +
"<form action="http://127.0.0.1:9090/post_multipart" enctype="multipart/form-data" method="POST">n" +
"n" +
"n" +
"
<input type="file" name=" +
" " +
"" +
""YOU_KEY">n" +
"n" +
"
<input type="submit" name="send">n" +
"n" +
"</form>n" +
"n" +
"</body>n" +
"</html>";
}
}

server端如下

static final boolean SSL = System.getProperty("ssl") != null;
//服务端启动run方法
public static void run(int port) throws Exception{
final SslContext sslCtx;
if (SSL) {
SelfSignedCertificate ssc = new SelfSignedCertificate();
sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
} else {
sslCtx = null;
}
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss,worker)
.channel(NioServerSocketChannel.class)
.childHandler(
new NettyServerInitHandler(sslCtx)
).childOption(ChannelOption.SO_KEEPALIVE,true);// 开启keepalive
ChannelFuture cf = bootstrap.bind(port).sync();
System.out.println("本地开放PORT:" + port);
cf.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
} finally {
//关闭loopGroup
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
public static void main(String[] args) {
if (WorkFactory.getStatus() != OsType_Enum.RUN.getOsType()) {
WorkFactory.start(Config.class.getName());
}
try {
// 配置启动端口
NettyHttpServer.run(9090);
} catch (Exception e) {
e.printStackTrace();
}
}

initHandler

import com.netty.server.DemoHttpUploadServerHandler;
import com.netty.server.codec.DemoHttpJsonRequestDecoder;
import com.netty.server.codec.DemoHttpJsonResponseEncoder;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
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.ssl.SslContext;
import io.netty.handler.stream.ChunkedWriteHandler;
public class NettyServerInitHandler extends ChannelInitializer<SocketChannel> {
private final SslContext sslCtx;
public NettyServerInitHandler(SslContext sslCtx) {
this.sslCtx = sslCtx;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
if (sslCtx != null) {
ch.pipeline().addLast(sslCtx.newHandler(ch.alloc()));
}
ch.pipeline().addLast("http-decoder",new HttpRequestDecoder());//httpRequest解码器
ch.pipeline().addLast("http-aggregator", new HttpObjectAggregator(Integer.MAX_VALUE));
ch.pipeline().addLast("http-encoder", new HttpResponseEncoder());
ch.pipeline().addLast("content-compressor",new HttpContentCompressor());// 默认压缩级别的处理器
ch.pipeline().addLast("chunked-write", new ChunkedWriteHandler());//
大数据流的支持
ch.pipeline().addLast("http-upload", new DemoHttpUploadServerHandler());// 上传文件处理handler
ch.pipeline().addLast("apiFilterHandler", new ApiFilterHandler());
}
}