netty如何实现优化百万并发http请求-蒲公英云

问题引入：如果项目存在数量大的并发，如何优化项目性能

首先判断项目性能瓶颈是什么， CPU,还是I/O请求
大部分是网络I/o瓶颈，主要是因为网络请求是阻塞，导致太慢

Netty 是一个异步事件驱动的网络应用程序框架，用于快速开发可维护的高性能协议服务器和客户端。
JDK 原生 NIO 程序的问题
JDK 原生也有一套网络应用程序 API，但是存在一系列问题，主要如下：
NIO 的类库和 API 繁杂，使用麻烦。你需要熟练掌握 Selector、ServerSocketChannel、SocketChannel、ByteBuffer 等。
需要具备其他的额外技能做铺垫。例如熟悉 Java 多线程编程，因为 NIO 编程涉及到 Reactor 模式，你必须对多线程和网路编程非常熟悉，才能编写出高质量的 NIO 程序。
可靠性能力补齐，开发工作量和难度都非常大。例如客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常码流的处理等等。
NIO 编程的特点是功能开发相对容易，但是可靠性能力补齐工作量和难度都非常大。
JDK NIO 的 Bug。例如臭名昭著的 Epoll Bug，它会导致 Selector 空轮询，最终导致 CPU 100%。
官方声称在 JDK 1.6 版本的 update 18 修复了该问题，但是直到 JDK 1.7 版本该问题仍旧存在，只不过该 Bug 发生概率降低了一些而已，它并没有被根本解决。
Netty 的特点
Netty 对 JDK 自带的 NIO 的 API 进行封装，解决上述问题，主要特点有：
设计优雅，适用于各种传输类型的统一 API 阻塞和非阻塞 Socket；基于灵活且可扩展的事件模型，可以清晰地分离关注点；高度可定制的线程模型 - 单线程，一个或多个线程池；真正的无连接数据报套接字支持（自 3.1 起）。
使用方便，详细记录的 Javadoc，用户指南和示例；没有其他依赖项，JDK 5（Netty 3.x）或 6（Netty 4.x）就足够了。
高性能，吞吐量更高，延迟更低；减少资源消耗；最小化不必要的内存复制。
安全，完整的 SSL/TLS 和 StartTLS 支持。
社区活跃，不断更新，社区活跃，版本迭代周期短，发现的 Bug 可以被及时修复，同时，更多的新功能会被加入。

Netty 常见使用场景

Netty 常见的使用场景如下：
互联网行业。在分布式系统中，各个节点之间需要远程服务调用，高性能的 RPC 框架必不可少，Netty 作为异步高性能的通信框架，往往作为基础通信组件被这些 RPC 框架使用。
典型的应用有：阿里分布式服务框架 Dubbo 的 RPC 框架使用 Dubbo 协议进行节点间通信，Dubbo 协议默认使用 Netty 作为基础通信组件，用于实现各进程节点之间的内部通信。
游戏行业。无论是手游服务端还是大型的网络游戏，Java 语言得到了越来越广泛的应用。Netty 作为高性能的基础通信组件，它本身提供了 TCP/UDP 和 HTTP 协议栈。
非常方便定制和开发私有协议栈，账号登录服务器，地图服务器之间可以方便的通过 Netty 进行高性能的通信。
大数据领域。经典的 Hadoop 的高性能通信和序列化组件 Avro 的 RPC 框架，默认采用 Netty 进行跨界点通信，它的 Netty Service 基于 Netty 框架二次封装实现。
有兴趣的读者可以了解一下目前有哪些开源项目使用了 Netty：Related Projects。
Netty 高性能设计
Netty 作为异步事件驱动的网络，高性能之处主要来自于其 I/O 模型和线程处理模型，前者决定如何收发数据，后者决定如何处理数据。

netty实现http客户端解决网络I/O瓶颈

