使用Netty開發Http服務

1. 關於netty

   1. netty主要特性
      1. NIO
      2. 事件驅動
2. 主服務編寫

   1. package com.myserver.receive;
      
      import io.netty.bootstrap.ServerBootstrap;
      import io.netty.channel.ChannelFuture;
      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.logging.LogLevel;
      import io.netty.handler.logging.LoggingHandler;
      
      public class ReceiveServer {
          
          private  int port;
          
          public ReceiveServer(int port){
              this.port = port;
          }
          
          public void run() throws Exception{
              EventLoopGroup boosGroup = new NioEventLoopGroup();
              EventLoopGroup workerGroup = new NioEventLoopGroup();
              
              try{
                  //輔助啓動類
                  ServerBootstrap b = new ServerBootstrap();
                  
                  b.option(ChannelOption.SO_BACKLOG, 1024);
                  b.group(boosGroup, workerGroup)
                          .channel(NioServerSocketChannel.class)
                          .handler(new LoggingHandler(LogLevel.INFO))
                          .childHandler(new ServerInitializer());
                  
                      ChannelFuture f = b.bind(port).sync();
                      
                      f.channel().closeFuture().sync();
                  
              }catch(Exception e){
                  
              }finally {
                  workerGroup.shutdownGracefully();
                  boosGroup.shutdownGracefully();
              }
              
          }
          
      }

      package com.myserver.receive;
      
      import io.netty.channel.ChannelInitializer;
      import io.netty.channel.ChannelPipeline;
      import io.netty.channel.socket.SocketChannel;
      import io.netty.handler.codec.http.*;
      import io.netty.handler.stream.ChunkedWriteHandler;
      
      public class ServerInitializer extends ChannelInitializer<SocketChannel> {
      
          @Override
          public void initChannel(SocketChannel ch) {
              ChannelPipeline p = ch.pipeline();
      
              // HttpServerCodec is a combination of HttpRequestDecoder and HttpResponseEncoder
              p.addLast(new HttpServerCodec());
      
              // add gizp compressor for http response content
              p.addLast(new HttpContentCompressor());
      
              p.addLast(new HttpObjectAggregator(1048576));
      
              p.addLast(new ChunkedWriteHandler());
      
              p.addLast(new DiscardServerHandler());
          }
      }

      package com.myserver.receive;
      
      
      import static io.netty.handler.codec.http.HttpHeaderNames.CONNECTION;
      import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
      import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
      
      import com.myserver.routehandler.RouteHandler;
      
      import io.netty.buffer.Unpooled;
      import io.netty.channel.ChannelHandlerContext;
      import io.netty.channel.ChannelInboundHandlerAdapter;
      import io.netty.handler.codec.http.DefaultFullHttpResponse;
      import io.netty.handler.codec.http.FullHttpResponse;
      import io.netty.handler.codec.http.HttpHeaders;
      import io.netty.handler.codec.http.HttpHeaders.Values;
      import io.netty.handler.codec.http.HttpRequest;
      import io.netty.handler.codec.http.HttpResponseStatus;
      import io.netty.handler.codec.http.HttpVersion;
      import io.netty.util.AsciiString;
      
      public class DiscardServerHandler extends ChannelInboundHandlerAdapter {
          private HttpRequest request;
          
          /* (non-Javadoc)
           * @see io.netty.channel.ChannelInboundHandlerAdapter#channelRead(io.netty.channel.ChannelHandlerContext, java.lang.Object)
           */
          @Override
          public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
              
              
              if (msg instanceof HttpRequest) {
                  request = (HttpRequest) msg;
                  String uri = request.uri();
                  System.out.println("Uri:" + uri);
              /*}
              if (msg instanceof HttpContent) {
                  HttpContent content = (HttpContent) msg;
                  ByteBuf buf = content.content();
                  System.out.println(buf.toString(CharsetUtil.UTF_8));
                  buf.release();*/
      
               //   String res = "I am OK";
                  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                         // HttpResponseStatus.OK, Unpooled.wrappedBuffer(res.getBytes("UTF-8")));
                  HttpResponseStatus.OK, Unpooled.wrappedBuffer(RouteHandler.transfer(ctx, msg)));
                  response.headers().set(CONTENT_TYPE, new AsciiString("application/json; charset=utf-8"));
                  response.headers().set(CONTENT_LENGTH,
                          response.content().readableBytes());
                  if (HttpHeaders.isKeepAlive(request)) {
                      response.headers().set(CONNECTION, Values.KEEP_ALIVE);
                  }
                  ctx.write(response);
                  ctx.flush();
              }
              
             /* if(msg instanceof HttpRequest ){
                  HttpRequest request   = (HttpRequest)msg;
                  System.out.println(request.uri());
                 // ByteBufAllocator alloc = new PooledByteBufAllocator();
                  ByteBuf buffer  = Unpooled.directBuffer(1024);
                  HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,buffer);
                  buffer.writeBytes("hello!! 你好".getBytes("UTF-8"));
                  ByteBuf content = ((DefaultFullHttpResponse) response).content();
                  HttpHeaders headers = response.headers();
                  headers.set("Content-Type", "text/html; charset=UTF-8");
                  headers.set("Content-Length", content.writerIndex());
                  response.setHeader("Content-Type", "text/html; charset=UTF-8");
                  response.setHeader("Content-Length", response.getContent().writerIndex());
                  Channel ch = ctx.channel();
                  // Write the initial line and the header.
                  ch.write(response);
                  ch.disconnect();
                  ch.close();
              }
            */
              /*ByteBuf in = (ByteBuf)msg;
              try{
                  while(in.isReadable()){
                      System.out.println((char)in.readByte());
                      System.out.flush();
                  }
              }finally {
                  ReferenceCountUtil.release(msg);
              }*/
          }
      
          /* (non-Javadoc)
           * @see io.netty.channel.ChannelInboundHandlerAdapter#exceptionCaught(io.netty.channel.ChannelHandlerContext, java.lang.Throwable)
           */
          @Override
          public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
              cause.printStackTrace();
              ctx.close();
          }
          
          public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
              ctx.flush();
          }
          
      }

      路由處理類編寫

      package com.myserver.routehandler;
      
      import java.lang.reflect.Constructor;
      import java.lang.reflect.InvocationTargetException;
      import java.lang.reflect.Method;
      
      import org.apache.commons.logging.Log;
      import org.apache.commons.logging.LogFactory;
      import org.json.JSONObject;
      
      import com.myserver.error.ErrorHandler;
      import com.myserver.protocol.DcProtocol;
      import com.myserver.state.StatusCode;
      import com.myserver.util.Config;
      
      import io.netty.channel.ChannelHandlerContext;
      
      
      
      public class RouteHandler {
          
          private static final Log logger = LogFactory.getLog(RouteHandler.class);
          
          public static byte[] transfer(ChannelHandlerContext ctx, Object msg) {
              DcProtocol apiProtocol = new DcProtocol(ctx, msg);
      
              if (apiProtocol.getEndpoint() == null) {
                  return encode(ErrorHandler.error(StatusCode.API_CAN_NOT_BE_NULL));
              }
      
              if (apiProtocol.getApi() == null) {
                  return encode(ErrorHandler.error(StatusCode.API_NOT_FOUND));
              }
      
              Object result = invoke(apiProtocol.getApi(), apiProtocol);
              if (result == null) {
                  return encode(ErrorHandler.error(StatusCode.UNKNOWN_ERROR));
              }
      
              return encode(result);
          }
      
          /**
           * invoke api resource method by apiName, but the request apiProtocol should observe routeMap regulations
           *
           * @param apiName
           * @param apiProtocol
           * @return
           */
          public static Object invoke(String apiName, DcProtocol apiProtocol) {
              Class<?> classname;
              Object   classObject;
              Constructor constructor;
              Method   method;
              Object   result = null;
      
              Route api = RouteReader.RouteMap.get(apiName);
              if (api == null) {
                  return ErrorHandler.error(StatusCode.API_NOT_FOUND);
              }
      
              if (apiProtocol.getBuild() < api.getBuild()){
                  return ErrorHandler.error(StatusCode.VERSION_IS_TOO_LOW);
              }
      
              if(api.getHttpMethod() != null && !api.getHttpMethod().contains(apiProtocol.getMethod().toString().toLowerCase())){
                  return ErrorHandler.error(StatusCode.REQUEST_MODE_ERROR);
              }
      
              try {
                  classname = Class.forName(Config.getString("resource.package.name") + "." + api.getResource());
                  constructor = classname.getConstructor(DcProtocol.class);
                  classObject = constructor.newInstance(apiProtocol);
              } catch (NoSuchMethodException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              } catch (ClassNotFoundException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              } catch (InvocationTargetException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              } catch (InstantiationException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              } catch (IllegalAccessException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              }
      
              try {
                  method = classname.getMethod(apiProtocol.getMethod().toString().toLowerCase());
              } catch (NoSuchMethodException e) {
                  logger.error(e.getMessage());
                  return ErrorHandler.error(StatusCode.API_SERVER_ERROR);
              }
      
              try {
                  result = method.invoke(classObject);
              } catch (InvocationTargetException e) {
                  e.printStackTrace();
                  logger.error(e.getMessage());
              } catch (IllegalAccessException e) {
                  logger.error(e.toString());
              }
      
              return result;
          }
      
          /**
           * exchange the api resource returns to a JSONObject
           *
           * @param object
           * @return
           */
          public static byte[] encode(Object object) {
              String data = new JSONObject(object).toString();
              data = filter(data);
              return data.getBytes();
          }
      
          /**
           * we always need filter something for some reason,
           * otherwise we can replace the timestamp to the string we defined, and so on.
           *
           * @param data
           * @return
           */
          public static String filter(String data){
              return data;
          }
      
          
          
      }

       

