厲害了，Netty 輕松實(shí)現(xiàn)文件上傳！
出處：www.cnblogs.com/rickiyang/p/11074222.html
今天我們來(lái)完成一個(gè)使用netty進(jìn)行文件傳輸?shù)娜蝿?wù)。在實(shí)際項(xiàng)目中，文件傳輸通常采用FTP或者HTTP附件的方式。事實(shí)上通過(guò)TCP Socket+File的方式進(jìn)行文件傳輸也有一定的應(yīng)用場(chǎng)景，盡管不是主流，但是掌握這種文件傳輸方式還是比較重要的，特別是針對(duì)兩個(gè)跨主機(jī)的JVM進(jìn)程之間進(jìn)行持久化數(shù)據(jù)的相互交換。
而使用netty來(lái)進(jìn)行文件傳輸也是利用netty天然的優(yōu)勢(shì)：零拷貝功能。很多同學(xué)都聽(tīng)說(shuō)過(guò)netty的”零拷貝”功能，但是具體體現(xiàn)在哪里又不知道，下面我們就簡(jiǎn)要介紹下：
Netty的“零拷貝”主要體現(xiàn)在如下三個(gè)方面：
具體的分析在此就不多做介紹，有興趣的可以查閱相關(guān)文檔。我們還是把重點(diǎn)放在文件傳輸上。Netty作為高性能的服務(wù)器端異步IO框架必然也離不開(kāi)文件讀寫(xiě)功能，我們可以使用netty模擬http的形式通過(guò)網(wǎng)頁(yè)上傳文件寫(xiě)入服務(wù)器，當(dāng)然要使用http的形式那你也用不著netty！大材小用。
netty4中如果想使用http形式上傳文件你還得借助第三方j(luò)ar包：okhttp。使用該jar完成http請(qǐng)求的發(fā)送。但是在netty5 中已經(jīng)為我們寫(xiě)好了，我們可以直接調(diào)用netty5的API就可以實(shí)現(xiàn)。
所以netty4和5的差別還是挺大的，至于使用哪個(gè)，那就看你們公司選擇哪一個(gè)了！本文目前使用netty4來(lái)實(shí)現(xiàn)文件上傳功能。下面我們上代碼：
pom文件：
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.5.Final</version>
</dependency>
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
public class FileUploadServer {
public void bind(int port) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new ObjectEncoder());
ch.pipeline().addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.weakCachingConcurrentResolver(null))); // 最大長(zhǎng)度
ch.pipeline().addLast(new FileUploadServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
try {
new FileUploadServer().bind(port);
} catch (Exception e) {
e.printStackTrace();
}
}
}
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.File;
import java.io.RandomAccessFile;
public class FileUploadServerHandler extends ChannelInboundHandlerAdapter {
private int byteRead;
private volatile int start = 0;
private String file_dir = "D:";
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof FileUploadFile) {
FileUploadFile ef = (FileUploadFile) msg;
byte[] bytes = ef.getBytes();
byteRead = ef.getEndPos();
String md5 = ef.getFile_md5();//文件名
String path = file_dir + File.separator + md5;
File file = new File(path);
RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
randomAccessFile.seek(start);
randomAccessFile.write(bytes);
start = start + byteRead;
if (byteRead > 0) {
ctx.writeAndFlush(start);
} else {
randomAccessFile.close();
ctx.close();
}
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import java.io.File;
public class FileUploadClient {
public void connect(int port, String host, final FileUploadFile fileUploadFile) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new ObjectEncoder());
ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(null)));
ch.pipeline().addLast(new FileUploadClientHandler(fileUploadFile));
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
try {
FileUploadFile uploadFile = new FileUploadFile();
File file = new File("c:/1.txt");
String fileMd5 = file.getName();// 文件名
uploadFile.setFile(file);
uploadFile.setFile_md5(fileMd5);
uploadFile.setStarPos(0);// 文件開(kāi)始位置
new FileUploadClient().connect(port, "127.0.0.1", uploadFile);
} catch (Exception e) {
e.printStackTrace();
}
}
}
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
public class FileUploadClientHandler extends ChannelInboundHandlerAdapter {
private int byteRead;
private volatile int start = 0;
private volatile int lastLength = 0;
public RandomAccessFile randomAccessFile;
private FileUploadFile fileUploadFile;
public FileUploadClientHandler(FileUploadFile ef) {
if (ef.getFile().exists()) {
if (!ef.getFile().isFile()) {
System.out.println("Not a file :" + ef.getFile());
return;
}
}
this.fileUploadFile = ef;
}
public void channelActive(ChannelHandlerContext ctx) {
try {
randomAccessFile = new RandomAccessFile(fileUploadFile.getFile(), "r");
randomAccessFile.seek(fileUploadFile.getStarPos());
lastLength = (int) randomAccessFile.length() / 10;
byte[] bytes = new byte[lastLength];
if ((byteRead = randomAccessFile.read(bytes)) != -1) {
fileUploadFile.setEndPos(byteRead);
fileUploadFile.setBytes(bytes);
ctx.writeAndFlush(fileUploadFile);
} else {
System.out.println("文件已經(jīng)讀完");
}
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException i) {
i.printStackTrace();
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof Integer) {
start = (Integer) msg;
if (start != -1) {
randomAccessFile = new RandomAccessFile(fileUploadFile.getFile(), "r");
randomAccessFile.seek(start);
System.out.println("塊兒長(zhǎng)度：" + (randomAccessFile.length() / 10));
System.out.println("長(zhǎng)度：" + (randomAccessFile.length() - start));
int a = (int) (randomAccessFile.length() - start);
int b = (int) (randomAccessFile.length() / 10);
if (a < b) {
lastLength = a;
}
byte[] bytes = new byte[lastLength];
System.out.println("-----------------------------" + bytes.length);
if ((byteRead = randomAccessFile.read(bytes)) != -1 && (randomAccessFile.length() - start) > 0) {
System.out.println("byte 長(zhǎng)度：" + bytes.length);
fileUploadFile.setEndPos(byteRead);
fileUploadFile.setBytes(bytes);
try {
ctx.writeAndFlush(fileUploadFile);
} catch (Exception e) {
e.printStackTrace();
}
} else {
randomAccessFile.close();
ctx.close();
System.out.println("文件已經(jīng)讀完--------" + byteRead);
}
}
}
}
// @Override
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws
// Exception {
// System.out.println("Server is speek ："+msg.toString());
// FileRegion filer = (FileRegion) msg;
// String path = "E://Apk//APKMD5.txt";
// File fl = new File(path);
// fl.createNewFile();
// RandomAccessFile rdafile = new RandomAccessFile(path, "rw");
// FileRegion f = new DefaultFileRegion(rdafile.getChannel(), 0,
// rdafile.length());
//
// System.out.println("This is" + ++counter + "times receive server:["
// + msg + "]");
// }
// @Override
// public void channelReadComplete(ChannelHandlerContext ctx) throws
// Exception {
// ctx.flush();
// }
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
// @Override
// protected void channelRead0(ChannelHandlerContext ctx, String msg)
// throws Exception {
// String a = msg;
// System.out.println("This is"+
// ++counter+"times receive server:["+msg+"]");
// }
}
import java.io.File;
import java.io.Serializable;
public class FileUploadFile implements Serializable {
private static final long serialVersionUID = 1L;
private File file;// 文件
private String file_md5;// 文件名
private int starPos;// 開(kāi)始位置
private byte[] bytes;// 文件字節(jié)數(shù)組
private int endPos;// 結(jié)尾位置
public int getStarPos() {
return starPos;
}
public void setStarPos(int starPos) {
this.starPos = starPos;
}
public int getEndPos() {
return endPos;
}
public void setEndPos(int endPos) {
this.endPos = endPos;
}
public byte[] getBytes() {
return bytes;
}
public void setBytes(byte[] bytes) {
this.bytes = bytes;
}
public File getFile() {
return file;
}
public void setFile(File file) {
this.file = file;
}
public String getFile_md5() {
return file_md5;
}
public void setFile_md5(String file_md5) {
this.file_md5 = file_md5;
}
}
輸出為：
塊兒長(zhǎng)度：894
長(zhǎng)度：8052
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：7158
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：6264
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：5370
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：4476
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：3582
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：2688
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：1794
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：900
-----------------------------894
byte 長(zhǎng)度：894
塊兒長(zhǎng)度：894
長(zhǎng)度：6
-----------------------------6
byte 長(zhǎng)度：6
塊兒長(zhǎng)度：894
長(zhǎng)度：0
-----------------------------0
文件已經(jīng)讀完--------0
Process finished with exit code 0
這樣就實(shí)現(xiàn)了服務(wù)器端文件的上傳，當(dāng)然我們也可以使用http的形式。
server端：
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class HttpFileServer implements Runnable {
private int port;
public HttpFileServer(int port) {
super();
this.port = port;
}
@Override
public void run() {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup);
serverBootstrap.channel(NioServerSocketChannel.class);
//serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
serverBootstrap.childHandler(new HttpChannelInitlalizer());
try {
ChannelFuture f = serverBootstrap.bind(port).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
HttpFileServer b = new HttpFileServer(9003);
new Thread(b).start();
}
}
Server端initializer：
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;
public class HttpChannelInitlalizer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpChannelHandler());
}
}
server端hadler：
import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
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.INTERNAL_SERVER_ERROR;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
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.HttpChunkedInput;
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.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.stream.ChunkedFile;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.SystemPropertyUtil;
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 HttpChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
public static final String HTTP_DATE_FORMAT = "EEE, dd MMM yyyy HH:mm:ss zzz";
public static final String HTTP_DATE_GMT_TIMEZONE = "GMT";
public static final int HTTP_CACHE_SECONDS = 60;
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
// 監(jiān)測(cè)解碼情況
if (!request.getDecoderResult().isSuccess()) {
sendError(ctx, BAD_REQUEST);
return;
}
final String uri = request.getUri();
final String path = sanitizeUri(uri);
System.out.println("get file："+path);
if (path == null) {
sendError(ctx, FORBIDDEN);
return;
}
//讀取要下載的文件
File file = new File(path);
if (file.isHidden() || !file.exists()) {
sendError(ctx, NOT_FOUND);
return;
}
if (!file.isFile()) {
sendError(ctx, FORBIDDEN);
return;
}
RandomAccessFile raf;
try {
raf = new RandomAccessFile(file, "r");
} catch (FileNotFoundException ignore) {
sendError(ctx, NOT_FOUND);
return;
}
long fileLength = raf.length();
HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
HttpHeaders.setContentLength(response, fileLength);
setContentTypeHeader(response, file);
//setDateAndCacheHeaders(response, file);
if (HttpHeaders.isKeepAlive(request)) {
response.headers().set("CONNECTION", HttpHeaders.Values.KEEP_ALIVE);
}
// Write the initial line and the header.
ctx.write(response);
// Write the content.
ChannelFuture sendFileFuture =
ctx.write(new HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)), ctx.newProgressivePromise());
//sendFuture用于監(jiān)視發(fā)送數(shù)據(jù)的狀態(tài)
sendFileFuture.addListener(new ChannelProgressiveFutureListener() {
@Override
public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {
if (total < 0) { // total unknown
System.err.println(future.channel() + " Transfer progress: " + progress);
} else {
System.err.println(future.channel() + " Transfer progress: " + progress + " / " + total);
}
}
@Override
public void operationComplete(ChannelProgressiveFuture future) {
System.err.println(future.channel() + " Transfer complete.");
}
});
// Write the end marker
ChannelFuture lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
// Decide whether to close the connection or not.
if (!HttpHeaders.isKeepAlive(request)) {
// Close the connection when the whole content is written out.
lastContentFuture.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
if (ctx.channel().isActive()) {
sendError(ctx, INTERNAL_SERVER_ERROR);
}
ctx.close();
}
private static final Pattern INSECURE_URI = Pattern.compile(".*[<>&\"].*");
private static String sanitizeUri(String uri) {
// Decode the path.
try {
uri = URLDecoder.decode(uri, "UTF-8");
} catch (UnsupportedEncodingException e) {
throw new Error(e);
}
if (!uri.startsWith("/")) {
return null;
}
// Convert file separators.
uri = uri.replace('/', File.separatorChar);
// Simplistic dumb security check.
// You will have to do something serious in the production environment.
if (uri.contains(File.separator + '.') || uri.contains('.' + File.separator) || uri.startsWith(".") || uri.endsWith(".")
|| INSECURE_URI.matcher(uri).matches()) {
return null;
}
// Convert to absolute path.
return SystemPropertyUtil.get("user.dir") + File.separator + uri;
}
private static void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
// Close the connection as soon as the error message is sent.
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/**
* Sets the content type header for the HTTP Response
*
* @param response
* HTTP response
* @param file
* file to extract content type
*/
private static void setContentTypeHeader(HttpResponse response, File file) {
MimetypesFileTypeMap m = new MimetypesFileTypeMap();
String contentType = m.getContentType(file.getPath());
if (!contentType.equals("application/octet-stream")) {
contentType += "; charset=utf-8";
}
response.headers().set(CONTENT_TYPE, contentType);
}
}
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpRequestEncoder;
import io.netty.handler.codec.http.HttpResponseDecoder;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.stream.ChunkedWriteHandler;
import java.net.URI;
public class HttpDownloadClient {
/**
* 下載http資源 向服務(wù)器下載直接填寫(xiě)要下載的文件的相對(duì)路徑
* （↑↑↑建議只使用字母和數(shù)字對(duì)特殊字符對(duì)字符進(jìn)行部分過(guò)濾可能導(dǎo)致異?！?br style="max-width: 100%;box-sizing: border-box !important;overflow-wrap: break-word !important;"> * 向互聯(lián)網(wǎng)下載輸入完整路徑
* @param host 目的主機(jī)ip或域名
* @param port 目標(biāo)主機(jī)端口
* @param url 文件路徑
* @param local 本地存儲(chǔ)路徑
* @throws Exception
*/
public void connect(String host, int port, String url, final String local) throws Exception {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChildChannelHandler(local));
// Start the client.
ChannelFuture f = b.connect(host, port).sync();
URI uri = new URI(url);
DefaultFullHttpRequest request = new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, HttpMethod.GET, uri.toASCIIString());
// 構(gòu)建http請(qǐng)求
request.headers().set(HttpHeaders.Names.HOST, host);
request.headers().set(HttpHeaders.Names.CONNECTION,
HttpHeaders.Values.KEEP_ALIVE);
request.headers().set(HttpHeaders.Names.CONTENT_LENGTH,
request.content().readableBytes());
// 發(fā)送http請(qǐng)求
f.channel().write(request);
f.channel().flush();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
String local;
public ChildChannelHandler(String local) {
this.local = local;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 客戶(hù)端接收到的是httpResponse響應(yīng)，所以要使用HttpResponseDecoder進(jìn)行解碼
ch.pipeline().addLast(new HttpResponseDecoder());
// 客戶(hù)端發(fā)送的是httprequest，所以要使用HttpRequestEncoder進(jìn)行編碼
ch.pipeline().addLast(new HttpRequestEncoder());
ch.pipeline().addLast(new ChunkedWriteHandler());
ch.pipeline().addLast(new HttpDownloadHandler(local));
}
}
public static void main(String[] args) throws Exception {
HttpDownloadClient client = new HttpDownloadClient();
//client.connect("127.0.0.1", 9003,"/file/pppp/1.doc","1.doc");
// client.connect("zlysix.gree.com", 80, "http://zlysix.gree.com/HelloWeb/download/20m.apk", "20m.apk");
client.connect("www.ghost64.com", 80, "http://www.ghost64.com/qqtupian/zixunImg/local/2017/05/27/1495855297602.jpg", "1495855297602.jpg");
}
}
import java.io.File;
import java.io.FileOutputStream;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.HttpContent;
//import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.util.internal.SystemPropertyUtil;
/**
* @Author:yangyue
* @Description:
* @Date: Created in 9:15 on 2017/5/28.
*/
public class HttpDownloadHandler extends ChannelInboundHandlerAdapter {
private boolean readingChunks = false; // 分塊讀取開(kāi)關(guān)
private FileOutputStream fOutputStream = null;// 文件輸出流
private File localfile = null;// 下載文件的本地對(duì)象
private String local = null;// 待下載文件名
private int succCode;// 狀態(tài)碼
public HttpDownloadHandler(String local) {
this.local = local;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
if (msg instanceof HttpResponse) {// response頭信息
HttpResponse response = (HttpResponse) msg;
succCode = response.getStatus().code();
if (succCode == 200) {
setDownLoadFile();// 設(shè)置下載文件
readingChunks = true;
}
// System.out.println("CONTENT_TYPE:"
// + response.headers().get(HttpHeaders.Names.CONTENT_TYPE));
}
if (msg instanceof HttpContent) {// response體信息
HttpContent chunk = (HttpContent) msg;
if (chunk instanceof LastHttpContent) {
readingChunks = false;
}
ByteBuf buffer = chunk.content();
byte[] dst = new byte[buffer.readableBytes()];
if (succCode == 200) {
while (buffer.isReadable()) {
buffer.readBytes(dst);
fOutputStream.write(dst);
buffer.release();
}
if (null != fOutputStream) {
fOutputStream.flush();
}
}
}
if (!readingChunks) {
if (null != fOutputStream) {
System.out.println("Download done->"+ localfile.getAbsolutePath());
fOutputStream.flush();
fOutputStream.close();
localfile = null;
fOutputStream = null;
}
ctx.channel().close();
}
}
/**
* 配置本地參數(shù)，準(zhǔn)備下載
*/
private void setDownLoadFile() throws Exception {
if (null == fOutputStream) {
local = SystemPropertyUtil.get("user.dir") + File.separator +local;
//System.out.println(local);
localfile = new File(local);
if (!localfile.exists()) {
localfile.createNewFile();
}
fOutputStream = new FileOutputStream(localfile);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
System.out.println("管道異常：" + cause.getMessage());
cause.printStackTrace();
ctx.channel().close();
}
}
這里客戶(hù)端我放的是網(wǎng)絡(luò)連接，下載的是一副圖片，啟動(dòng)服務(wù)端和客戶(hù)端就可以看到這個(gè)圖片被下載到了工程的根目錄下。
評(píng)論
圖片
表情