netty去实现文件传输

桃扇骨 2022-06-08 08:14 240阅读 0赞

在写出了Netty Hello World 和 netty对象传输之后，又觉得不够，看了官网的例子，所以有了现在的这个文件传输。  
顺便说下，netty官网的例子真的好，如果要学习netty，还是看官网例子的好。  
不过我英文不太好，刚开始走了绕了好大一圈，但是现在熟悉了之后，回过头来看，还是官网的牛X。  
在这里再说下netty的零拷贝，这个零拷贝是netty在3.2版本中新加入的功能。  
其主要指的是在进行一些比较大的传输比如对象或者文件传输的时候，通过改变数组索引的方式，将数据传输到特定的channel上。  
是的channel之间的转换是零拷贝，例如：ByteBuffer和ChannelBuffer  
下面就把文件传输的例子贴出来吧。  
1、FileClient.java

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **package** filetrans;  
2.    
3.  **import** **static** org.jboss.netty.channel.Channels.pipeline;  
4.    
5.  **import** java.net.InetSocketAddress;  
6.  **import** java.util.concurrent.Executors;  
7.    
8.  **import** org.jboss.netty.bootstrap.ClientBootstrap;  
9.  **import** org.jboss.netty.channel.ChannelFuture;  
10. **import** org.jboss.netty.channel.ChannelPipeline;  
11. **import** org.jboss.netty.channel.ChannelPipelineFactory;  
12. **import** org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;  
13. **import** org.jboss.netty.handler.codec.http.DefaultHttpRequest;  
14. **import** org.jboss.netty.handler.codec.http.HttpMethod;  
15. **import** org.jboss.netty.handler.codec.http.HttpRequest;  
16. **import** org.jboss.netty.handler.codec.http.HttpRequestEncoder;  
17. **import** org.jboss.netty.handler.codec.http.HttpResponseDecoder;  
18. **import** org.jboss.netty.handler.codec.http.HttpVersion;  
19. **import** org.jboss.netty.handler.stream.ChunkedWriteHandler;  
20.   
21. **public** **class** FileClient  
22. \{  
23.     **public** **static** **void** main(String\[\] args)  
24.     \{  
25.   
26.         ClientBootstrap bootstrap = **new** ClientBootstrap(  
27.                 **new** NioClientSocketChannelFactory(  
28.                         Executors.newCachedThreadPool(),  
29.                         Executors.newCachedThreadPool()));  
30.         bootstrap.setPipelineFactory(**new** ChannelPipelineFactory()  
31.         \{  
32.   
33.             @Override  
34.             **public** ChannelPipeline getPipeline() **throws** Exception  
35.             \{  
36.                 ChannelPipeline pipeline = pipeline();  
37.   
38.                 pipeline.addLast("decoder", **new** HttpResponseDecoder());  
39.                   
40.                 /\* 
41.                  \* 不能添加这个，对传输文件 进行了大小的限制。。。。。 
42.                  \*/  
43. //                pipeline.addLast("aggregator", new HttpChunkAggregator(6048576));  
44.                 pipeline.addLast("encoder", **new** HttpRequestEncoder());  
45.                 pipeline.addLast("chunkedWriter", **new** ChunkedWriteHandler());  
46.                 pipeline.addLast("handler", **new** FileClientHandler());  
47.   
48.                 **return** pipeline;  
49.             \}  
50.   
51.         \});  
52.   
53.         ChannelFuture future = bootstrap.connect(**new** InetSocketAddress(  
54.                 "localhost", 8080));  
55.   
56.         /\* 
57.          \* 这里为了保证connect连接，所以才进行了sleep 
58.          \* 当然也可以通过future的connect属性判断 
59.          \*/  
60.         **try**  
61.         \{  
62.             Thread.sleep(3000);  
63.         \} **catch** (InterruptedException e)  
64.         \{  
65.             e.printStackTrace();  
66.         \}  
67.         HttpRequest request = **new** DefaultHttpRequest(HttpVersion.HTTP\_1\_1,  
68.                 HttpMethod.GET, "DSC01575.JPG");  
69.         future.getChannel().write(request);  
70.   
71.         // Wait until the connection is closed or the connection attempt fails.  
72.         future.getChannel().getCloseFuture().awaitUninterruptibly();  
73.   
74.         // Shut down thread pools to exit.  
75.         bootstrap.releaseExternalResources();  
76.   
77.     \}  
78. \}