3. 自定義協議編寫

package com.myserver.protocol;

import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.myserver.routehandler.Route;
import com.myserver.routehandler.RouteReader;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
import io.netty.handler.codec.http.multipart.InterfaceHttpData;
import io.netty.handler.codec.http.multipart.MixedAttribute;
import io.netty.util.CharsetUtil;

public class DcProtocol {
    
    private static final Log log = LogFactory.getLog(DcProtocol.class);
    private String url;  //url
    private int build = 101;
    private String version = "1.0";
    private String                    clientIP   = null;
    private String                    serverIP   = null;
    private String                    api        = null;
    private String                    endpoint   = null;
    private String                    auth       = null;
    private int                       offset     = 0;
    private int                       limit      = 10;
    private HttpMethod                method     = HttpMethod.GET;
    private Map<String, List<String>> parameters = new HashMap<String, List<String>>(); // get 和 post 的鍵值對都存儲在這裏
    private String                    postBody   = null; // post 請求時的非鍵值對內容
    /**
     * @return the url
     */
    public String getUrl() {
        return url;
    }
    /**
     * @return the build
     */
    public int getBuild() {
        return build;
    }
    /**
     * @return the version
     */
    public String getVersion() {
        return version;
    }
    /**
     * @return the clientIP
     */
    public String getClientIP() {
        return clientIP;
    }
    /**
     * @return the serverIP
     */
    public String getServerIP() {
        return serverIP;
    }
    /**
     * @return the api
     */
    public String getApi() {
        return api;
    }
    /**
     * @return the endpoint
     */
    public String getEndpoint() {
        return endpoint;
    }
    /**
     * @return the auth
     */
    public String getAuth() {
        return auth;
    }
    /**
     * @return the offset
     */
    public int getOffset() {
        return offset;
    }
    /**
     * @return the limit
     */
    public int getLimit() {
        return limit;
    }
    /**
     * @return the method
     */
    public HttpMethod getMethod() {
        return method;
    }
    /**
     * @return the parameters
     */
    public Map<String, List<String>> getParameters() {
        return parameters;
    }
    /**
     * @return the postBody
     */
    public String getPostBody() {
        return postBody;
    }
    
