基于Netty实现HttpClinet/HttpServer

清疚 2022-10-21 01:23 311阅读 0赞

基于Netty实现 HTTP Server 和 HTTP Client

【订阅专栏合集,作者所有付费文章都能看(持续更新)】

推荐【Kafka教程】https://bigbird.blog.csdn.net/article/details/108770504
推荐【rabbitmq教程】https://bigbird.blog.csdn.net/article/details/81436980
推荐【Flink教程】https://blog.csdn.net/hellozpc/article/details/109413465
推荐【JVM面试与调优教程】https://bigbird.blog.csdn.net/article/details/113888604
推荐【SpringBoot全套教程】https://blog.csdn.net/hellozpc/article/details/107095951
推荐【SpringCloud教程】https://blog.csdn.net/hellozpc/article/details/83692496
推荐【Mybatis教程】https://blog.csdn.net/hellozpc/article/details/80878563
推荐【SnowFlake教程】https://blog.csdn.net/hellozpc/article/details/108248227
推荐【并发限流教程】https://blog.csdn.net/hellozpc/article/details/107582771
推荐【Redis教程】https://bigbird.blog.csdn.net/article/details/81267030
推荐【Netty教程】https://blog.csdn.net/hellozpc/category_10945233.html

文章目录

  • 基于Netty实现 HTTP Server 和 HTTP Client
    • 简单HTTP Server/Client
      • 服务端
      • 客户端
    • 进阶版HTTP Server/Client
      • 服务端
      • 客户端
    • 通用HTTP Client
    • 通用HTTP Server
    • 小结

适合人群：对 Netty 感兴趣，需要进行实战训练的同学。实际开发中需要使用高性能的 HTTP 客户端工具请求服务端接口的研发人员。

GitChat订阅地址：《基于 Netty 实现 HTTP Server 和 HTTP Client》

HTTP服务器就是通常我们所说的 Web 服务器，比如 Tomcat 就是我们最常用 Web Server 之一。传统的web服务器有时也被称作web容器或者Servlet容器，因为其基本上都是基于Servlet规范的。通过之前对 Netty的学习，我们知道Netty支持各种各样的协议，当然也包括HTTP协议。因此使用 Netty 可以方便地开发一个和传统 Web 容器 Tomcat、Jetty 类似的 HTTP 服务器。同样的，也可以基于 Netty 开发一个通用的 HTTP Client，类似 Apache HTTP Client 或 Spring Resttemplate 的功能。本文就一步一步引导大家基于Netty实现一个HTTP Server 和 HTTP Client。关于Netty的基础知识，请读者先行补充，可以参见《循序渐进学Netty》。

简单HTTP Server/Client

我们首先从一个最简单的 HTTP 服务器开始，通过程序实例展示Netty程序如何配置启动，以及netty中的引导器(BootStrap)如何与核心组件产生联系。

开发步骤一般分为3步：

1）搭建 HTTP 服务器程序，配置相关参数并启动。

2）从浏览器或者其它终端发起 HTTP 请求。

3）客户端成功得到服务端的响应结果。

本文使用的netty版本如下：

  1. <dependency>
  2.     <groupId>io.netty</groupId>
  3.     <artifactId>netty-all</artifactId>
  4.     <version>4.1.32.Final</version>
  5. </dependency>

服务端

  1. public class HttpServer1 { 
  2.     public void start(int port) throws InterruptedException { 
  3.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  4.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  6.         ServerBootstrap b = new ServerBootstrap();
  7.         b.group(bossGroup, workerGroup)
  8.                 .channel(NioServerSocketChannel.class)
  9.                 .localAddress(new InetSocketAddress(port))
  10.                 .childHandler(new ChannelInitializer<SocketChannel>() { 
  11.                     @Override
  12.                     protected void initChannel(SocketChannel ch) throws Exception { 
  13.                         ch.pipeline().addLast("decoder", new HttpRequestDecoder()) // HTTP解码
  14.                                 .addLast("aggregator", new HttpObjectAggregator(2 * 1024 * 1024))//HTTP消息聚合
  15.                                 .addLast("encoder", new HttpResponseEncoder())// HTTP编码
  16.                                 .addLast("compressor", new HttpContentCompressor())//HttpContent压缩
  17.                                 .addLast("handler", new HttpServerHandler1());
  18.                     }
  19.                 })
  20.                 .childOption(ChannelOption.SO_KEEPALIVE, true)
  21.                 .childOption(ChannelOption.TCP_NODELAY, true);
  23.         b.bind().sync();//同步阻塞，避免程序退出
  24.         System.out.println("netty http server started on port(" + port + ")");
  25.     }
  27.     public static void main(String[] args) throws InterruptedException { 
  28.         new HttpServer1().start(8800);
  29.     }
  30. }

熟悉netty编程流程的同学都知道，一般业务开发都是编写一个个的ChannelHandler，并将其加入到Pipeline中处理数据。上述代码中使用Netty内置的针对HTTP的处理器：HttpResponseEncoder、HttpResponseDecoder。业务处理器如下，只是简单的将数据回显。

  1. public class HttpServerHandler1 extends SimpleChannelInboundHandler<FullHttpRequest> { 
  2.     @Override
  3.     protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception { 
  4.         String content = String.format("Received http request, uri: %s, method: %s, content: %s%n",
  5.                 msg.uri(), msg.method(), msg.content().toString(CharsetUtil.UTF_8));
  6.         //将服务器生成的内容回显到客户端
  7.         FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
  8.                 Unpooled.wrappedBuffer(content.getBytes()));
  10.         HttpHeaders heads = response.headers();
  11.         heads.add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
  12.         heads.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  14.         ctx.writeAndFlush(response);
  15.     }
  17.     @Override
  18.     public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  19.         System.out.println("channelActive 客户端地址:" + ctx.channel().remoteAddress());
  20.         super.channelActive(ctx);
  21.     }
  22. }

