SpringBoot集成Netty实现http服务（类似SpingMvc的contoller层实现）

SpringBoot中使用Netty与spring中使用Netty没有差别，在Spring中使用Netty可以考虑Netty的启动时机，可以在Bean加载的时候启动，可以写一个自执行的函数启动，这里采用监听Spring容器的启动事件来启动Netty。

实现类似SpingMvc的contoller层实现:

添加依赖:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-tomcat</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.1.Final</version>
</dependency>
<dependency>
    <groupId>commons-lang</groupId>
    <artifactId>commons-lang</artifactId>
    <version>${commons.lang.version}</version>
</dependency>

排除tomcat的依赖

Netty Http服务端编写:

handler 处理类

@Component
@Slf4j
@ChannelHandler.Sharable //@Sharable 注解用来说明ChannelHandler是否可以在多个channel直接共享使用
public class HttpServerHandler  extends ChannelInboundHandlerAdapter {

    private static Map<String, Action> actionMapping = null;

    public Map<String, Action> getActionMapping(){
        if(actionMapping == null){
            return actionMapping = ClassLoaderUtil.buildActionMapping();
        }
        return actionMapping;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            if (msg instanceof FullHttpRequest) {
                FullHttpRequest req = (FullHttpRequest) msg;
                if (HttpUtil.is100ContinueExpected(req)) {
                    ctx.write(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
                }
                //封装请求和响应
                HttpRequestWrapper httpRequestWrapper = buildRequestWraper(req);
                //建造netty响应
                HttpResponseWrapper httpResponseWrapper = new HttpResponseWrapper();
                Action action = getActionMapping().get(httpRequestWrapper.getUri());
                if(action != null){
                    Object responseBody = null;
                    Object object = action.getMethod().invoke(action.getBean(),buildArgs(
                            action, httpRequestWrapper.getParams(),httpRequestWrapper));

                    if(StringUtils.isNotEmpty(action.getResponseType()) &&
                            action.getResponseType().equals("JSON")){
                        responseBody = JSON.toJSONString(object);
                    }else{
                        responseBody = object;
                    }
                    httpResponseWrapper.write(object.toString().getBytes("UTF-8"));
                }
                FullHttpResponse response = buildResponse(httpResponseWrapper);
                boolean keepAlive = HttpUtil.isKeepAlive(req);
                if (!keepAlive) {
                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
                } else {
                    response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
                }
            }
            //负责显式释放与池的ByteBuf实例相关联的内存,SimpleChannelInboundHandler会自动释放资源，因此无需显式释放
            ReferenceCountUtil.release(msg);
        } catch (Exception e) {
            log.error("system exception:{}", e);
        }
    }

    /**
     * 构建请求对象
     * @param req
     * @return
     */
    private HttpRequestWrapper buildRequestWraper(FullHttpRequest req) {
        HashMap<String, String> headersMap = new HashMap<String, String>(16);
        for (Map.Entry<String, String> entry : req.headers()) {
            headersMap.put(entry.getKey(), entry.getValue());
        }
        byte[] content = new byte[req.content().readableBytes()];
        req.content().readBytes(content);
        String url = req.uri();
        String params = "";
        if(url.indexOf("?")>0){
            String[] urls = url.split("\?");
            url = urls[0];
            params = urls[1];
        }
        return new HttpRequestWrapper(req.method().name(), url, headersMap, content, params);
    }