**\[java\]**  [view plain][view plain 1]   [copy][view plain 1]

2、FileClientHandler.java

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **package** filetrans;  
2.    
3.  **import** java.io.File;  
4.  **import** java.io.FileOutputStream;  
5.    
6.  **import** org.jboss.netty.buffer.ChannelBuffer;  
7.  **import** org.jboss.netty.channel.ChannelHandlerContext;  
8.  **import** org.jboss.netty.channel.ExceptionEvent;  
9.  **import** org.jboss.netty.channel.MessageEvent;  
10. **import** org.jboss.netty.channel.SimpleChannelUpstreamHandler;  
11. **import** org.jboss.netty.handler.codec.http.DefaultHttpResponse;  
12. **import** org.jboss.netty.handler.codec.http.HttpChunk;  
13. **import** org.jboss.netty.handler.codec.http.HttpResponse;  
14.   
15. **public** **class** FileClientHandler **extends** SimpleChannelUpstreamHandler  
16. \{  
17.     **private** **volatile** **boolean** readingChunks;  
18.     **private** File downloadFile;  
19.     **private** FileOutputStream fOutputStream = **null**;  
20.   
21.     @Override  
22.     **public** **void** messageReceived(ChannelHandlerContext ctx, MessageEvent e)  
23.             **throws** Exception  
24.     \{  
25.         /\* 
26.          \* 按照channle的顺序进行处理 
27.          \* server先发送HttpResponse过来，所以这里先对HttpResponse进行处理，进行文件判断之类 
28.          \* 之后，server发送的都是ChunkedFile了。 
29.          \*/  
30.           
31.         **if** (e.getMessage() **instanceof** HttpResponse)  
32.         \{  
33.             DefaultHttpResponse httpResponse = (DefaultHttpResponse) e  
34.                     .getMessage();  
35.             String fileName = httpResponse.getHeader("fileName");  
36.             downloadFile = **new** File(System.getProperty("user.dir")  
37.                     + File.separator + "recived\_" + fileName);  
38.             readingChunks = httpResponse.isChunked();  
39.         \} **else**  
40.         \{  
41.             HttpChunk httpChunk = (HttpChunk) e.getMessage();  
42.             **if** (!httpChunk.isLast())  
43.             \{  
44.                 ChannelBuffer buffer = httpChunk.getContent();  
45.                 **if** (fOutputStream == **null**)  
46.                 \{  
47.                     fOutputStream = **new** FileOutputStream(downloadFile);  
48.                 \}  
49.                 **while** (buffer.readable())  
50.                 \{  
51.                     **byte**\[\] dst = **new** **byte**\[buffer.readableBytes()\];  
52.                     buffer.readBytes(dst);  
53.                     fOutputStream.write(dst);  
54.                 \}  
55.             \} **else**  
56.             \{  
57.                 readingChunks = **false**;  
58.             \}  
59.             fOutputStream.flush();  
60.         \}  
61.         **if** (!readingChunks)  
62.         \{  
63.             fOutputStream.close();  
64.             e.getChannel().close();  
65.         \}  
66.     \}  
67.   
68.     @Override  
69.     **public** **void** exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)  
70.             **throws** Exception  
71.     \{  
72.         System.out.println(e.getCause());  
73.     \}  
74. \}

**\[java\]**  [view plain][view plain 1]   [copy][view plain 1]

3、FileServer.java

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

**\[java\]**  [view plain][view plain 1]   [copy][view plain 1]