客户端

  1. public class HttpClient1 { 
  2.     public void connect(String host, int port) throws Exception { 
  3.         EventLoopGroup group = new NioEventLoopGroup();
  4.         try { 
  5.             Bootstrap b = new Bootstrap();
  6.             b.group(group);
  7.             b.channel(NioSocketChannel.class);
  8.             b.option(ChannelOption.SO_KEEPALIVE, true);
  9.             b.handler(new ChannelInitializer<SocketChannel>() { 
  10.                 @Override
  11.                 public void initChannel(SocketChannel ch) { 
  12.                     ch.pipeline().addLast(new HttpResponseDecoder());//HTTP编码
  13.                     ch.pipeline().addLast(new HttpRequestEncoder());//HTTP解码
  14.                     ch.pipeline().addLast(new HttpClientHandler1());//业务处理器
  15.                 }
  16.             });
  18.             //建立长连接
  19.             ChannelFuture f = b.connect(host, port).sync();
  20.             System.out.println("netty http client connected on host(" + host + ") port(" + port + ")");
  22.             //发送http请求
  23.             URI uri = new URI("http://127.0.0.1:8800");
  24.             String content = "hello netty http Server!";
  25.             DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET,
  26.                     uri.toASCIIString(), Unpooled.wrappedBuffer(content.getBytes(StandardCharsets.UTF_8)));
  27.             request.headers().set(HttpHeaderNames.HOST, host);
  28.             request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  29.             request.headers().set(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes());
  30.             f.channel().writeAndFlush(request);
  31.             f.channel().closeFuture().sync();
  32.         } finally { 
  33.             group.shutdownGracefully();
  34.         }
  35.     }
  37.     public static void main(String[] args) throws Exception { 
  38.         new HttpClient1().connect("127.0.0.1", 8800);
  39.     }
  40. }

客户端业务处理器

  1. public class HttpClientHandler1 extends ChannelInboundHandlerAdapter { 
  2.     @Override
  3.     public void channelRead(ChannelHandlerContext ctx, Object msg) { 
  4.         if (msg instanceof HttpContent) { 
  5.             HttpContent content = (HttpContent) msg;
  6.             ByteBuf buf = content.content();
  7.             System.out.println("收到服务端的消息:" + buf.toString(io.netty.util.CharsetUtil.UTF_8));
  8.             buf.release();
  9.         }
  10.     }
  11. }

进阶版HTTP Server/Client

上述案例以最简化的代码清晰地展示了基于Netty的http客户端、服务端通信流程。下面给出进阶版的代码案例。

服务端

  1. public class NettyHttpServer { 
  2.     String host = "*";
  3.     int port = 8800;
  5.     int backlog = 128;
  6.     int maxContentLength = 1024 * 1024;
  8.     EventLoopGroup bossGroup;
  9.     EventLoopGroup workerGroup;
  10.     ServerBootstrap serverBootstrap;
  12.     AtomicBoolean stopFlag = new AtomicBoolean();
  14.     public NettyHttpServer() { 
  15.     }
  17.     public NettyHttpServer(int port) { 
  18.         this.port = port;
  19.     }
  21.     public void init() { 
  22.         bossGroup = new NioEventLoopGroup();
  23.         workerGroup = new NioEventLoopGroup();
  24.         serverBootstrap = new ServerBootstrap();
  25.         serverBootstrap.group(bossGroup, workerGroup)
  26.                 .channel(NioServerSocketChannel.class)
  27.                 .childHandler(new ChannelInitializer<SocketChannel>() { 
  28.                     @Override
  29.                     protected void initChannel(SocketChannel ch) throws Exception { 
  30.                         ChannelPipeline pipeline = ch.pipeline();
  31.                         pipeline.addLast("codec", new HttpServerCodec());// HTTP编解码
  32.                         pipeline.addLast("decompressor", new HttpContentDecompressor());//HttpContent解压缩
  33.                         pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));//HTTP 消息聚合
  34.                         pipeline.addLast("compressor", new HttpContentCompressor());//HttpContent压缩
  35.                         pipeline.addLast("handler", new NettyHttpServerHandler());//自定义业务逻辑处理器
  36.                     }
  37.                 });
  38.         serverBootstrap.option(ChannelOption.SO_REUSEADDR, true);
  39.         serverBootstrap.option(ChannelOption.SO_BACKLOG, backlog);
  40.         serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
  41.         serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
  42.     }
  44.     public void start() { 
  45.         InetSocketAddress addr = null;
  46.         if (host == null || "*".equals(host)) { 
  47.             addr = new InetSocketAddress(port);
  48.         } else { 
  49.             addr = new InetSocketAddress(host, port);
  50.         }
  51.         try { 
  52.             serverBootstrap.bind(addr).sync();
  53.             System.out.println("netty http server started on host(" + addr.getHostName() + ") port(" + port + ")");
  54.         } catch (Exception e) { 
  55.             System.out.println("netty http server bind exception, port=" + port);
  56.             System.exit(-1);
  57.         }
  58.     }
  60.     public void close() { 
  61.         System.out.println("stopping netty server");
  62.         if (bossGroup != null) { 
  63.             bossGroup.shutdownGracefully();
  64.             bossGroup = null;
  65.         }
  67.         if (workerGroup != null) { 
  68.             workerGroup.shutdownGracefully();
  69.             workerGroup = null;
  70.         }
  71.         System.out.println("netty server stopped");
  72.     }
  74.     public void stop() { 
  75.         stopFlag.set(true);
  76.     }
  78.     public String getHost() { 
  79.         return host;
  80.     }
  82.     public void setHost(String host) { 
  83.         this.host = host;
  84.     }
  86.     public int getPort() { 
  87.         return port;
  88.     }
  90.     public void setPort(int port) { 
  91.         this.port = port;
  92.     }
  94.     public int getBacklog() { 
  95.         return backlog;
  96.     }
  98.     public void setBacklog(int backlog) { 
  99.         this.backlog = backlog;
  100.     }
  102.     public int getMaxContentLength() { 
  103.         return maxContentLength;
  104.     }
  106.     public void setMaxContentLength(int maxContentLength) { 
  107.         this.maxContentLength = maxContentLength;
  108.     }
  109. }