@Component
@Order(2)
public class HttpClient { 
    public static void start() { 
        EventLoopGroup group = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
        Bootstrap bootstrap = new Bootstrap();
        try { 
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<Channel>() { 
                        @Override
                        protected void initChannel(Channel channel)
                                throws Exception { 
                            /*//包含编码器和解码器 channel.pipeline().addLast(new HttpClientCodec()); //聚合 channel.pipeline().addLast(new HttpObjectAggregator(1024 * 10 * 1024)); //解压 channel.pipeline().addLast(new HttpContentDecompressor());*/
                            //请求编码
                            channel.pipeline().addLast(new HttpRequestEncoder());
                            //响应解码
                            channel.pipeline().addLast(new HttpResponseDecoder());
                            channel.pipeline().addLast(new HttpClientHandler());
                        }
                    });
            SimpleClientHttpRequestFactory httpRequestFactory = new SimpleClientHttpRequestFactory();
            SocketAddress address = new InetSocketAddress("net-proxy-pub-test.sftcwl.com", 1080);
            SocketAddress address1 = new InetSocketAddress("gis.sf-express.com", 8080);
            Proxy proxy = new Proxy(Proxy.Type.HTTP, address);
            httpRequestFactory.setProxy(proxy);
            ChannelFuture future = bootstrap.connect("localhost",8162).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) { 
            e.printStackTrace();
        } finally { 
            group.shutdownGracefully();
        }
    }
    public static void main(String[] args) { 
        start();
    }
public class HttpClientHandler extends ChannelInboundHandlerAdapter { 
    private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
    private static final String CLOSE_FREE_CHANNEL = "closeFreeChannel";
    /** * 关闭channel的原因，为closeFreeChannel时表示原因为关闭空闲socket连接 */
    public static final String CLOSE_CHANNEL_CAUSE = "closeChannelCause";
    private static Set<String> channelIDSet = new CopyOnWriteArraySet<>();
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
         URI uri = new URI("/abc/project/list");
        Map<String, Object> param = new HashMap<>(2);
        param.put("address", "雁塔区");
        param.put("ak", "58008e825a944a45909e056be12f511f");
        String prefix = "?";
        StringBuilder urlBuilder = new StringBuilder();
        for (String key : param.keySet()) { 
            urlBuilder.append(prefix).append(key).append("=").append(param.get(key));
            prefix = "&";
        }
        LOGGER.info(urlBuilder.toString());
        //配置HttpRequest的请求数据和一些配置信息
        FullHttpRequest request = new DefaultFullHttpRequest(
                HTTP_1_1,
                HttpMethod.GET,
                uri.toASCIIString()
                );
        LOGGER.info(request.toString());
        request.headers()
                .set(HttpHeaderNames.HOST, "127.0.0.1")
                .set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
                //开启长连接
                .set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
                //设置传递请求内容的长度
                .set(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes());
        ctx.writeAndFlush(request);
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception { 
        System.out.println("msg -> " + msg);
        if (msg instanceof HttpContent) { 
            HttpContent content = (HttpContent) msg;
            ByteBuf buf = content.content();
            String result = buf.toString(CharsetUtil.UTF_8);
            System.out.println("response -> " + result);
            LOGGER.info(result);
        }
    }
    /** * 通道失活事件，用于断线重连 * * @param ctx */
    @Override
    public void channelInactive(final ChannelHandlerContext ctx) { 
        LOGGER.info("Channel为 {} 断开，Disconnected Server：{}", ctx.channel().id(), ctx.channel().remoteAddress());
        EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.schedule(new Runnable() { 
            @Override
            public void run() { 
                SocketChannel channel = (SocketChannel) ctx.channel();
                int port = channel.remoteAddress().getPort();
                //获取channel中存储的关闭连接的原因，如果是关闭空闲连接，则不再重连
                String closeChannelCause = channel.attr(AttributeKey.<String>valueOf(CLOSE_CHANNEL_CAUSE)).get();
                if (CLOSE_FREE_CHANNEL.equals(closeChannelCause)) { 
                    LOGGER.info("检测到Channel为 {} 的空闲连接，已经断开.", channel.id());
                    return;
                }
                LOGGER.info("检测到与服务端断开连接，NettyClient开始重连{}:{}", channel.remoteAddress().getHostName(), port);
            }
        }, 3, TimeUnit.SECONDS);
        ctx.fireChannelInactive();
    }
    /** * 监听idle事件，用于监听心跳 * * @param ctx * @param evt * @throws Exception */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
        Channel channel = ctx.channel();
        if (evt instanceof IdleStateEvent) { 
            if (channel.isActive()) { 
                LOGGER.info("[客户端心跳监测] 通道编号: {} ,客户端信息: {}", channel.id(), channel.localAddress());
                channelIDSet.add(channel.id().toString());
                LOGGER.info("[客户端心跳监测] 关闭当前空闲，通道编号: {} ,客户端信息: {}", channel.id(), channel.localAddress());
                return;
            }
        }
    }
    /** * 异常处理 * * @param ctx * @param cause * @throws Exception */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
        LOGGER.error("Socket通信过程中出现异常,异常原因为：", cause);
        super.exceptionCaught(ctx, cause);
    }
}