4、FileServerHandler.java

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **package** filetrans;  
2.    
3.  **import** **static** org.jboss.netty.handler.codec.http.HttpHeaders.\*;  
4.  **import** **static** org.jboss.netty.handler.codec.http.HttpHeaders.Names.\*;  
5.  **import** **static** org.jboss.netty.handler.codec.http.HttpMethod.\*;  
6.  **import** **static** org.jboss.netty.handler.codec.http.HttpResponseStatus.\*;  
7.  **import** **static** org.jboss.netty.handler.codec.http.HttpVersion.\*;  
8.    
9.  **import** java.io.File;  
10. **import** java.io.FileNotFoundException;  
11. **import** java.io.RandomAccessFile;  
12. **import** java.io.UnsupportedEncodingException;  
13. **import** java.net.URLDecoder;  
14.   
15. **import** org.jboss.netty.buffer.ChannelBuffers;  
16. **import** org.jboss.netty.channel.Channel;  
17. **import** org.jboss.netty.channel.ChannelFuture;  
18. **import** org.jboss.netty.channel.ChannelFutureListener;  
19. **import** org.jboss.netty.channel.ChannelFutureProgressListener;  
20. **import** org.jboss.netty.channel.ChannelHandlerContext;  
21. **import** org.jboss.netty.channel.DefaultFileRegion;  
22. **import** org.jboss.netty.channel.ExceptionEvent;  
23. **import** org.jboss.netty.channel.FileRegion;  
24. **import** org.jboss.netty.channel.MessageEvent;  
25. **import** org.jboss.netty.channel.SimpleChannelUpstreamHandler;  
26. **import** org.jboss.netty.handler.codec.frame.TooLongFrameException;  
27. **import** org.jboss.netty.handler.codec.http.DefaultHttpResponse;  
28. **import** org.jboss.netty.handler.codec.http.HttpRequest;  
29. **import** org.jboss.netty.handler.codec.http.HttpResponse;  
30. **import** org.jboss.netty.handler.codec.http.HttpResponseStatus;  
31. **import** org.jboss.netty.handler.ssl.SslHandler;  
32. **import** org.jboss.netty.handler.stream.ChunkedFile;  
33. **import** org.jboss.netty.util.CharsetUtil;  
34.   
35. /\*\* 
36.  \* 这里的代码主要来源于官网的例子，http里面有个例子，自己仿照server写了client 
37.  \* @author Ransom 
38.  \*/  
39. **public** **class** FileServerHandler **extends** SimpleChannelUpstreamHandler  
40. \{  
41.   
42.     @Override  
43.     **public** **void** messageReceived(ChannelHandlerContext ctx, MessageEvent e)  
44.             **throws** Exception  
45.     \{  
46.         HttpRequest request = (HttpRequest) e.getMessage();  
47.         **if** (request.getMethod() != GET)  
48.         \{  
49.             sendError(ctx, METHOD\_NOT\_ALLOWED);  
50.             **return**;  
51.         \}  
52.   
53.         **final** String path = sanitizeUri(request.getUri());  
54.         **if** (path == **null**)  
55.         \{  
56.             sendError(ctx, FORBIDDEN);  
57.             **return**;  
58.         \}  
59.   
60.         File file = **new** File(path);  
61.         **if** (file.isHidden() || !file.exists())  
62.         \{  
63.             sendError(ctx, NOT\_FOUND);  
64.             **return**;  
65.         \}  
66.         **if** (!file.isFile())  
67.         \{  
68.             sendError(ctx, FORBIDDEN);  
69.             **return**;  
70.         \}  
71.   
72.         RandomAccessFile raf;  
73.         **try**  
74.         \{  
75.             raf = **new** RandomAccessFile(file, "r");  
76.         \} **catch** (FileNotFoundException fnfe)  
77.         \{  
78.             sendError(ctx, NOT\_FOUND);  
79.             **return**;  
80.         \}  
81.         **long** fileLength = raf.length();  
82.   
83.         HttpResponse response = **new** DefaultHttpResponse(HTTP\_1\_1, OK);  
84.           
85.         /\* 
86.          \* 由于是异步传输，所以不得已加入了一些属性，用来进行文件识别 
87.          \*/  
88.         response.addHeader("fileName", request.getUri());  
89.           
90.         setContentLength(response, fileLength);  
91.   
92.         Channel ch = e.getChannel();  
93.   
94.         // Write the initial line and the header.  
95.         ch.write(response);  
96.   
97.         // Write the content.  
98.         ChannelFuture writeFuture;  
99.         **if** (ch.getPipeline().get(SslHandler.**class**) != **null**)  
100.         \{  
101.             // Cannot use zero-copy with HTTPS.  
102.             writeFuture = ch.write(**new** ChunkedFile(raf, 0, fileLength, 8192));  
103.         \} **else**  
104.         \{  
105.             // No encryption - use zero-copy.  
106.             **final** FileRegion region = **new** DefaultFileRegion(raf.getChannel(),  
107.                     0, fileLength);  
108.             writeFuture = ch.write(region);  
109.             writeFuture.addListener(**new** ChannelFutureProgressListener()  
110.             \{  
111.                 **public** **void** operationComplete(ChannelFuture future)  
112.                 \{  
113.                     region.releaseExternalResources();  
114.                 \}  
115.   
116.                 **public** **void** operationProgressed(ChannelFuture future,  
117.                         **long** amount, **long** current, **long** total)  
118.                 \{  
119.                     System.out.printf("%s: %d / %d (+%d)%n", path, current,  
120.                             total, amount);  
121.                 \}  
122.             \});  
123.         \}  
124.   
125.         // Decide whether to close the connection or not.  
126.         **if** (!isKeepAlive(request))  
127.         \{  
128.             // Close the connection when the whole content is written out.  
129.             writeFuture.addListener(ChannelFutureListener.CLOSE);  
130.         \}  
131.     \}  
132.   
133.     @Override  
134.     **public** **void** exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)  
135.             **throws** Exception  
136.     \{  
137.         Channel ch = e.getChannel();  
138.         Throwable cause = e.getCause();  
139.         **if** (cause **instanceof** TooLongFrameException)  
140.         \{  
141.             sendError(ctx, BAD\_REQUEST);  
142.             **return**;  
143.         \}  
144.   
145.         cause.printStackTrace();  
146.         **if** (ch.isConnected())  
147.         \{  
148.             sendError(ctx, INTERNAL\_SERVER\_ERROR);  
149.         \}  
150.     \}  
151.   
152.     **private** String sanitizeUri(String uri)  
153.     \{  
154.         // Decode the path.  
155.         **try**  
156.         \{  
157.             uri = URLDecoder.decode(uri, "UTF-8");  
158.         \} **catch** (UnsupportedEncodingException e)  
159.         \{  
160.             **try**  
161.             \{  
162.                 uri = URLDecoder.decode(uri, "ISO-8859-1");  
163.             \} **catch** (UnsupportedEncodingException e1)  
164.             \{  
165.                 **throw** **new** Error();  
166.             \}  
167.         \}  
168.   
169.         // Convert file separators.  
170.         uri = uri.replace('/', File.separatorChar);  
171.   
172.         // Simplistic dumb security check.  
173.         // You will have to do something serious in the production environment.  
174.         **if** (uri.contains(File.separator + ".")  
175.                 || uri.contains("." + File.separator) || uri.startsWith(".")  
176.                 || uri.endsWith("."))  
177.         \{  
178.             **return** **null**;  
179.         \}  
180.   
181.         // Convert to absolute path.  
182.         **return** System.getProperty("user.dir") + File.separator + uri;  
183.     \}  
184.   
185.     **private** **void** sendError(ChannelHandlerContext ctx, HttpResponseStatus status)  
186.     \{  
187.         HttpResponse response = **new** DefaultHttpResponse(HTTP\_1\_1, status);  
188.         response.setHeader(CONTENT\_TYPE, "text/plain; charset=UTF-8");  
189.         response.setContent(ChannelBuffers.copiedBuffer(  
190.                 "Failure: " + status.toString() + "\\r\\n", CharsetUtil.UTF\_8));  
191.   
192.         // Close the connection as soon as the error message is sent.  
193.         ctx.getChannel().write(response)  
194.                 .addListener(ChannelFutureListener.CLOSE);  
195.     \}  
196. \}

**\[java\]**  [view plain][view plain 1]   [copy][view plain 1]

Netty的例子到这里就算贴完了，后面还打算进行一些Netty实现原理，架构，代码解读方面的分析。到时候如果有什么心得，再写出来吧。

[view plain]: http://blog.csdn.net/cxhzqhzq/article/details/6618420#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/196789
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/196789/fork
[view plain 1]: http://blog.csdn.net/a351945755/article/details/19566125#