业务处理器

  1. public class NettyHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> { 
  2.     @Override
  3.     protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception { 
  4.         System.out.println("NettyHttpServerHandler.channelRead0");
  5.         String content = String.format("NettyHttpServerHandler Received http request, uri: %s, method: %s, content: %s%n", msg.uri(), msg.method(), msg.content().toString(CharsetUtil.UTF_8));
  6.         FullHttpResponse response = new DefaultFullHttpResponse(
  7.                 HttpVersion.HTTP_1_1,
  8.                 HttpResponseStatus.OK,
  9.                 Unpooled.wrappedBuffer(content.getBytes()));
  11.         ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
  12.     }
  14.     @Override
  15.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception { 
  16.         System.out.println("NettyHttpServerHandler.channelRegistered");
  17.         super.channelRegistered(ctx);
  18.     }
  20.     @Override
  21.     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { 
  22.         System.out.println("NettyHttpServerHandler.channelUnregistered");
  23.         super.channelUnregistered(ctx);
  24.     }
  26.     @Override
  27.     public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  28.         System.out.println("NettyHttpServerHandler.channelActive");
  29.         super.channelActive(ctx);
  30.     }
  32.     @Override
  33.     public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
  34.         System.out.println("NettyHttpServerHandler.channelInactive");
  35.         super.channelInactive(ctx);
  36.     }
  38.     @Override
  39.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
  40.         System.out.println("NettyHttpServerHandler.channelReadComplete");
  41.         super.channelReadComplete(ctx);
  42.     }
  44.     @Override
  45.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
  46.         System.out.println("NettyHttpServerHandler.userEventTriggered");
  47.         super.userEventTriggered(ctx, evt);
  48.     }
  50.     @Override
  51.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
  52.         System.out.println("NettyHttpServerHandler.exceptionCaught:" + cause);
  53.         super.exceptionCaught(ctx, cause);
  54.     }
  56.     @Override
  57.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception { 
  58.         System.out.println("NettyHttpServerHandler.handlerAdded");
  59.         super.handlerAdded(ctx);
  60.     }
  62.     @Override
  63.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { 
  64.         System.out.println("NettyHttpServerHandler.handlerRemoved");
  65.         super.handlerRemoved(ctx);
  66.     }
  67. }

启动类

  1. public class MainAppServer { 
  2.     public static void main(String[] args) throws InterruptedException { 
  3.         NettyHttpServer nettyHttpServer = new NettyHttpServer();
  4.         nettyHttpServer.init();
  5.         nettyHttpServer.start();
  6.         Thread.sleep(1200000);
  7.         nettyHttpServer.close();
  8.     }
  9. }

客户端

  1. public class NettyHttpClient { 
  2.     String host = "*";
  3.     int port = 8800;
  5.     int maxResponseContentLength = 1024 * 1024;
  7.     EventLoopGroup workerGroup;
  8.     Bootstrap bootstrap;
  10.     public void init() { 
  11.         workerGroup = new NioEventLoopGroup();
  12.         bootstrap = new Bootstrap();
  13.         bootstrap.group(workerGroup).channel(NioSocketChannel.class)
  14.                 .handler(new ChannelInitializer<SocketChannel>() { 
  15.                     @Override
  16.                     protected void initChannel(SocketChannel ch) throws Exception { 
  17.                         ChannelPipeline pipeline = ch.pipeline();
  18.                         pipeline.addLast("codec", new HttpClientCodec());
  19.                         pipeline.addLast("decompressor", new HttpContentDecompressor());
  20.                         pipeline.addLast("aggregator", new HttpObjectAggregator(maxResponseContentLength));
  21.                         pipeline.addLast("handler", new NettyHttpClientHandler());
  22.                     }
  23.                 });
  25.         bootstrap.option(ChannelOption.TCP_NODELAY, true);
  26.         bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
  27.     }
  29.     public ChannelFuture connect() { 
  30.         ChannelFuture channelFuture = null;
  31.         try { 
  32.             channelFuture = bootstrap.connect(host, port).sync();
  33.             System.out.println("netty http client connected on host(" + host + ") port(" + port + ")");
  34.         } catch (Exception e) { 
  35.             System.out.println("netty http client connect exception, host=" + host + ",port=" + port);
  36.             System.exit(-1);
  37.         }
  38.         return channelFuture;
  39.     }
  41.     public void close() { 
  42.         System.out.println("stopping netty client");
  43.         if (workerGroup != null) { 
  44.             workerGroup.shutdownGracefully();
  45.             workerGroup = null;
  46.         }
  48.         System.out.println("netty client stopped");
  49.     }
  51.     public String getHost() { 
  52.         return host;
  53.     }
  55.     public void setHost(String host) { 
  56.         this.host = host;
  57.     }
  59.     public int getPort() { 
  60.         return port;
  61.     }
  63.     public void setPort(int port) { 
  64.         this.port = port;
  65.     }
  66. }

业务处理器

  1. public class NettyHttpClientHandler extends ChannelInboundHandlerAdapter { 
  2.     @Override
  3.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
  4.         if (msg instanceof HttpContent) { 
  5.             HttpContent content = (HttpContent) msg;
  6.             ByteBuf buf = content.content();
  7.             System.out.println(buf.toString(CharsetUtil.UTF_8));
  8.             buf.release();
  9.         }
  10.     }
  11. }

启动类

  1. public class MainAppClient { 
  2.     public static void main(String[] args) throws InterruptedException, URISyntaxException { 
  3.         NettyHttpClient nettyHttpClient = new NettyHttpClient();
  4.         nettyHttpClient.setHost("127.0.0.1");
  5.         nettyHttpClient.setPort(8800);
  6.         nettyHttpClient.init();
  7.         ChannelFuture f = nettyHttpClient.connect();
  8.         if (f != null) { 
  9.             URI uri = new URI("http://127.0.0.1:8800");
  10.             String content = "hello netty http Server!";
  11.             DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
  12.                     HttpMethod.GET, uri.toASCIIString(), Unpooled.wrappedBuffer(content.getBytes(StandardCharsets.UTF_8)));
  13.             request.headers().set(HttpHeaderNames.HOST, "127.0.0.1");
  14.             request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  15.             request.headers().set(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes());
  16.             f.channel().write(request);//发送消息
  17.             f.channel().flush();
  18.         }
  20.         Thread.sleep(1200000);
  21.         nettyHttpClient.close();
  22.     }
  23. }

进阶版的代码显得更加规范化，也可以传入一些配置参数。读者可以比较一下相对初始版的变化，领悟每个版本的改进之处。

通用HTTP Client

基于 Netty 开发的一个通用的 HTTP Client，类似 Apache HTTP Client 或 Spring Resttemplate 的功能。可以在实际开发中使用，用作一个发送http请求的客户端工具。
完整代码github地址：https://github.com/bigbirditedu/nettylearn

  1. /** * 基于Netty实现的通用的HTTP客户端工具，可以在实际项目中使用 */@Sharablepublic class CommonHttpClient extends ChannelDuplexHandler implements HttpClient, InitClose {     static Logger log = LoggerFactory.getLogger(CommonHttpClient.class);    int maxResponseContentLength = 2 * 1024 * 1024;    int workerThreads = 1;    int minSizeToGzip = 2048;    int keepAliveSeconds = 60;    //保持的请求数 int keepAliveRequests = 100; //缓存的连接数 int keepAliveConnections = 1; //自定义线程池 NamedThreadFactory workThreadFactory = new NamedThreadFactory("netty_httpclient"); EventLoopGroup workerGroup; GZip gzip; SslContext sslCtx; Bootstrap bootstrap; Bootstrap sslBootstrap; //保存每一次的请求、响应信息 映射关系 ConcurrentHashMap<String, ReqResInfo> dataMap = new ConcurrentHashMap<>(); //保存每个目标地址的缓存连接数映射关系 ConcurrentHashMap<String, ArrayBlockingQueue<ChannelInfo>> channelMap = new ConcurrentHashMap<>(); public void init() { gzip = new GZip(); try { sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build(); } catch (Exception e) { log.error("ssl context init failed"); } workerGroup = new NioEventLoopGroup(workerThreads, workThreadFactory); bootstrap = new Bootstrap(); bootstrap.group(workerGroup).channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("codec", new HttpClientCodec()); pipeline.addLast("decompressor", new HttpContentDecompressor()); pipeline.addLast("aggregator", new HttpObjectAggregator(maxResponseContentLength)); pipeline.addLast("handler", CommonHttpClient.this); //CommonHttpClient本身就是一个ChannelHandler，同时支持出站、入站操作 } }); bootstrap.option(ChannelOption.TCP_NODELAY, true); bootstrap.option(ChannelOption.SO_KEEPALIVE, true); sslBootstrap = new Bootstrap(); sslBootstrap.group(workerGroup).channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc())); pipeline.addLast("codec", new HttpClientCodec()); pipeline.addLast("decompressor", new HttpContentDecompressor()); pipeline.addLast("aggregator", new HttpObjectAggregator(maxResponseContentLength)); pipeline.addLast("handler", CommonHttpClient.this); } }); sslBootstrap.option(ChannelOption.TCP_NODELAY, true); sslBootstrap.option(ChannelOption.SO_KEEPALIVE, true); log.info("netty http client started"); } public void close() { if (workerGroup == null) return; workerGroup.shutdownGracefully(); workerGroup = null; log.info("netty http client stopped"); } /** * 发起Http调用 * * @param req * @return */ public HttpClientRes call(HttpClientReq req) { HttpClientRes res = null; ConnectResult cr = null; String connId = null; try { cr = getConnection(req); if (cr.retCode != 0) return new HttpClientRes(cr.retCode); Channel channel = cr.channelInfo.channel; connId = getConnId(channel); ReqResInfo info = new ReqResInfo(req); dataMap.put(connId, info); channel.writeAndFlush(req); res = info.future.get(req.getTimeout(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { res = new HttpClientRes(RetCodes.HTTPCLIENT_INTERRUPTED); } catch (TimeoutException e) { res = new HttpClientRes(RetCodes.HTTPCLIENT_TIMEOUT); } catch (Exception e) { log.error("http client call exception", e); res = new HttpClientRes(RetCodes.HTTPCLIENT_CONNECTION_BROKEN); } if (connId != null) dataMap.remove(connId); if (cr != null) closeConnection(req, res, cr.channelInfo); return res; } String getKey(HttpClientReq req) { URL url = req.getUrlObj(); String schema = url.getProtocol(); String host = url.getHost(); int port = url.getPort(); if (port == -1) port = url.getDefaultPort(); return schema + "//" + host + ":" + port; } ConnectResult getConnection(HttpClientReq req) throws InterruptedException { if (!req.isKeepAlive()) { return getConnectionNoCache(req); } String key = getKey(req); ArrayBlockingQueue<ChannelInfo> queue = channelMap.get(key); if (queue == null) { return getConnectionNoCache(req); } ChannelInfo channelInfo = queue.poll(); if (channelInfo == null || !channelInfo.channel.isActive()) { return getConnectionNoCache(req); } if (keepAliveSeconds > 0) { if (System.currentTimeMillis() - channelInfo.createTime >= keepAliveSeconds * 1000) { channelInfo.channel.close(); return getConnectionNoCache(req); } } if (keepAliveRequests > 0) { if (channelInfo.count >= keepAliveRequests) { channelInfo.channel.close(); return getConnectionNoCache(req); } channelInfo.count++; } return new ConnectResult(channelInfo); } ConnectResult getConnectionNoCache(HttpClientReq req) throws InterruptedException { URL url = req.getUrlObj(); if (url == null) return new ConnectResult(RetCodes.HTTPCLIENT_URL_PARSE_ERROR); String schema = url.getProtocol(); String host = url.getHost(); int port = url.getPort(); if (port == -1) port = url.getDefaultPort(); ChannelFuture future = schema.equals("http") ? bootstrap.connect(host, port) : sslBootstrap.connect(host, port); Channel channel = future.channel(); boolean done = future.await(req.getTimeout(), TimeUnit.MILLISECONDS); if (!done) { channel.close(); return new ConnectResult(RetCodes.HTTPCLIENT_TIMEOUT); } if (channel.isActive()) return new ConnectResult(new ChannelInfo(channel)); else return new ConnectResult(RetCodes.HTTPCLIENT_NO_CONNECT); } void closeConnection(HttpClientReq req, HttpClientRes res, ChannelInfo channelInfo) { if (!req.isKeepAlive() || res.getRetCode() != 0) { channelInfo.channel.close(); return; } String key = getKey(req); ArrayBlockingQueue<ChannelInfo> queue = channelMap.get(key); if (queue == null) { queue = new ArrayBlockingQueue<ChannelInfo>(keepAliveConnections); ArrayBlockingQueue<ChannelInfo> old = channelMap.putIfAbsent(key, queue); if (old != null) queue = old; } boolean ok = queue.offer(channelInfo); if (!ok) { channelInfo.channel.close(); return; } } @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { if (msg instanceof HttpClientReq) { HttpClientReq data = (HttpClientReq) msg; DefaultFullHttpRequest req = convertReq(data, ctx); ctx.writeAndFlush(req, promise); } else { super.write(ctx, msg, promise); } } DefaultFullHttpRequest convertReq(HttpClientReq data, ChannelHandlerContext ctx) { URL url = data.getUrlObj(); DefaultFullHttpRequest req = null; if (data.getContent() != null && !data.getContent().isEmpty()) { ByteBuf bb = null; boolean allowGzip = data.isGzip() && data.getContent().length() >= minSizeToGzip; if (!allowGzip) { bb = ByteBufUtil.writeUtf8(ctx.alloc(), data.getContent()); } else { bb = ctx.alloc().buffer(); try { String charset = parseCharSet(data.getContentType()); byte[] bytes = data.getContent().getBytes(charset); gzip.zip(bytes, bb); } catch (Exception e) { ReferenceCountUtil.release(bb); throw new RuntimeException("encode request exception", e); } } req = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.valueOf(data.getMethod()), data.getPathQuery(), bb); req.headers().set(HttpHeaderNames.CONTENT_TYPE, data.getContentType()); req.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, bb.readableBytes()); if (allowGzip) { req.headers().set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP); } } else { req = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.valueOf(data.getMethod()), data.getPathQuery()); } req.headers().set(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP); if (data.getHeaders() != null) { for (Map.Entry<String, String> entry : data.getHeaders().entrySet()) { req.headers().set(entry.getKey(), entry.getValue()); } } String host = url.getHost(); int port = url.getPort(); if (port != -1) host += ":" + port; req.headers().set(HttpHeaderNames.HOST, host); HttpUtil.setKeepAlive(req, data.isKeepAlive()); req.headers().set(HttpHeaderNames.USER_AGENT, "common httpclient 1.0"); return req; } @Override public void channelRead(final ChannelHandlerContext ctx, final Object msg) { String connId = getConnId(ctx.channel()); ReqResInfo info = dataMap.get(connId); try { FullHttpResponse httpRes = (FullHttpResponse) msg; if (!httpRes.decoderResult().isSuccess()) { if (info != null) info.setRes(new HttpClientRes(RetCodes.HTTPCLIENT_RES_PARSE_ERROR)); return; } HttpClientRes res = convertRes(httpRes); if (info != null) info.setRes(res); } finally { ReferenceCountUtil.release(msg); } } HttpClientRes convertRes(FullHttpResponse data) { HttpClientRes res = new HttpClientRes(0); res.setHttpCode(data.status().code()); res.setHeaders(data.headers()); String contentType = parseContentType(data.headers().get(HttpHeaderNames.CONTENT_TYPE)); if (contentType != null) res.setContentType(contentType); ByteBuf bb = data.content(); if (bb != null) { if (contentType != null && contentType.equals("application/octet-stream")) { byte[] buff = new byte[bb.readableBytes()]; bb.readBytes(buff); res.setRawContent(buff); } else { String content = bb.toString(Charset.forName("utf-8")); res.setContent(content); } } // System.out.println(HttpUtil.isKeepAlive(data)); return res; } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { String connId = getConnId(ctx.channel()); log.debug("http connection started, connId={}", connId); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { String connId = getConnId(ctx.channel()); log.debug("http connection ended, connId={}", connId); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { String connId = getConnId(ctx.channel()); log.error("http connection exception, connId=" + connId + ",msg=" + cause.toString(), cause); ctx.close(); } String getConnId(Channel ch) { SocketAddress addr = ch.remoteAddress(); return parseIpPort(addr.toString()) + ":" + ch.id().asShortText(); } String parseIpPort(String s) { int p = s.indexOf("/"); if (p >= 0) return s.substring(p + 1); else return s; } String parseContentType(String contentType) { if (contentType == null) return null; int p = contentType.indexOf(";"); if (p >= 0) return contentType.substring(0, p).trim(); return contentType; } String parseCharSet(String contentType) { if (contentType == null) return null; int p = contentType.indexOf(";"); if (p >= 0) return contentType.substring(p + 1).trim(); return "utf-8"; } public int getWorkerThreads() { return workerThreads; } public void setWorkerThreads(int workerThreads) { this.workerThreads = workerThreads; } public int getKeepAliveSeconds() { return keepAliveSeconds; } public void setKeepAliveSeconds(int keepAliveSeconds) { this.keepAliveSeconds = keepAliveSeconds; } public int getKeepAliveConnections() { return keepAliveConnections; } public void setKeepAliveConnections(int keepAliveConnections) { this.keepAliveConnections = keepAliveConnections; } public int getKeepAliveRequests() { return keepAliveRequests; } public void setKeepAliveRequests(int keepAliveRequests) { this.keepAliveRequests = keepAliveRequests; } public int getMaxResponseContentLength() { return maxResponseContentLength; } public void setMaxResponseContentLength(int maxResponseContentLength) { this.maxResponseContentLength = maxResponseContentLength; } public int getMinSizeToGzip() { return minSizeToGzip; } public void setMinSizeToGzip(int minSizeToGzip) { this.minSizeToGzip = minSizeToGzip; } static class ReqResInfo { HttpClientReq req; HttpClientRes res; CompletableFuture<HttpClientRes> future; ReqResInfo(HttpClientReq req) { this.req = req; this.future = new CompletableFuture<HttpClientRes>(); } void setRes(HttpClientRes res) { this.res = res; future.complete(res); } } static class ChannelInfo { Channel channel = null; long createTime = System.currentTimeMillis(); int count = 1; ChannelInfo(Channel channel) { this.channel = channel; } } static class ConnectResult { int retCode = 0; ChannelInfo channelInfo = null; ConnectResult(int retCode) { this.retCode = retCode; } ConnectResult(ChannelInfo channelInfo) { this.channelInfo = channelInfo; } }}

测试类

  1. public class TestHttpClient {     public static void main(String[] args) {         CommonHttpClient commonHttpClient = new CommonHttpClient();        commonHttpClient.init();        HttpClientReq httpClientReq = new HttpClientReq("POST", "http://127.0.0.1:8080/userService/queryUserInfo");        httpClientReq.setContent("{\"uid\":\"121\"}");        httpClientReq.setContentType("application/json");        HttpClientRes res = commonHttpClient.call(httpClientReq);        System.out.println(res.getContent());        commonHttpClient.close();    }}

其它代码由于篇幅限制，不直接贴出来了。可以参考：https://github.com/bigbirditedu/nettylearn

我们另起一个SpringBoot工程，充当服务方。使用上述httpClient调用服务方的接口。

先运行下面的SpringBoot工程，再运行上述HttpClient测试类。

UserController

  1. @RestController@RequestMapping("userService")public class UserController {     @RequestMapping("/queryUserInfo")    public QueryUserInfoRes queryUserInfo(@RequestBody QueryUserInfoReq req) {         String uid = req.getUid();        //模拟查询数据库返回数据 User user = new User(); user.setUid(uid); user.setAge(22); user.setAddress("南京市栖霞区"); user.setName("吊哥"); user.setSex(1); QueryUserInfoRes res = new QueryUserInfoRes(); res.setUser(user); res.setRetCode(0); res.setRetMsg("成功"); return res; }}

QueryUserInfoReq

  1. public class QueryUserInfoReq {     private String uid;    public String getUid() {         return uid;    }    public void setUid(String uid) {         this.uid = uid;    }}

QueryUserInfoRes

  1. public class QueryUserInfoRes {     private User user;    private String retMsg;    private int retCode;    public User getUser() {         return user;    }    public void setUser(User user) {         this.user = user;    }    public String getRetMsg() {         return retMsg;    }    public void setRetMsg(String retMsg) {         this.retMsg = retMsg;    }    public int getRetCode() {         return retCode;    }    public void setRetCode(int retCode) {         this.retCode = retCode;    }}
  3. public class User {     private String name;    private String uid;    private int age;    private int sex;    private String address;    //省略getter/setter方法}

httpclient调用输出：

{"user":{"name":"吊哥","uid":"121","age":22,"sex":1,"address":"南京市栖霞区"},"retMsg":"成功","retCode":0}

通用HTTP Server

由于Netty 内置了支持Http协议的处理器(各种编解码器)。因此基于Netty 可以方便地开发一个类似传统 Web服务器Tomcat、Jetty 的HTTP 服务器。本文给出一个样例代码，有更好的实现欢迎留言。完整代码请参考：https://github.com/bigbirditedu/nettylearn

  1. @ChannelHandler.Sharablepublic class MyNettyHttpServer extends ChannelDuplexHandler implements InitClose, StartStop {     static Logger log = LoggerFactory.getLogger(MyNettyHttpServer.class);    int port = 8888;    String host = "*";    int idleSeconds = 30;    int maxConnections = 100000;    int workerThreads = 0;    int backlog = 128;    boolean nativeNetty = false;    int maxInitialLineLength = 4096;    int maxHeaderSize = 8192;    int maxChunkSize = 8192;    int maxContentLength = 2 * 1024 * 1024;    NamedThreadFactory bossThreadFactory = new NamedThreadFactory("netty_webserver_boss");    NamedThreadFactory workThreadFactory = new NamedThreadFactory("netty_webserver_worker");    EventLoopGroup bossGroup;    EventLoopGroup workerGroup;    Channel serverChannel;    ConcurrentHashMap<String, Channel> conns = new ConcurrentHashMap<String, Channel>();    AtomicBoolean stopFlag = new AtomicBoolean();    ServerBootstrap serverBootstrap;    public MyNettyHttpServer() {     }    public MyNettyHttpServer(int port) {         this.port = port;    }    @Override    public void init() {         int processors = Runtime.getRuntime().availableProcessors();        if (workerThreads == 0) {             workerThreads = processors;        }        String osName = System.getProperty("os.name");        if (nativeNetty && osName != null && osName.toLowerCase().indexOf("linux") >= 0) {             nativeNetty = true;        } else {             nativeNetty = false;        }        if (nativeNetty) {             bossGroup = new EpollEventLoopGroup(1, bossThreadFactory);            workerGroup = new EpollEventLoopGroup(workerThreads, workThreadFactory);        } else {             bossGroup = new NioEventLoopGroup(1, bossThreadFactory);            workerGroup = new NioEventLoopGroup(workerThreads, workThreadFactory);        }        serverBootstrap = new ServerBootstrap();        serverBootstrap.group(bossGroup, workerGroup).channel(nativeNetty ? EpollServerSocketChannel.class : NioServerSocketChannel.class)                .childHandler(new ChannelInitializer<SocketChannel>() {                     @Override                    protected void initChannel(SocketChannel ch) throws Exception {                         ChannelPipeline pipeline = ch.pipeline();                        pipeline.addLast("timeout", new IdleStateHandler(0, 0, idleSeconds));                        pipeline.addLast("codec", new HttpServerCodec(maxInitialLineLength, maxHeaderSize, maxChunkSize));                        pipeline.addLast("decompressor", new HttpContentDecompressor());                        pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));                        pipeline.addLast("compressor", new HttpContentCompressor());                        pipeline.addLast("handler", MyNettyHttpServer.this);                    }                });        serverBootstrap.option(ChannelOption.SO_REUSEADDR, true);        serverBootstrap.option(ChannelOption.SO_BACKLOG, backlog);        serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);    }    @Override    public void start() {         InetSocketAddress addr;        if (host == null || "*".equals(host)) {             addr = new InetSocketAddress(port); // "0.0.0.0" } else { addr = new InetSocketAddress(host, port); } try { serverChannel = serverBootstrap.bind(addr).syncUninterruptibly().channel(); log.info("netty http server started on host(" + addr.getHostString() + ") port(" + port + ")"); } catch (Exception e) { log.error("netty http server bind exception, port=" + port); System.exit(-1); } } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { if (stopFlag.get()) { ctx.close(); return; } String connId = getConnId(ctx); if (conns.size() >= maxConnections) { ctx.close(); log.error("connection started, connId={}, but not allowed,server max connections exceeded.", connId); return; } log.info("connection started, connId={}", connId); conns.put(connId, ctx.channel()); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { String connId = getConnId(ctx); conns.remove(connId); log.info("connection ended, connId={}", connId); } @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleStateEvent e = (IdleStateEvent) evt; if (e.state() == IdleState.ALL_IDLE) { String connId = getConnId(ctx); ctx.close(); log.error("connection timeout, ctx closed,connId={}", connId); } } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { String connId = getConnId(ctx); log.error("connection exception, connId=" + connId + ",msg=" + cause.toString(), cause); ctx.close(); } @Override public void channelRead(final ChannelHandlerContext ctx, final Object msg) { try { //由于使用了HttpObjectAggregator聚合处理器,这里的msg类型必然是FullHttpRequest FullHttpRequest fullHttpRequest = (FullHttpRequest) msg; if (!fullHttpRequest.decoderResult().isSuccess()) { sendError(ctx, HttpResponseStatus.BAD_REQUEST, RetCodes.DECODE_REQ_ERROR); return; } if (fullHttpRequest.method() != HttpMethod.GET && fullHttpRequest.method() != HttpMethod.POST && fullHttpRequest.method() != HttpMethod.PUT && fullHttpRequest.method() != HttpMethod.DELETE && fullHttpRequest.method() != HttpMethod.HEAD && fullHttpRequest.method() != HttpMethod.OPTIONS) { sendError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED, RetCodes.HTTP_METHOD_NOT_ALLOWED); return; } if (fullHttpRequest.method() == HttpMethod.OPTIONS) { String origin = fullHttpRequest.headers().get(HttpHeaderNames.ORIGIN); String requestMethod = fullHttpRequest.headers().get(HttpHeaderNames.ACCESS_CONTROL_REQUEST_METHOD); if (origin == null || requestMethod == null) { sendError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED, RetCodes.HTTP_METHOD_NOT_ALLOWED); return; } } //为了方便地处理http请求响应，封装了WebReq、WebRes对象 DefaultWebReq req = convertReq(fullHttpRequest); DefaultWebRes res = process(req); ctx.channel().writeAndFlush(res); } finally { //如果不在pipeline中传递ChannelHandler了，则此处手动释放 ReferenceCountUtil.release(msg); } } /** * 处理Http请求，通常是调用服务端某个业务接口，并返回响应 * 这里只是简单演示使用反射方式调用服务端接口方法，并返回json数据，未做复杂的接口路由 * * @param req * @return */ private DefaultWebRes process(DefaultWebReq req) { String path = req.getPath(); int i = path.lastIndexOf("/"); String methodName = path.substring(i + 1); String reqContent = req.getContent(); DefaultWebRes webRes = null; try { webRes = new DefaultWebRes(req, 200); webRes.setContentType("application/json"); Method method = OrderService.class.getDeclaredMethod(methodName, QueryOrderInfoReq.class); method.setAccessible(true); Class<?>[] parameterTypes = method.getParameterTypes(); Class<?> parameterType = parameterTypes[0]; Object param = JSON.parseObject(reqContent, parameterType); Object resContent = method.invoke(new OrderService(), param); webRes.setContent(JSON.toJSONString(resContent)); } catch (NoSuchMethodException noSuchMethodException) { String s = String.format(WebConstants.ContentFormat, -627, RetCodes.retCodeText(-627)); webRes.setContent(s); } catch (Exception e) { log.error("http call exception", e); webRes = new DefaultWebRes(req, 500); } return webRes; } @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { //这里可以自定义数据处理逻辑，如果已经是FullHttpMessage类型，比如sendError()，则直接透传到pipeline中的出站处理器中 if (msg instanceof DefaultWebRes) { DefaultWebRes data = (DefaultWebRes) msg; writeHttpResponse(ctx, promise, data); } else { super.write(ctx, msg, promise); } } private void writeHttpResponse(ChannelHandlerContext ctx, ChannelPromise promise, DefaultWebRes data) { DefaultFullHttpResponse res = null; if (data.getContent() != null && !data.getContent().isEmpty()) { int size = ByteBufUtil.utf8MaxBytes(data.getContent()); ByteBuf bb = ctx.alloc().buffer(size); bb.writeCharSequence(data.getContent(), Charset.forName(data.getCharSet())); int len = bb.readableBytes(); if (data.isHeadMethod()) { res = new DefaultFullHttpResponse(data.getVersion(), HttpResponseStatus.valueOf(data.getHttpCode())); ReferenceCountUtil.release(bb); } else { res = new DefaultFullHttpResponse(data.getVersion(), HttpResponseStatus.valueOf(data.getHttpCode()), bb); } res.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, len); } else { res = new DefaultFullHttpResponse(data.getVersion(), HttpResponseStatus.valueOf(data.getHttpCode())); res.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, 0); // 302 不返回长度会导致客户端卡住 } res.headers().set(HttpHeaderNames.SERVER, WebConstants.Server); if (stopFlag.get()) res.headers().set(WebConstants.ShutdownFlag, "1"); if (data.isKeepAlive()) { res.headers().set(HttpHeaderNames.CONNECTION, "keep-alive"); } if (data.getHeaders() != null) { for (String key : data.getHeaders().names()) { res.headers().set(key, data.getHeaders().get(key)); } } if (data.getCookies() != null) { for (io.netty.handler.codec.http.cookie.Cookie c : data.getCookies()) { String s = ServerCookieEncoder.STRICT.encode(c); res.headers().add(HttpHeaderNames.SET_COOKIE, s); } } ChannelFuture future = ctx.writeAndFlush(res, promise); if (!data.isKeepAlive()) { //如果没有KeepAlive则在数据传输完成后关闭通道 future.addListener(ChannelFutureListener.CLOSE); } } DefaultWebReq convertReq(FullHttpRequest fullHttpRequest) { DefaultWebReq req = new DefaultWebReq(); req.setVersion(fullHttpRequest.protocolVersion()); req.setMethod(fullHttpRequest.method()); req.setKeepAlive(HttpUtil.isKeepAlive(fullHttpRequest)); String uri = fullHttpRequest.uri(); int p1 = findPathEndIndex(uri); String path = p1 >= 0 ? uri.substring(0, p1) : uri; path = WebUtils.decodeUrl(path); int p2 = uri.indexOf('?'); String queryString = p2 >= 0 ? uri.substring(p2 + 1) : ""; req.setPath(path); req.setQueryString(queryString); req.setHeaders(fullHttpRequest.headers()); ByteBuf bb = fullHttpRequest.content(); if (bb != null) { String cs = req.getCharSet(); String content = bb.toString(Charset.forName(cs)); req.setContent(content); } return req; } @Override public void stop() { stopFlag.set(true); } @Override public void close() { if (workerGroup != null) { log.info("stopping netty http server"); bossGroup.shutdownGracefully(); bossGroup = null; ChannelGroup allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); allChannels.add(serverChannel); for (Channel ch : conns.values()) { allChannels.add(ch); } ChannelGroupFuture future = allChannels.close(); future.awaitUninterruptibly(); workerGroup.shutdownGracefully(); workerGroup = null; log.info("netty http server stopped"); } } Channel getChannel(String connId) { return conns.get(connId); } String getConnId(ChannelHandlerContext ctx) { Channel ch = ctx.channel(); return parseIpPort(ch.remoteAddress().toString()) + ":" + ch.id().asShortText(); } public String getAddr(String connId) { int p = connId.lastIndexOf(":"); return connId.substring(0, p); } String parseIp(String connId) { int p = connId.lastIndexOf(":"); if (p < 0) return connId; int p2 = connId.lastIndexOf(":", p - 1); if (p2 < 0) return connId; return connId.substring(0, p2); } String parseIpPort(String s) { int p = s.indexOf("/"); if (p >= 0) return s.substring(p + 1); else return s; } int findPathEndIndex(String uri) { int len = uri.length(); for (int i = 0; i < len; i++) { char c = uri.charAt(i); if (c == '?' || c == '#') { return i; } } return -1; } void sendError(ChannelHandlerContext ctx, HttpResponseStatus status, int retCode) { ByteBuf bb = ctx.alloc().buffer(32); String s = String.format(WebConstants.ContentFormat, retCode, RetCodes.retCodeText(retCode)); bb.writeCharSequence(s, CharsetUtil.UTF_8); int len = bb.readableBytes(); DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, bb); response.headers().set(HttpHeaderNames.CONTENT_TYPE, WebConstants.MIMETYPE_JSON); response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, len); ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE); }}

业务接口

这里服务器端只是简单提供了一个queryOrderInfo的业务接口。

服务器启动后，客户端访问http://127.0.0.1:8899/orderService/queryOrderInfo即可测试。

  1. public class OrderService {     public QueryOrderInfoRes queryOrderInfo(QueryOrderInfoReq req) {         QueryOrderInfoRes res = new QueryOrderInfoRes();        Order order = new Order();        order.setOrderId(req.getOrderId());        order.setOrderAmount(19999);        order.setProductId(UUID.randomUUID().toString());        order.setUserId(UUID.randomUUID().toString().replace("-", "").toUpperCase());        res.setOrder(order);        res.setRetMsg("success");        return res;    }}

启动测试

  1. public class MyNettyHttpServerTest {     public static void main(String[] args) throws InterruptedException {         MyNettyHttpServer nettyHttpServer = new MyNettyHttpServer(8899);        nettyHttpServer.init();        nettyHttpServer.start();        Thread.sleep(5 * 60 * 1000);        nettyHttpServer.stop();        nettyHttpServer.close();    }}

无论使用postman还是我们自己基于Netty编写的httpclient都可以访问http://127.0.0.1:8899/orderService/queryOrderInfo请求接口。服务端/客户端数据交互格式使用post/json。

小结

要基于Netty完整地实现一个高性能、功能完备、健壮易用的 HTTP 服务器非常复杂，相当于要实现Tomcat以及SpringMVC的部分功能。本文仅为了方便理解 Netty网络应用开发的基本过程，所以在案例中只是实现最基本的http请求-响应的流程。如果读者发现有更好的案例或者学习资源也欢迎留言哦。本文代码地址：https://github.com/bigbirditedu/nettylearn

持续关注【程猿薇茑】，后续获取更多实战教程。

发表评论

表情：
评论列表 （有 0 条评论，311人围观）

还没有评论，来说两句吧...

相关阅读

    相关 基于Netty实现简单的RPC框架

            Dubbo采用Netty作为基础通信组件，模仿Dubbo实现简单版的RPC框架。服务消费者和服务提供者约定接口和协议，用于远程调用和TCP请求验证。服务提供者作

    女爷i女爷i/ 0 赞/ 253 阅读