    //構造方法
    public DcProtocol(ChannelHandlerContext ctx, Object msg){
        HttpRequest req = (HttpRequest)msg;
        String uri = req.uri();
        
        if(uri.length()<=0){
            return;
        }
        log.info(uri);
        this.method = req.method();
        praseEndpoint(uri);
        setIp(ctx, req);
        queryStringHandler(uri);
        
        requestParametersHandler(req);
        requestBodyHandler(msg);

        if (this.parameters.size() > 0) {
            setFields();
        }
    }
    
    /**
     * 功能：解析url
     * @param uri
     */
    private void praseEndpoint(String uri){
        String endPoint = uri.split("\\?")[0];
        if(endPoint.endsWith("/")){
            endPoint = endPoint.substring(0,endPoint.length());
        }
        
        this.endpoint = endPoint;
        Set<Entry<String, Route>> entrySet = RouteReader.RouteMap.entrySet();
        for (Entry<String, Route> entry : entrySet) {
            Route route = entry.getValue();
            Pattern pattern = Pattern.compile("^" + route.getRegex() + "$");
            Matcher matcher = pattern.matcher(endpoint);
            if (matcher.find()) {
                this.api = route.getName();
                if (matcher.groupCount() > 0) {
                    for (int i = 0; i < matcher.groupCount(); i++) {
                        addParameter(route.getParameterNames().get(i), matcher.group(i + 1));
                    }
                }
                break;
            }
            
        }
        
    }
    
    private void addParameter(String key, String param) {
        List<String> params = new ArrayList<>();
        params.add(param);
        this.parameters.put(key, params);
    }

    private void setIp(ChannelHandlerContext ctx, HttpRequest req) {
        String clientIP = (String) req.headers().get("X-Forwarded-For");
        if (clientIP == null) {
            InetSocketAddress remoteSocket = (InetSocketAddress) ctx.channel().remoteAddress();
            clientIP = remoteSocket.getAddress().getHostAddress();
        }
        this.clientIP = clientIP;

        InetSocketAddress serverSocket = (InetSocketAddress) ctx.channel().localAddress();
        this.serverIP = serverSocket.getAddress().getHostAddress();
    }
    