    /**
     * 构建响应对象
     * @param httpResponseWrapper
     * @return
     */
    private FullHttpResponse buildResponse(HttpResponseWrapper httpResponseWrapper) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                HttpResponseStatus.valueOf(httpResponseWrapper.getStatusCode()), Unpooled.wrappedBuffer(httpResponseWrapper.content()));
        HttpHeaders headers = response.headers();
        headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
        headers.set(HttpHeaderNames.CONTENT_TYPE, new AsciiString("application/json; charset=utf-8"));
        for (Map.Entry<String, String> entry : httpResponseWrapper.headers().entrySet()) {
            headers.set(entry.getKey(), entry.getValue());
        }
        return response;
    }


    /**
     * 构建请求参数
     * @param action
     * @param urlParams
     * @param httpRequestWrapper
     * @return
     */
    public Object[] buildArgs(Action action,String urlParams,HttpRequestWrapper httpRequestWrapper){
        if(action == null){
            return null;
        }
        LocalVariableTableParameterNameDiscoverer u = new LocalVariableTableParameterNameDiscoverer();
        //获取处理方法的参数
        String[] params = u.getParameterNames(action.getMethod());
        Object[] objects = new Object[params.length];
        Map<String,String> paramMap = new HashMap<>();
        try{
            if(StringUtils.isNotEmpty(urlParams)){
                paramMap = UrlUtils.URLRequest(urlParams);
            }
            if( httpRequestWrapper.content()!=null){
                Parameter[] parameters = action.getMethod().getParameters();
                for(Parameter parameter : parameters){
                    Annotation annotation = parameter.getAnnotation(ActionBody.class);
                    if(annotation == null){
                        continue;
                    }
                    int index = Integer.parseInt(parameter.getName().substring(3));
                    paramMap.put(params[index],new String(httpRequestWrapper.content(),"UTF-8"));
                }
            }

            for( int i = 0 ;i<params.length; i++){
                final int flag = i;
                paramMap.forEach((k,v)->{
                    if(k.equals(params[flag])){
                        objects[flag] = v;
                    }
                });
            }
        }catch(Exception e){
            log.error(e.getMessage());
        }
        return objects;
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }
    
    /**
     * 当客户端断开连接
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.close();
    }

}

controller层实现:

@Component
@ActionMapping(actionKey="controller")
@ResponseType
public class HttpNettyController implements BaseActionController{

    @ActionMapping(actionKey = "method")
    public String method(String a, String b){

        return String.format("a:%s,b:%s",a,b);
    }
}

ChannelPipeline 实现:

@Component
@ConditionalOnProperty(  //配置文件属性是否为true
        value = {"netty.http.enabled"},
        matchIfMissing = false
)
public class HttpPipeline extends ChannelInitializer<SocketChannel> {

    @Autowired
    HttpServerHandler httpServerHandler;


    @Override
    public void initChannel(SocketChannel ch) {
    //    log.error("test", this);
        ChannelPipeline p = ch.pipeline();        
        p.addLast(new HttpServerCodec());
        p.addLast(new HttpContentCompressor());
        p.addLast(new HttpObjectAggregator(1048576));
        p.addLast(new ChunkedWriteHandler());
        // http请求根处理器
        p.addLast(httpServerHandler);
    }
    
}

服务实现:

package cn.myframe.netty.server;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import cn.myframe.netty.pipeline.HttpPipeline;
import cn.myframe.properties.NettyHttpProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
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;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * HTTP服务
 * @version 创建时间：2019年6月25日 下午5:39:36
 */
@Configuration
@EnableConfigurationProperties({NettyHttpProperties.class})
@ConditionalOnProperty(  //配置文件属性是否为true
        value = {"netty.http.enabled"},
        matchIfMissing = false
)
@Slf4j
public class HttpServer {
    
        @Autowired
        HttpPipeline httpPipeline;

        @Autowired
        NettyHttpProperties nettyHttpProperties;

        @Bean("starHttpServer")
        public String start() {
            // 准备配置
            // HttpConfiguration.me().setContextPath(contextPath).setWebDir(webDir).config();
            // 启动服务器
           Thread thread =  new Thread(() -> {
                NioEventLoopGroup bossGroup =
                        new NioEventLoopGroup(nettyHttpProperties.getBossThreads());
                NioEventLoopGroup workerGroup =
                        new NioEventLoopGroup(nettyHttpProperties.getWorkThreads());
                try {
                    log.info("start netty [HTTP] server ,port: " + nettyHttpProperties.getPort());
                    ServerBootstrap boot = new ServerBootstrap();
                    options(boot).group(bossGroup, workerGroup)
                            .channel(NioServerSocketChannel.class)
                            .handler(new LoggingHandler(LogLevel.INFO))
                            .childHandler(httpPipeline);
                    Channel ch = null;
                  //是否绑定IP
                    if(StringUtils.isNotEmpty(nettyHttpProperties.getBindIp())){
                        ch = boot.bind(nettyHttpProperties.getBindIp(),
                                nettyHttpProperties.getPort()).sync().channel();
                    }else{
                        ch = boot.bind(nettyHttpProperties.getPort()).sync().channel();
                    }
                    ch.closeFuture().sync();
                } catch (InterruptedException e) {
                    log.error("启动NettyServer错误", e);
                } finally {
                    bossGroup.shutdownGracefully();
                    workerGroup.shutdownGracefully();
                }
            });
            thread.setName("http_Server");
        //    thread.setDaemon(true);
            thread.start();

            return "http start";
        }

        private ServerBootstrap options(ServerBootstrap boot) {
            /*if (HttpConfiguration.me().getSoBacklog() > 0) {
                boot.option(ChannelOption.SO_BACKLOG, HttpConfiguration.me().getSoBacklog());
            }*/
            return boot;
        }
    
}

启动配置:

---application.yml
spring.profiles.active: http

---application-http.yml
netty:
   http:
     enabled: true
     port: 1999
     bossThreads: 2
     workThreads: 4

测试: