Netty（三）文件上传下载、心跳检测

文件上传

Http服务

//http服务
public final class HttpServer {static final boolean SSL = System.getProperty("ssl") != null;static final int PORT = Integer.parseInt(System.getProperty("port", SSL ? "8843" : "8080"));public static void main(String[] args) throws Exception {// 配置sslfinal SslContext sslCtx;if (SSL) {SelfSignedCertificate ssc = new SelfSignedCertificate();sslCtx = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());} else {sslCtx = null;}// 配置服务EventLoopGroup bGroup = new NioEventLoopGroup(1);EventLoopGroup wGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.option(ChannelOption.SO_BACKLOG, 1024);b.group(bGroup,wGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new HttpServerInitializer(sslCtx));Channel channel = b.bind(PORT).sync().channel();System.err.println("打开浏览器，输入链接 " + (SSL ? "https" : "http") + "://127.0.0.1:" + PORT + '/');channel.closeFuture().sync();} finally {bGroup.shutdownGracefully();wGroup.shutdownGracefully();}}
}public class HttpServerInitializer extends ChannelInitializer<SocketChannel>{public final SslContext sslCtx;public HttpServerInitializer(SslContext sslCtx) {this.sslCtx = sslCtx;}@Overrideprotected void initChannel(SocketChannel sc) throws Exception {ChannelPipeline pipeline = sc.pipeline();if (sslCtx != null) {pipeline.addLast(sslCtx.newHandler(sc.alloc()));}pipeline.addLast(new HttpServerCodec());//添加httpServerpipeline.addLast(new HttpServerHandler());}}public class HttpServerHandler extends ChannelHandlerAdapter {private static final byte[] CONTENT = new byte[]{ 'H', 'e', 'l', 'l', 'o', ' ', 'W', 'o', 'r', 'l', 'd' };@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof HttpRequest) {HttpRequest request = (HttpRequest) msg;if (HttpHeaderUtil.is100ContinueExpected(request)) {ctx.write(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));}boolean keepAlive = HttpHeaderUtil.isKeepAlive(request);FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,Unpooled.wrappedBuffer(CONTENT));response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH,response.content().readableBytes());if (!keepAlive) {ctx.write(response).addListener(ChannelFutureListener.CLOSE);} else {response.headers().set(HttpHeaderNames.CONNECTION,HttpHeaderValues.KEEP_ALIVE);ctx.write(response);}}}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}}

文件上传

public final class HttpUploadServer {static final boolean SSL = System.getProperty("ssl") != null;static final int PORT = Integer.parseInt(System.getProperty("port", SSL? "8443" : "8080"));public static void main(String[] args) throws Exception {// Configure SSL.final SslContext sslCtx;if (SSL) {SelfSignedCertificate ssc = new SelfSignedCertificate();sslCtx = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());} else {sslCtx = null;}EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup);b.channel(NioServerSocketChannel.class);b.handler(new LoggingHandler(LogLevel.INFO));b.childHandler(new HttpUploadServerInitializer(sslCtx));Channel ch = b.bind(PORT).sync().channel();System.err.println("Open your web browser and navigate to " +(SSL? "https" : "http") + "://127.0.0.1:" + PORT + '/');ch.closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}public class HttpUploadServerHandler extends SimpleChannelInboundHandler<HttpObject> {private static final Logger logger = Logger.getLogger(HttpUploadServerHandler.class.getName());private HttpRequest request;private boolean readingChunks;private final StringBuilder responseContent = new StringBuilder();private static final HttpDataFactory factory =new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE); // Disk if size exceedprivate HttpPostRequestDecoder decoder;static {DiskFileUpload.deleteOnExitTemporaryFile = true; // should delete file// on exit (in normal// exit)DiskFileUpload.baseDirectory = "D:" + File.separatorChar + "aa"; // system temp directoryDiskAttribute.deleteOnExitTemporaryFile = true; // should delete file on// exit (in normal exit)DiskAttribute.baseDirectory = "D:" + File.separatorChar + "aa"; // system temp directory}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {if (decoder != null) {decoder.cleanFiles();}}@Overridepublic void messageReceived(ChannelHandlerContext ctx, HttpObject msg) throws Exception {if (msg instanceof HttpRequest) {HttpRequest request = this.request = (HttpRequest) msg;URI uri = new URI(request.uri());if (!uri.getPath().startsWith("/form")) {// Write MenuwriteMenu(ctx);return;}responseContent.setLength(0);responseContent.append("WELCOME TO THE WILD WILD WEB SERVER\r\n");responseContent.append("===================================\r\n");responseContent.append("VERSION: " + request.protocolVersion().text() + "\r\n");responseContent.append("REQUEST_URI: " + request.uri() + "\r\n\r\n");responseContent.append("\r\n\r\n");// new getMethodfor (Entry<CharSequence, CharSequence> entry : request.headers()) {responseContent.append("HEADER: " + entry.getKey() + '=' + entry.getValue() + "\r\n");}responseContent.append("\r\n\r\n");// new getMethodSet<Cookie> cookies;String value = request.headers().getAndConvert(HttpHeaderNames.COOKIE);if (value == null) {cookies = Collections.emptySet();} else {cookies = ServerCookieDecoder.decode(value);}for (Cookie cookie : cookies) {responseContent.append("COOKIE: " + cookie + "\r\n");}responseContent.append("\r\n\r\n");QueryStringDecoder decoderQuery = new QueryStringDecoder(request.uri());Map<String, List<String>> uriAttributes = decoderQuery.parameters();for (Entry<String, List<String>> attr: uriAttributes.entrySet()) {for (String attrVal: attr.getValue()) {responseContent.append("URI: " + attr.getKey() + '=' + attrVal + "\r\n");}}responseContent.append("\r\n\r\n");// if GET Method: should not try to create a HttpPostRequestDecoderif (request.method().equals(HttpMethod.GET)) {// GET Method: should not try to create a HttpPostRequestDecoder// So stop hereresponseContent.append("\r\n\r\nEND OF GET CONTENT\r\n");// Not now: LastHttpContent will be sent writeResponse(ctx.channel());return;}try {decoder = new HttpPostRequestDecoder(factory, request);} catch (ErrorDataDecoderException e1) {e1.printStackTrace();responseContent.append(e1.getMessage());writeResponse(ctx.channel());ctx.channel().close();return;}readingChunks = HttpHeaderUtil.isTransferEncodingChunked(request);responseContent.append("Is Chunked: " + readingChunks + "\r\n");responseContent.append("IsMultipart: " + decoder.isMultipart() + "\r\n");if (readingChunks) {// Chunk versionresponseContent.append("Chunks: ");readingChunks = true;}}// check if the decoder was constructed before// if not it handles the form getif (decoder != null) {if (msg instanceof HttpContent) {// New chunk is receivedHttpContent chunk = (HttpContent) msg;try {decoder.offer(chunk);} catch (ErrorDataDecoderException e1) {e1.printStackTrace();responseContent.append(e1.getMessage());writeResponse(ctx.channel());ctx.channel().close();return;}responseContent.append('o');// example of reading chunk by chunk (minimize memory usage due to// Factory)readHttpDataChunkByChunk();// example of reading only if at the endif (chunk instanceof LastHttpContent) {writeResponse(ctx.channel());readingChunks = false;reset();}}} else {writeResponse(ctx.channel());}}private void reset() {request = null;// destroy the decoder to release all resourcesdecoder.destroy();decoder = null;}/*** Example of reading request by chunk and getting values from chunk to chunk*/private void readHttpDataChunkByChunk() throws Exception {try {while (decoder.hasNext()) {InterfaceHttpData data = decoder.next();if (data != null) {try {// new valuewriteHttpData(data);} finally {data.release();}}}} catch (EndOfDataDecoderException e1) {// endresponseContent.append("\r\n\r\nEND OF CONTENT CHUNK BY CHUNK\r\n\r\n");}}private void writeHttpData(InterfaceHttpData data) throws Exception {if (data.getHttpDataType() == HttpDataType.Attribute) {Attribute attribute = (Attribute) data;String value;try {value = attribute.getValue();} catch (IOException e1) {// Error while reading data from File, only print name and errore1.printStackTrace();responseContent.append("\r\nBODY Attribute: " + attribute.getHttpDataType().name() + ": "+ attribute.getName() + " Error while reading value: " + e1.getMessage() + "\r\n");return;}if (value.length() > 100) {responseContent.append("\r\nBODY Attribute: " + attribute.getHttpDataType().name() + ": "+ attribute.getName() + " data too long\r\n");} else {responseContent.append("\r\nBODY Attribute: " + attribute.getHttpDataType().name() + ": "+ attribute + "\r\n");}} else {responseContent.append("\r\n -----------start-------------" + "\r\n");responseContent.append("\r\nBODY FileUpload: " + data.getHttpDataType().name() + ": " + data+ "\r\n");responseContent.append("\r\n ------------end------------" + "\r\n");if (data.getHttpDataType() == HttpDataType.FileUpload) {FileUpload fileUpload = (FileUpload) data;if (fileUpload.isCompleted()) {System.out.println("file name : " + fileUpload.getFilename());System.out.println("file length: " + fileUpload.length());System.out.println("file maxSize : " + fileUpload.getMaxSize());System.out.println("file path :" + fileUpload.getFile().getPath());System.out.println("file absolutepath :" + fileUpload.getFile().getAbsolutePath());System.out.println("parent path :" + fileUpload.getFile().getParentFile());if (fileUpload.length() < 1024*1024*10) {responseContent.append("\tContent of file\r\n");try {//responseContent.append(fileUpload.getString(fileUpload.getCharset()));} catch (Exception e1) {// do nothing for the examplee1.printStackTrace();}responseContent.append("\r\n");} else {responseContent.append("\tFile too long to be printed out:" + fileUpload.length() + "\r\n");}// fileUpload.isInMemory();// tells if the file is in Memory// or on FilefileUpload.renameTo(new File(fileUpload.getFile().getPath())); // enable to move into another// File dest//decoder.removeFileUploadFromClean(fileUpload); //remove// the File of to delete file} else {responseContent.append("\tFile to be continued but should not!\r\n");}}}}private void writeResponse(Channel channel) {// Convert the response content to a ChannelBuffer.ByteBuf buf = copiedBuffer(responseContent.toString(), CharsetUtil.UTF_8);responseContent.setLength(0);// Decide whether to close the connection or not.boolean close = request.headers().contains(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE, true)|| request.protocolVersion().equals(HttpVersion.HTTP_1_0)&& !request.headers().contains(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE, true);// Build the response object.FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);response.headers().set(HttpHeaderNames.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.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, buf.readableBytes());}Set<Cookie> cookies;String value = request.headers().getAndConvert(HttpHeaderNames.COOKIE);if (value == null) {cookies = Collections.emptySet();} else {cookies = ServerCookieDecoder.decode(value);}if (!cookies.isEmpty()) {// Reset the cookies if necessary.for (Cookie cookie : cookies) {response.headers().add(HttpHeaderNames.SET_COOKIE, ServerCookieEncoder.encode(cookie));}}// Write the response.ChannelFuture future = channel.writeAndFlush(response);// Close the connection after the write operation is done if necessary.if (close) {future.addListener(ChannelFutureListener.CLOSE);}}private void writeMenu(ChannelHandlerContext ctx) {// print several HTML forms// Convert the response content to a ChannelBuffer.responseContent.setLength(0);// create Pseudo MenuresponseContent.append("<html>");responseContent.append("<head>");responseContent.append("<title>Netty Test Form</title>\r\n");responseContent.append("</head>\r\n");responseContent.append("<body bgcolor=white><style>td{font-size: 12pt;}</style>");responseContent.append("<table border=\"0\">");responseContent.append("<tr>");responseContent.append("<td>");responseContent.append("<h1>Netty Test Form</h1>");responseContent.append("Choose one FORM");responseContent.append("</td>");responseContent.append("</tr>");responseContent.append("</table>\r\n");// GETresponseContent.append("<CENTER>GET FORM<HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");responseContent.append("<FORM ACTION=\"/formget\" METHOD=\"GET\">");responseContent.append("<input type=hidden name=getform value=\"GET\">");responseContent.append("<table border=\"0\">");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"info\" size=10></td></tr>");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"secondinfo\" size=20>");responseContent.append("<tr><td>Fill with value: <br> <textarea name=\"thirdinfo\" cols=40 rows=10></textarea>");responseContent.append("</td></tr>");responseContent.append("<tr><td><INPUT TYPE=\"submit\" NAME=\"Send\" VALUE=\"Send\"></INPUT></td>");responseContent.append("<td><INPUT TYPE=\"reset\" NAME=\"Clear\" VALUE=\"Clear\" ></INPUT></td></tr>");responseContent.append("</table></FORM>\r\n");responseContent.append("<CENTER><HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");// POSTresponseContent.append("<CENTER>POST FORM<HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");responseContent.append("<FORM ACTION=\"/formpost\" METHOD=\"POST\">");responseContent.append("<input type=hidden name=getform value=\"POST\">");responseContent.append("<table border=\"0\">");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"info\" size=10></td></tr>");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"secondinfo\" size=20>");responseContent.append("<tr><td>Fill with value: <br> <textarea name=\"thirdinfo\" cols=40 rows=10></textarea>");responseContent.append("<tr><td>Fill with file (only file name will be transmitted): <br> "+ "<input type=file name=\"myfile\">");responseContent.append("</td></tr>");responseContent.append("<tr><td><INPUT TYPE=\"submit\" NAME=\"Send\" VALUE=\"Send\"></INPUT></td>");responseContent.append("<td><INPUT TYPE=\"reset\" NAME=\"Clear\" VALUE=\"Clear\" ></INPUT></td></tr>");responseContent.append("</table></FORM>\r\n");responseContent.append("<CENTER><HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");// POST with enctype="multipart/form-data"responseContent.append("<CENTER>POST MULTIPART FORM<HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");responseContent.append("<FORM ACTION=\"/formpostmultipart\" ENCTYPE=\"multipart/form-data\" METHOD=\"POST\">");responseContent.append("<input type=hidden name=getform value=\"POST\">");responseContent.append("<table border=\"0\">");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"info\" size=10></td></tr>");responseContent.append("<tr><td>Fill with value: <br> <input type=text name=\"secondinfo\" size=20>");responseContent.append("<tr><td>Fill with value: <br> <textarea name=\"thirdinfo\" cols=40 rows=10></textarea>");responseContent.append("<tr><td>Fill with file: <br> <input type=file name=\"myfile\">");responseContent.append("</td></tr>");responseContent.append("<tr><td><INPUT TYPE=\"submit\" NAME=\"Send\" VALUE=\"Send\"></INPUT></td>");responseContent.append("<td><INPUT TYPE=\"reset\" NAME=\"Clear\" VALUE=\"Clear\" ></INPUT></td></tr>");responseContent.append("</table></FORM>\r\n");responseContent.append("<CENTER><HR WIDTH=\"75%\" NOSHADE color=\"blue\"></CENTER>");responseContent.append("</body>");responseContent.append("</html>");ByteBuf buf = copiedBuffer(responseContent.toString(), CharsetUtil.UTF_8);// Build the response object.FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, buf.readableBytes());// Write the response.ctx.channel().writeAndFlush(response);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {logger.log(Level.WARNING, responseContent.toString(), cause);ctx.channel().close();}
}public class HttpUploadServerInitializer extends ChannelInitializer<SocketChannel> {private final SslContext sslCtx;public HttpUploadServerInitializer(SslContext sslCtx) {this.sslCtx = sslCtx;}@Overridepublic void initChannel(SocketChannel ch) {ChannelPipeline pipeline = ch.pipeline();if (sslCtx != null) {pipeline.addLast(sslCtx.newHandler(ch.alloc()));}pipeline.addLast(new HttpRequestDecoder());pipeline.addLast(new HttpResponseEncoder());// Remove the following line if you don't want automatic content compression.pipeline.addLast(new HttpContentCompressor());pipeline.addLast(new HttpUploadServerHandler());}
}

文件下载

Http协议请求方式：
- GET:请求Request-URI所标识的资源
- POST：在Request-URI所标识的资源附加的提交数据
- HEAD：请求获取由Request-URI所标识的只有的响应消息头
- PUT：请求服务器存储一个资源，并用Request-URI作为标识
- DELETE：请求服务器删除Request—URI所标识的资源
- TRACE：请求服务器回送收到的请求信息，主要是测试和诊断使用（@trace）
- CONNECT：保留将来使用
- OPTIONS：请求查询服务器的性能，查询相关资源的选项和需求。

Http协议的响应消息：响应消息由三部分组成：状态行、消息头、响应正文。
响应状态分类：
- 1xx：提示信息，表示请求已经接收继续处理。
- 2xx：成功。表示qq已经接收成功。
- 3xx：重定向。要完成的请求必须要更进一步的操作。
- 4xx：客户端错误，可能是请求语法错误或者qq无法实现。
- 5xx：服务器端错误。服务器未能处理请求（内部出现异常）。

常见状态码：
- 200 OK 成功。
- 400 Bad Request 错误的请求语法，不被服务器理解。
- 401 Unauthorized：请求未经授权。
- 403 Forbidden：服务器收到请求，但请求被服务器拒绝。
- 404 Not Found 请求资源不存在。
- 405 Method Not Allowed：请求方式不被允许，如只支持get请求，但客户端使用了post请求
- 500 Inernal Server Error ：服务器发送不可预测的错误。
- 503 Server Unavailable：服务器当前不能处理客户端请求，一段时间后恢复正常。

public class HttpFileServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {private final String url;public HttpFileServerHandler(String url) {this.url = url;}@Overridepublic void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {//对请求的解码结果进行判断：if (!request.decoderResult().isSuccess()) {// 400sendError(ctx, BAD_REQUEST);return;}//对请求方式进行判断：如果不是get方式（如post方式）则返回异常if (request.method() != GET) {// 405sendError(ctx, METHOD_NOT_ALLOWED);return;}//获取请求uri路径final String uri = request.uri();//对url进行分析，返回本地系统final String path = sanitizeUri(uri);//如果 路径构造不合法，则path为nullif (path == null) {//403sendError(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()) {// 403sendError(ctx, FORBIDDEN);return;}//随机文件读写类RandomAccessFile randomAccessFile = null;try {randomAccessFile = new RandomAccessFile(file, "r");// 以只读的方式打开文件} catch (FileNotFoundException fnfe) {// 404sendError(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_ALIVEif (HttpHeaderUtil.isKeepAlive(request)) {response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);}//进行写出ctx.write(response);//构造发送文件线程，将文件写入到Chunked缓冲区中ChannelFuture sendFileFuture;//写出ChunkedFilesendFileFuture = ctx.write(new ChunkedFile(randomAccessFile, 0, fileLength, 8192), ctx.newProgressivePromise());//添加传输监听sendFileFuture.addListener(new ChannelProgressiveFutureListener() {@Overridepublic void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {if (total < 0) { // total unknownSystem.err.println("Transfer progress: " + progress);} else {System.err.println("Transfer progress: " + progress + " / " + total);}}@Overridepublic 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);}}@Overridepublic 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(".*[<>&\"].*");/*** <B>方法名称：</B>解析URI<BR>* <B>概要说明：</B>对URI进行分析<BR>* @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-1uri = 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("<!DOCTYPE html>\r\n");ret.append("<html><head><title>");ret.append(dirPath);ret.append(" 目录：");ret.append("</title></head><body>\r\n");ret.append("<h3>");ret.append(dirPath).append(" 目录：");ret.append("</h3>\r\n");ret.append("<ul>");ret.append("<li>链接：<a href=\"../\">..</a></li>\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("<li>链接：<a href=\"");ret.append(name);ret.append("\">");ret.append(name);ret.append("</a></li>\r\n");}ret.append("</ul></body></html>\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()));}
}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<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch)throws Exception {// 加入http的解码器ch.pipeline().addLast("http-decoder", new HttpRequestDecoder());// 加入ObjectAggregator解码器，作用是他会把多个消息转换为单一的FullHttpRequest或者FullHttpResponsech.pipeline().addLast("http-aggregator", new HttpObjectAggregator(65536));// 加入http的编码器ch.pipeline().addLast("http-encoder", new HttpResponseEncoder());// 加入chunked 主要作用是支持异步发送的码流（大文件传输），但不专用过多的内存，防止java内存溢出ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());// 加入自定义处理文件服务器的业务逻辑handlerch.pipeline().addLast("fileServerHandler",new HttpFileServerHandler(url));}});ChannelFuture future = b.bind("192.168.134.1", port).sync();System.out.println("HTTP文件目录服务器启动，网址是 : " + "http://192.168.134.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);}
}

心跳检测

使用Socket通信一般经常会处理多个服务器之间的心跳检测，一般来讲维护服务器集群，要有一台或多台服务器主机，然后还应该有N台（Slave），主服务器要时刻知道从服务器的情况，进行实时监控，这个在分布式架构中叫心跳检测或者心跳监控。

/*** Marshalling工厂*/
public final class MarshallingCodeCFactory {/*** 创建Jboss Marshalling解码器MarshallingDecoder* @return MarshallingDecoder*/public static MarshallingDecoder buildMarshallingDecoder() {//首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");//创建了MarshallingConfiguration对象，配置了版本号为5 final MarshallingConfiguration configuration = new MarshallingConfiguration();configuration.setVersion(5);//根据marshallerFactory和configuration创建providerUnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);//构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1);return decoder;}/*** 创建Jboss Marshalling编码器MarshallingEncoder* @return MarshallingEncoder*/public static MarshallingEncoder buildMarshallingEncoder() {final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");final MarshallingConfiguration configuration = new MarshallingConfiguration();configuration.setVersion(5);MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);//构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组MarshallingEncoder encoder = new MarshallingEncoder(provider);return encoder;}
}public class RequestInfo implements Serializable {private String ip ;private HashMap<String, Object> cpuPercMap ;private HashMap<String, Object> memoryMap;public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public HashMap<String, Object> getCpuPercMap() {return cpuPercMap;}public void setCpuPercMap(HashMap<String, Object> cpuPercMap) {this.cpuPercMap = cpuPercMap;}public HashMap<String, Object> getMemoryMap() {return memoryMap;}public void setMemoryMap(HashMap<String, Object> memoryMap) {this.memoryMap = memoryMap;}}

public class Server {public static void main(String[] args) throws Exception{EventLoopGroup pGroup = new NioEventLoopGroup();EventLoopGroup cGroup = new NioEventLoopGroup();ServerBootstrap b = new ServerBootstrap();b.group(pGroup, cGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)//设置日志.handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel sc) throws Exception {sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());sc.pipeline().addLast(new ServerHeartBeatHandler());}});ChannelFuture cf = b.bind(8765).sync();cf.channel().closeFuture().sync();pGroup.shutdownGracefully();cGroup.shutdownGracefully();}
}public class ServerHeartBeatHandler extends ChannelHandlerAdapter {/** key:ip value:auth */private static HashMap<String, String> AUTH_IP_MAP = new HashMap<String, String>();private static final String SUCCESS_KEY = "auth_success_key";static {AUTH_IP_MAP.put("192.168.134.1", "1234");}private boolean auth(ChannelHandlerContext ctx, Object msg){//System.out.println(msg);String [] ret = ((String) msg).split(",");String auth = AUTH_IP_MAP.get(ret[0]);if(auth != null && auth.equals(ret[1])){ctx.writeAndFlush(SUCCESS_KEY);return true;} else {ctx.writeAndFlush("auth failure !").addListener(ChannelFutureListener.CLOSE);return false;}}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if(msg instanceof String){auth(ctx, msg);} else if (msg instanceof RequestInfo) {RequestInfo info = (RequestInfo) msg;System.out.println("--------------------------------------------");System.out.println("当前主机ip为: " + info.getIp());System.out.println("当前主机cpu情况: ");HashMap<String, Object> cpu = info.getCpuPercMap();System.out.println("总使用率: " + cpu.get("combined"));System.out.println("用户使用率: " + cpu.get("user"));System.out.println("系统使用率: " + cpu.get("sys"));System.out.println("等待率: " + cpu.get("wait"));System.out.println("空闲率: " + cpu.get("idle"));System.out.println("当前主机memory情况: ");HashMap<String, Object> memory = info.getMemoryMap();System.out.println("内存总量: " + memory.get("total"));System.out.println("当前内存使用量: " + memory.get("used"));System.out.println("当前内存剩余量: " + memory.get("free"));System.out.println("--------------------------------------------");ctx.writeAndFlush("info received!");} else {ctx.writeAndFlush("connect failure!").addListener(ChannelFutureListener.CLOSE);}}}public class Client {public static void main(String[] args) throws Exception{EventLoopGroup group = new NioEventLoopGroup();Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel sc) throws Exception {sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());sc.pipeline().addLast(new ClienHeartBeattHandler());}});ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();cf.channel().closeFuture().sync();group.shutdownGracefully();}
}public class ClienHeartBeattHandler extends ChannelHandlerAdapter {private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);private ScheduledFuture<?> heartBeat;//主动向服务器发送认证信息private InetAddress addr ;private static final String SUCCESS_KEY = "auth_success_key";@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {addr = InetAddress.getLocalHost();String ip = addr.getHostAddress();String key = "1234";//证书String auth = ip + "," + key;ctx.writeAndFlush(auth);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {try {if(msg instanceof String){String ret = (String)msg;if(SUCCESS_KEY.equals(ret)){// 握手成功，主动发送心跳消息this.heartBeat = this.scheduler.scheduleWithFixedDelay(new HeartBeatTask(ctx), 0, 2, TimeUnit.SECONDS);System.out.println(msg);                }else {System.out.println(msg);}}} finally {ReferenceCountUtil.release(msg);}}private class HeartBeatTask implements Runnable {private final ChannelHandlerContext ctx;public HeartBeatTask(final ChannelHandlerContext ctx) {this.ctx = ctx;}@Overridepublic void run() {try {RequestInfo info = new RequestInfo();//ipinfo.setIp(addr.getHostAddress());Sigar sigar = new Sigar();//cpu precCpuPerc cpuPerc = sigar.getCpuPerc();HashMap<String, Object> cpuPercMap = new HashMap<String, Object>();cpuPercMap.put("combined", cpuPerc.getCombined());cpuPercMap.put("user", cpuPerc.getUser());cpuPercMap.put("sys", cpuPerc.getSys());cpuPercMap.put("wait", cpuPerc.getWait());cpuPercMap.put("idle", cpuPerc.getIdle());// memoryMem mem = sigar.getMem();HashMap<String, Object> memoryMap = new HashMap<String, Object>();memoryMap.put("total", mem.getTotal() / 1024L);memoryMap.put("used", mem.getUsed() / 1024L);memoryMap.put("free", mem.getFree() / 1024L);info.setCpuPercMap(cpuPercMap);info.setMemoryMap(memoryMap);ctx.writeAndFlush(info);} catch (Exception e) {e.printStackTrace();}}public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();if (heartBeat != null) {heartBeat.cancel(true);heartBeat = null;}ctx.fireExceptionCaught(cause);}}
}