    private void queryStringHandler(String uri) {

        QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri);
        if (queryStringDecoder.parameters().size() > 0) {
            this.parameters.putAll(queryStringDecoder.parameters());
        }
    }
    
    private void requestParametersHandler(HttpRequest req) {
        if (req.method().equals(HttpMethod.POST)) {
            HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(req);
            try {
                List<InterfaceHttpData> postList = decoder.getBodyHttpDatas();
                for (InterfaceHttpData data : postList) {
                    List<String> values = new ArrayList<String>();
                    MixedAttribute value = (MixedAttribute) data;
                    value.setCharset(CharsetUtil.UTF_8);
                    values.add(value.getValue());
                    this.parameters.put(data.getName(), values);
                }
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        }
    }
    
    private void requestBodyHandler(Object msg) {
        if (msg instanceof HttpContent) {
            HttpContent httpContent = (HttpContent) msg;
            ByteBuf content = httpContent.content();
            StringBuilder buf = new StringBuilder();
            buf.append(content.toString(CharsetUtil.UTF_8));
            this.postBody = buf.toString();
        }
    }
    
    private void setFields() {
        Field[] fields = this.getClass().getDeclaredFields();

        for (int i = 0, length = fields.length; i < length; i++) {
            Field field = fields[i];
            String fieldName = field.getName();

            if (fieldName.equals("logger")
                    || fieldName.equals("method")
                    || fieldName.equals("parameters")
                    || fieldName.equals("postBody")) {
                continue;
            }

            if (!this.parameters.containsKey(fieldName)) {
                continue;
            }

            Class fieldType = field.getType();
            field.setAccessible(true);
            try {
                if (fieldType == int.class) {
                    field.set(this, Integer.parseInt(this.parameters.get(fieldName).get(0)));
                } else {
                    field.set(this, this.parameters.get(fieldName).get(0));
                }
            } catch (NumberFormatException | IllegalAccessException e) {
                log.error("field set error", e);
            }

            this.parameters.remove(fieldName);
        }
    }
}

 

路由配置文件讀取

package com.myserver.routehandler;

import java.util.HashMap;
import java.util.Map;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;



public class RouteReader {
    private static final Logger logger = LoggerFactory.getLogger(RouteReader.class);

    private static final String routeName = "/routeMap.xml";

    private static final String RouteNode       = "route";
    private static final String RouteName       = "name";
    private static final String RouteHttpMethod = "method";
    private static final String RouteResource   = "resource";
    private static final String RouteBuild      = "build";

    public static final Map<String, Route> RouteMap = new HashMap<String, Route>();

    static {
        init();
    }

    public static void init() {
        DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
        try {
            DocumentBuilder db = dbf.newDocumentBuilder();
            Document doc = db.parse(RouteReader.class.getResourceAsStream(routeName));

            NodeList RouteList = doc.getElementsByTagName(RouteNode);
            for (int i = 0, RouteLength = RouteList.getLength(); i < RouteLength; i++) {
                Element element = (Element) RouteList.item(i);

                Route Route = new Route();

                for (Node node = element.getFirstChild(); node != null; node = node.getNextSibling()) {
                    if (node.getNodeType() == Node.ELEMENT_NODE) {

                        String name = node.getNodeName();
                        String value = node.getFirstChild().getNodeValue();

                        switch (name) {
                            case RouteName:
                                Route.setName(value);
                                break;
                            case RouteHttpMethod:
                                Route.addHttpMethod(value);
                                break;
                            case RouteResource:
                                Route.setResource(value);
                                break;
                            case RouteBuild:
                                try {
                                    Route.setBuild(Integer.parseInt(value));
                                } catch (NumberFormatException e) {
                                    logger.error(e.getMessage());
                                }
                                break;
                            default:
                                break;
                        }
                    }
                }

                RouteMap.put(Route.getName(), Route);
            }
        } catch (Exception e) {
            logger.error(e.getMessage());
        }
    }
}

路由配置文件

<?xml version="1.0" encoding="UTF-8"?>
<routeMap>
    <route>
        <name>/user</name>
        <method>post</method>
        <resource>UserResource</resource>
    </route>
    <route>
        <name>/user/:uid</name>
        <method>get</method>
        <method>patch</method>
        <method>delete</method>
        <resource>UserResource</resource>
        <build>101</build>
    </route>
    <route>
        <name>/user/:uid/album/:aid</name>
        <method>get</method>
        <method>post</method>
        <resource>AlbumResource</resource>
        <build>102</build>
    </route>
    <route>
        <name>/sendData</name>
        <method>post</method>
        <method>get</method>
        <resource>ActionResource</resource>
        <build>101</build>
    </route>
</routeMap>