【初學與研發之NETTY】netty3之檔案上傳
阿新 • • 發佈：2019-01-30
客戶端：
package netty3.socket.client; import static org.jboss.netty.channel.Channels.pipeline; import java.io.File; import java.net.InetSocketAddress; import java.util.List; import java.util.concurrent.Executors; import org.jboss.netty.bootstrap.ClientBootstrap; import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelFuture; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineFactory; import org.jboss.netty.channel.ExceptionEvent; import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelUpstreamHandler; import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; import org.jboss.netty.handler.codec.http.DefaultHttpRequest; import org.jboss.netty.handler.codec.http.HttpChunk; import org.jboss.netty.handler.codec.http.HttpClientCodec; import org.jboss.netty.handler.codec.http.HttpHeaders; import org.jboss.netty.handler.codec.http.HttpMethod; import org.jboss.netty.handler.codec.http.HttpRequest; import org.jboss.netty.handler.codec.http.HttpRequestEncoder; import org.jboss.netty.handler.codec.http.HttpResponse; import org.jboss.netty.handler.codec.http.HttpResponseDecoder; import org.jboss.netty.handler.codec.http.HttpVersion; import org.jboss.netty.handler.codec.http.multipart.DefaultHttpDataFactory; import org.jboss.netty.handler.codec.http.multipart.HttpDataFactory; import org.jboss.netty.handler.codec.http.multipart.HttpPostRequestEncoder; import org.jboss.netty.handler.codec.http.multipart.InterfaceHttpData; import org.jboss.netty.handler.stream.ChunkedWriteHandler; import org.jboss.netty.util.CharsetUtil; public class UploadFileClient { private ClientBootstrap bootstrap = null; private ChannelFuture future = null; private HttpDataFactory factory = null; // 服務端處理完成後返回的訊息 private StringBuffer retMsg = new StringBuffer(); public UploadFileClient() { bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); bootstrap.setPipelineFactory(new UploadChannelFactory()); // 連線超時時間為3s bootstrap.setOption("connectTimeoutMillis", 3000); future = bootstrap.connect(new InetSocketAddress("127.0.0.1", 2777)); // 獲得一個閾值，它是來控制上傳檔案時記憶體/硬碟的比值，防止出現記憶體溢位 factory = new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE); } /** * 方法描述：關閉檔案傳送通道（為阻塞式） */ public void shutdownClient() { // 等待資料的傳輸通道關閉 future.getChannel().getCloseFuture().awaitUninterruptibly(); bootstrap.releaseExternalResources(); // Really clean all temporary files if they still exist factory.cleanAllHttpDatas(); } /** * 方法描述：獲取傳送檔案過程中服務端反饋的訊息 * @return 服務端反饋的訊息 */ public String getRetMsg() { return retMsg.toString(); } /** * 方法描述：將檔案上傳到服務端 * @param file 待上傳的檔案 */ public void uploadFile(File file) { if (!file.canRead()) { return; } // Simple Post form: factory used for big attributes List<InterfaceHttpData> bodylist = formpost(file); if (bodylist == null) { return; } // Multipart Post form: factory used uploadFileToServer(file.getName(), factory, bodylist); } /** * @param file * @return */ private List<InterfaceHttpData> formpost(File file) { // Prepare the HTTP request. HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, ""); // Use the PostBody encoder HttpPostRequestEncoder bodyRequestEncoder = null; try { bodyRequestEncoder = new HttpPostRequestEncoder(factory, request, false); bodyRequestEncoder.addBodyAttribute("getform", "POST"); bodyRequestEncoder.addBodyFileUpload("myfile", file, "application/x-zip-compressed", false); } catch(Exception e) { // should not be since args are not null e.printStackTrace(); return null; } // Create the bodylist to be reused on the last version with Multipart support List<InterfaceHttpData> bodylist = bodyRequestEncoder.getBodyListAttributes(); return bodylist; } /** * Multipart example */ private void uploadFileToServer(String fileName, HttpDataFactory factory, List<InterfaceHttpData> bodylist) { // Wait until the connection attempt succeeds or fails. Channel channel = future.awaitUninterruptibly().getChannel(); if (!future.isSuccess()) { future.getCause().printStackTrace(); bootstrap.releaseExternalResources(); return; } // Prepare the HTTP request. HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, fileName); // 設定該屬性表示服務端檔案接收完畢後會關閉傳送通道 request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE); // Use the PostBody encoder HttpPostRequestEncoder bodyRequestEncoder = null; try { bodyRequestEncoder = new HttpPostRequestEncoder(factory, request, true); bodyRequestEncoder.setBodyHttpDatas(bodylist); bodyRequestEncoder.finalizeRequest(); } catch(Exception e) { // should not be since no null args e.printStackTrace(); } System.out.println("開始時間："+System.currentTimeMillis()); // send request channel.write(request); // test if request was chunked and if so, finish the write if (bodyRequestEncoder.isChunked()) { channel.write(bodyRequestEncoder).awaitUninterruptibly(); } // Now no more use of file representation (and list of HttpData) bodyRequestEncoder.cleanFiles(); } private class UploadChannelFactory implements ChannelPipelineFactory { public ChannelPipeline getPipeline() throws Exception { ChannelPipeline pipeline = pipeline(); pipeline.addLast("decoder", new HttpResponseDecoder()); pipeline.addLast("encoder", new HttpRequestEncoder()); pipeline.addLast("codec", new HttpClientCodec()); pipeline.addLast("chunkedWriter", new ChunkedWriteHandler()); pipeline.addLast("handler", new UploadClientHandler()); return pipeline; } } private class UploadClientHandler extends SimpleChannelUpstreamHandler { private boolean readingChunks; /** * 方法描述：接收服務端返回的訊息 * @param ctx 傳送訊息的通道物件 * @param e 訊息傳送事件物件 */ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { if (!readingChunks) { HttpResponse response = (HttpResponse)e.getMessage(); // 收到服務端反饋的訊息，並且連結正常、且還有後續訊息 if (response.getStatus().getCode() == 200 && response.isChunked()) { readingChunks = true; } else { // 服務端有反饋訊息，但沒有後續的訊息了 ChannelBuffer content = response.getContent(); if (content.readable()) { retMsg.append(content.toString(CharsetUtil.UTF_8)); } } } else { HttpChunk chunk = (HttpChunk)e.getMessage(); if (chunk.isLast()) { // 服務端的訊息接收完畢 readingChunks = false; } else { // 連續接收服務端發過來的訊息 retMsg.append(chunk.getContent().toString(CharsetUtil.UTF_8)); } } } /** * 方法描述：訊息接收或傳送過程中出現異常 * @param ctx 傳送訊息的通道物件 * @param e 異常事件物件 */ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { System.out.println("異常--：" + e.getCause()); e.getChannel().close(); // 有異常後釋放客戶端佔用的通道資源 shutdownClient(); } } }
服務端：
package netty3.socket.server; import static org.jboss.netty.channel.Channels.pipeline; import java.net.InetSocketAddress; import java.util.concurrent.Executors; import org.jboss.netty.bootstrap.ServerBootstrap; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineFactory; import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; import org.jboss.netty.handler.codec.http.HttpRequestDecoder; import org.jboss.netty.handler.codec.http.HttpResponseEncoder; import org.jboss.netty.handler.stream.ChunkedWriteHandler; public class InitServer { private static InitServer sockServer = null; private static ServerBootstrap bootstrap = null; public static InitServer getInstance() { if (sockServer == null) { sockServer = new InitServer(); } return sockServer; } public InitServer() { bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { public ChannelPipeline getPipeline() throws Exception { ChannelPipeline pipeline = pipeline(); pipeline.addLast("decoder", new HttpRequestDecoder()); pipeline.addLast("encoder", new HttpResponseEncoder()); pipeline.addLast("chunkedWriter", new ChunkedWriteHandler()); pipeline.addLast("handler", new ServerHandler()); return pipeline; } }); bootstrap.bind(new InetSocketAddress("127.0.0.1", 2777)); } public void shutdownServer() { bootstrap.releaseExternalResources(); } }
package netty3.socket.server; import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.CACHE_CONTROL; import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE; import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.DATE; import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.EXPIRES; import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.LAST_MODIFIED; import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1; import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; import java.io.RandomAccessFile; import java.io.UnsupportedEncodingException; import java.net.URLDecoder; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; import java.util.GregorianCalendar; import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Random; import java.util.TimeZone; import javax.activation.MimetypesFileTypeMap; import netty3.socket.client.SendMsgClient; import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBuffers; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelFuture; import org.jboss.netty.channel.ChannelFutureListener; import org.jboss.netty.channel.ChannelFutureProgressListener; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelStateEvent; import org.jboss.netty.channel.Channels; import org.jboss.netty.channel.DefaultFileRegion; import org.jboss.netty.channel.ExceptionEvent; import org.jboss.netty.channel.FileRegion; import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelHandler; import org.jboss.netty.handler.codec.frame.TooLongFrameException; import org.jboss.netty.handler.codec.http.DefaultHttpRequest; import org.jboss.netty.handler.codec.http.DefaultHttpResponse; import org.jboss.netty.handler.codec.http.HttpChunk; import org.jboss.netty.handler.codec.http.HttpHeaders; import org.jboss.netty.handler.codec.http.HttpMethod; import org.jboss.netty.handler.codec.http.HttpRequest; import org.jboss.netty.handler.codec.http.HttpResponse; import org.jboss.netty.handler.codec.http.HttpResponseStatus; import org.jboss.netty.handler.codec.http.HttpVersion; import org.jboss.netty.handler.codec.http.multipart.Attribute; import org.jboss.netty.handler.codec.http.multipart.DefaultHttpDataFactory; import org.jboss.netty.handler.codec.http.multipart.DiskFileUpload; import org.jboss.netty.handler.codec.http.multipart.FileUpload; import org.jboss.netty.handler.codec.http.multipart.HttpDataFactory; import org.jboss.netty.handler.codec.http.multipart.HttpPostRequestDecoder; import org.jboss.netty.handler.codec.http.multipart.HttpPostRequestDecoder.EndOfDataDecoderException; import org.jboss.netty.handler.codec.http.multipart.InterfaceHttpData; import org.jboss.netty.handler.codec.http.multipart.InterfaceHttpData.HttpDataType; import org.jboss.netty.handler.ssl.SslHandler; import org.jboss.netty.handler.stream.ChunkedFile; import org.jboss.netty.util.CharsetUtil; public class ServerHandler extends SimpleChannelHandler { 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; private static final HttpDataFactory factory = new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE); // Disk if size exceed MINSIZE private HttpPostRequestDecoder decoder; private HttpRequest request; private String receiveFileName = ""; private Map<String, String> msgMap = new HashMap<String, String>(); private boolean readingChunks = false; static { DiskFileUpload.baseDirectory = "/home/build1/file_test/"; } public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { if (e.getMessage() instanceof HttpRequest) { HttpRequest request = (DefaultHttpRequest)e.getMessage(); String uri = sanitizeUri(request.getUri()); System.out.println(request.isChunked()); if (request.getMethod() == HttpMethod.POST) { // 接收客戶端上傳的檔案 receiveFileName = uri; this.request = request; // clean previous FileUpload if Any if (decoder != null) { decoder.cleanFiles(); decoder = null; } // if GET Method: should not try to create a HttpPostRequestDecoder try { decoder = new HttpPostRequestDecoder(factory, request); } catch(Exception e1) { e1.printStackTrace(); writeResponse(e.getChannel(), "接收檔案資訊時出現異常：" + e1.toString()); Channels.close(e.getChannel()); return; } if (!request.isChunked()) { readHttpDataAllReceive(e.getChannel()); writeResponse(e.getChannel(), "服務端檔案接收完畢！"); } } } else { // New chunk is received HttpChunk chunk = (HttpChunk)e.getMessage(); // example of reading only if at the end if (!chunk.isLast()) { try { decoder.offer(chunk); } catch(Exception e1) { e1.printStackTrace(); writeResponse(e.getChannel(), "接收檔案資料時出現異常：" + e1.toString()); Channels.close(e.getChannel()); return; } // example of reading chunk by chunk (minimize memory usage due to Factory) readHttpDataChunkByChunk(); } else { readHttpDataAllReceive(e.getChannel()); //writeResponse(e.getChannel(), "服務端資料接收完畢！"); String sendMsg = msgMap.get("sendMsg"); System.out.println("服務端收到訊息：" + sendMsg); sendReturnMsg(ctx, HttpResponseStatus.OK, "服務端返回的訊息！"); } } } /** * Example of reading all InterfaceHttpData from finished transfer */ private void readHttpDataAllReceive(Channel channel) { List<InterfaceHttpData> datas; try { datas = decoder.getBodyHttpDatas(); } catch(Exception e1) { e1.printStackTrace(); writeResponse(channel, "接收檔案資料時出現異常：" + e1.toString()); Channels.close(channel); return; } for (InterfaceHttpData data : datas) { writeHttpData(data); } } /** * Example of reading request by chunk and getting values from chunk to chunk */ private void readHttpDataChunkByChunk() { try { while(decoder.hasNext()) { InterfaceHttpData data = decoder.next(); if (data != null) { // new value writeHttpData(data); } } } catch(EndOfDataDecoderException e1) { e1.printStackTrace(); } } private void writeHttpData(InterfaceHttpData data) { if (data.getHttpDataType() == HttpDataType.FileUpload) { FileUpload fileUpload = (FileUpload)data; if (fileUpload.isCompleted()) { try { Random r = new Random(); StringBuffer fileNameBuf = new StringBuffer(); fileNameBuf.append(DiskFileUpload.baseDirectory).append("U").append(System.currentTimeMillis()); fileNameBuf.append(String.valueOf(r.nextInt(10))).append(String.valueOf(r.nextInt(10))); fileNameBuf.append(receiveFileName.substring(receiveFileName.lastIndexOf("."))); fileUpload.renameTo(new File(fileNameBuf.toString())); } catch(IOException e) { e.printStackTrace(); } System.out.println("結束時間："+System.currentTimeMillis()); } else { System.out.println("\tFile to be continued but should not!\r\n"); } } else if (data.getHttpDataType() == HttpDataType.Attribute) { Attribute attribute = (Attribute)data; try { msgMap.put(attribute.getName(), attribute.getString()); } catch(IOException e) { e.printStackTrace(); } } } private void writeResponse(Channel channel, String retMsg) { // Convert the response content to a ChannelBuffer. ChannelBuffer buf = ChannelBuffers.copiedBuffer(retMsg, CharsetUtil.UTF_8); // Decide whether to close the connection or not. boolean close = HttpHeaders.Values.CLOSE.equalsIgnoreCase(request.getHeader(HttpHeaders.Names.CONNECTION)) || request.getProtocolVersion().equals(HttpVersion.HTTP_1_0) && !HttpHeaders.Values.KEEP_ALIVE.equalsIgnoreCase(request.getHeader(HttpHeaders.Names.CONNECTION)); // Build the response object. HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); response.setContent(buf); response.setHeader(HttpHeaders.Names.CONTENT_TYPE, "text/plain; charset=UTF-8"); if (!close) { // There's no need to add 'Content-Length' header // if this is the last response. response.setHeader(HttpHeaders.Names.CONTENT_LENGTH, String.valueOf(buf.readableBytes())); } // Write the response. ChannelFuture future = channel.write(response); // Close the connection after the write operation is done if necessary. if (close) { future.addListener(ChannelFutureListener.CLOSE); } } private String sanitizeUri(String uri) { try { uri = URLDecoder.decode(uri, "UTF-8"); } catch(UnsupportedEncodingException e) { try { uri = URLDecoder.decode(uri, "ISO-8859-1"); } catch(UnsupportedEncodingException e1) { throw new Error(); } } return uri; } /** * 方法描述：設定請求響應的header資訊 * @param response 請求響應物件 * @param fileToCache 下載檔案 */ private static void setContentTypeHeader(HttpResponse response, File fileToCache) { MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap(); response.setHeader(CONTENT_TYPE, mimeTypesMap.getContentType(fileToCache.getPath())); SimpleDateFormat dateFormatter = new SimpleDateFormat(HTTP_DATE_FORMAT, Locale.US); dateFormatter.setTimeZone(TimeZone.getTimeZone(HTTP_DATE_GMT_TIMEZONE)); // Date header Calendar time = new GregorianCalendar(); response.setHeader(DATE, dateFormatter.format(time.getTime())); // Add cache headers time.add(Calendar.SECOND, HTTP_CACHE_SECONDS); response.setHeader(EXPIRES, dateFormatter.format(time.getTime())); response.setHeader(CACHE_CONTROL, "private, max-age=" + HTTP_CACHE_SECONDS); response.setHeader(LAST_MODIFIED, dateFormatter.format(new Date(fileToCache.lastModified()))); } /** * 方法描述：給客戶端傳送反饋訊息 * @param ctx 傳送訊息的通道 * @param status 狀態 * @param retMsg 反饋訊息 */ private static void sendReturnMsg(ChannelHandlerContext ctx, HttpResponseStatus status, String retMsg) { HttpResponse response = new DefaultHttpResponse(HTTP_1_1, status); response.setHeader(CONTENT_TYPE, "text/plain; charset=UTF-8"); response.setContent(ChannelBuffers.copiedBuffer(retMsg, CharsetUtil.UTF_8)); // 資訊傳送成功後，關閉連線通道 ctx.getChannel().write(response).addListener(ChannelFutureListener.CLOSE); } public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { if (decoder != null) { decoder.cleanFiles(); } System.out.println("連線斷開:" + e.getChannel().getRemoteAddress().toString()); } public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { String remoteIp = e.getChannel().getRemoteAddress().toString(); System.out.println(remoteIp.substring(1, remoteIp.indexOf(":"))); System.out.println("收到連線:" + e.getChannel().getRemoteAddress().toString()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { Channel ch = e.getChannel(); Throwable cause = e.getCause(); if (cause instanceof TooLongFrameException) { return; } System.err.println("連線的通道出現異常：" + cause.toString()); if (ch.isConnected()) { System.out.println("連線還沒有關閉！"); ch.close(); } } }