Netty架构

Netty官网 Netty 是一个开源的网络应用框架，用于快速开发高性能的网络和分布式应用程序。它具有许多核心概念，这些概念帮助我们构建可伸缩、高性能和可维护的网络应用。以下是 Netty 的一些核心概念：

Channel（通道）： Channel 是数据通信的抽象，它可以表示底层的网络连接，如套接字。Netty 提供了多种类型的通道，用于不同的传输协议（例如，NIO、OIO、本地传输）。

EventLoop（事件循环）： EventLoop 是一个用于处理事件的循环，它是 Netty 中的核心组件。每个 Channel 都关联了一个 EventLoop，它负责处理 Channel 的所有事件，如接收数据、处理数据、发送数据等。

ChannelHandler（通道处理器）： ChannelHandler 是处理 Channel 事件的组件，可以用于实现协议编解码、业务逻辑处理等。ChannelHandler 链是在通道上执行的操作的序列。

Bootstrap（引导器）： Bootstrap 是一个用于启动和配置网络应用的工具，它通常用于创建和连接 Channel。它帮助设置 EventLoopGroup、Channel 类型、ChannelHandler 等配置。

ChannelPipeline（通道管道）： ChannelPipeline 是一个用于维护和处理 ChannelHandler 链的数据结构。它用于指定事件的处理顺序和流向，以便在 Channel 上执行特定的操作。

ByteBuffer（字节缓冲区）： ByteBuffer 是 Netty 中处理数据的基本数据结构，它提供了对数据的读取和写入操作，支持零拷贝的特性。

Codec（编解码器）：编解码器是用于将原始数据转换为特定协议的消息以及将消息转换为字节数据的组件。Netty 提供了许多内置的编解码器，同时也支持自定义编解码器。

Promise（承诺）： Promise 是一种用于处理异步操作结果的抽象。它可以用于监视和获取异步操作的结果或状态。

Future（未来）： Future 也是用于处理异步操作的抽象，表示一个尚未完成的操作。Netty 中的异步操作通常返回一个 Future，它可以在操作完成时获取结果。

ByteBuf（字节缓冲区）： ByteBuf 是 Netty 中的字节数据容器，它提供了高效的读写操作，并支持引用计数，用于提高性能和降低内存开销。我们将基于Netty Reactor工作架构进行开发，如下是架构图（来源网络）：在这里插入图片描述

实现NettyHttpServer

上面简单的了解了一下Netty的架构，在这个点我将会基于Netty实现一个客户端。大概有如下步骤是我们需要去实现去做的： 1：封装属性： 2：实现构造和init方法 3：epoll优化 4：实现start方法 5：实现shutdown方法

在这个部分我们需要编写完毕的代码大概如下：


java复制代码
package blossom.gateway.core.netty;

import blossom.gateway.common.utils.SystemUtil;
import blossom.gateway.core.LifeCycle;
import blossom.gateway.core.config.GatewayConfig;
import blossom.gateway.core.rule.Rule;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerDomainSocketChannel;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

/**
 * @author: ZhangBlossom
 * @date: 2023/10/24 9:38
 * @contact: QQ:4602197553
 * @contact: WX:qczjhczs0114
 * @blog: https://blog.csdn.net/Zhangsama1
 * @github: https://github.com/ZhangBlossom
 * NettyHttpServer类
 * 类定义:

NettyHttpServer类实现了LifeCycle接口，这意味着它具有初始化、启动和关闭服务器的功能。
构造函数接受一个GatewayConfig对象作为参数，用于配置服务器。
类字段:

config: 存储服务器配置的GatewayConfig对象。
serverBootstrap: 用于配置和启动服务器的Netty ServerBootstrap对象。
eventLoopGroupBoss 和 eventLoopGroupWorker: 用于处理事件循环的两个EventLoopGroup对象。它们用于管理传入连接的事件处理和多线程工作。根据操作系统和系统能力，可能会选择使用NIO或Epoll事件模型。
canUserEpoll()方法:

该方法用于检查当前系统是否可以使用Epoll事件模型。它返回一个布尔值，表示是否可以使用Epoll。Epoll是一种在Linux系统上提供更高性能的事件模型。
init()方法:

init()方法在服务器启动前进行初始化。
如果系统支持Epoll，将创建Epoll事件循环组，否则创建NIO事件循环组。这些事件循环组用于处理服务器的网络事件。
start()方法:

start()方法用于启动服务器。
它配置serverBootstrap，指定事件循环组、通道类型（NIO或Epoll），以及服务器的处理器链。处理器链包括HTTP编解码、HTTP对象聚合、连接管理和HTTP请求处理器。
最后，它绑定服务器并启动，监听指定端口，并记录服务器已启动的消息。
shutdown()方法:

shutdown()方法用于服务器的优雅停机，即关闭服务器。
它关闭eventLoopGroupBoss和eventLoopGroupWorker，释放资源，确保服务器可以安全地关闭。
总的来说，这段代码定义了一个基于Netty的HTTP服务器，它可以在不同操作系统上运行，根据系统支持的特性选择使用NIO或Epoll事件循环组。服务器的主要目标是接受和处理HTTP请求，并且它在初始化、启动和关闭过程中遵循了生命周期模型。
 */
@Slf4j
public class NettyHttpServer implements LifeCycle {
    private final GatewayConfig config;

    private ServerBootstrap serverBootstrap;

    private EventLoopGroup eventLoopGroupBoss;

    private EventLoopGroup eventLoopGroupWorker;

    public NettyHttpServer(GatewayConfig config) {
        this.config = config;
        init();
    }

    @Override
    public void init() {
        //判断当前是否是linux系统且是否可以使用epoll
        //如果是则使用epoll进一步提升性能
        if (canUserEpoll()) {
            this.serverBootstrap = new ServerBootstrap();
            this.eventLoopGroupBoss = new EpollEventLoopGroup(config.getEventLoopGroupBossNum(),
                    new DefaultThreadFactory("netty-boss-nio"));
            this.eventLoopGroupWorker = new EpollEventLoopGroup(config.getEventLoopGroupWorkerNum(),
                    new DefaultThreadFactory("netty-worker-nio"));
        } else {
            this.serverBootstrap = new ServerBootstrap();
            this.eventLoopGroupBoss = new NioEventLoopGroup(config.getEventLoopGroupBossNum(),
                    new DefaultThreadFactory("netty-boss-nio"));
            this.eventLoopGroupWorker = new NioEventLoopGroup(config.getEventLoopGroupWorkerNum(),
                    new DefaultThreadFactory("netty-worker-nio"));

        }
    }

    /**
     * 判断当前系统是否可以使用epoll
     *
     * @return
     */
    public boolean canUserEpoll() {
        return SystemUtil.isLinux() && Epoll.isAvailable();
    }


    /**
     * 开机
     */
    @Override
    public void start() {
        this.serverBootstrap.group(eventLoopGroupBoss,
                eventLoopGroupWorker)
                .channel(canUserEpoll() ?
                EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<Channel>() {

            @Override
            protected void initChannel(Channel channel) throws Exception {
                channel.pipeline().addLast(
                        //http编解码
                        new HttpServerCodec(),
                        new HttpObjectAggregator(config.getMaxContentLength()),
                        new NettyServerConnectionManager(),
                        new NettyHttpServerHandler());
            }
        });
        try {
            this.serverBootstrap.bind().sync();
            log.info("server startup on port {}",config.getPort());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 优雅停机
     */
    @Override
    public void shutdown() {
        if (eventLoopGroupBoss!=null){
            eventLoopGroupBoss.shutdownGracefully();
        }
        if (eventLoopGroupWorker!=null){
            eventLoopGroupWorker.shutdownGracefully();
        }
    }
}

实现NettyHttpServerHandler

我们的NettyHttpServerHandler继承了ChannelInboundHandlerAdapter。继承 ChannelInboundHandlerAdapter 是为了实现自定义的入站（inbound）数据处理逻辑。ChannelInboundHandlerAdapter 是 Netty 提供的抽象类，实现了 ChannelInboundHandler 接口，它提供了一组默认的方法，可以让开发者轻松地处理入站事件。具体的功能可以在网络上搜索到。在这个模块中，我们大概需要实现的功能和实现步骤如下： 1：继承ChannelInboundHandlerAdapter 2：实现channelRead 3：把逻辑委托给NettyProcessor


java复制代码
package blossom.gateway.core.netty;

import blossom.gateway.core.wrapper.HttpRequestWrapper;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.FullHttpRequest;

/**
 * @author: ZhangBlossom
 * @date: 2023/10/24 10:05
 * @contact: QQ:4602197553
 * @contact: WX:qczjhczs0114
 * @blog: https://blog.csdn.net/Zhangsama1
 * @github: https://github.com/ZhangBlossom
 * NettyHttpServerHandler类
 * channelRead() 是一个覆盖 ChannelInboundHandlerAdapter 的方法，用于处理入站数据。
在该方法中，它将传入的 msg 强制转换为 FullHttpRequest，这表示它期望处理的是完整的 HTTP 请求。
然后，它创建了一个 HttpRequestWrapper 对象，用于封装 HTTP 请求，并将请求对象和通道上下文（ctx）设置到 HttpRequestWrapper 中。
最后，它将核心的请求处理逻辑委托给 processor 对象的 process() 方法，将 HttpRequestWrapper 作为参数传递。这意味着实际的请求处理逻辑在 NettyProcessor 中进行。
 */
public class NettyHttpServerHandler extends ChannelInboundHandlerAdapter {

    private final NettyProcessor processor;

    public NettyHttpServerHandler(NettyProcessor processor) {
        this.processor = processor;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
        FullHttpRequest request = (FullHttpRequest)msg;
        HttpRequestWrapper wrapper = new HttpRequestWrapper();
        wrapper.setFullHttpRequest(request);
        wrapper.setContext(ctx);

        //吧核心逻辑委托给另外一个对象
        processor.process(wrapper);
    }
}

实现NettyProcessor

这个接口作为核心接口，我们需要实现如下几个功能： 1：定义接口 2：最小可用版本实现 3：路由函数实现 4：获取异步配置，实现complete方法 5：异常处理 6：写回响应信息并释放资源


java复制代码
package blossom.project.core.netty.processor;

import blossom.gateway.common.enums.ResponseCode;
import blossom.gateway.common.exception.BaseException;
import blossom.gateway.common.exception.ConnectException;
import blossom.gateway.common.exception.ResponseException;
import blossom.project.core.ConfigLoader;
import blossom.project.core.context.GatewayContext;
import blossom.project.core.context.HttpRequestWrapper;
import blossom.project.core.helper.AsyncHttpHelper;
import blossom.project.core.helper.RequestHelper;
import blossom.project.core.helper.ResponseHelper;
import blossom.project.core.response.GatewayResponse;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.asynchttpclient.Request;
import org.asynchttpclient.Response;


import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

/**
 * @author: ZhangBlossom
 * @date: 2023/10/23 19:57
 * @contact: QQ:4602197553
 * @contact: WX:qczjhczs0114
 * @blog: https://blog.csdn.net/Zhangsama1
 * @github: https://github.com/ZhangBlossom
 */
@Slf4j
public class NettyCoreProcessor implements NettyProcessor {
    @Override
    public void process(HttpRequestWrapper wrapper) {
        FullHttpRequest request = wrapper.getRequest();
        ChannelHandlerContext ctx = wrapper.getCtx();

        try {
            GatewayContext gatewayContext = RequestHelper.doContext(request, ctx);
            route(gatewayContext);
        } catch (BaseException e) {
            log.error("process error {} {}", e.getCode().getCode(), e.getCode().getMessage());
            FullHttpResponse httpResponse = ResponseHelper.getHttpResponse(e.getCode());
            //回写数据并且释放资源
            doWriteAndRelease(ctx, request, httpResponse);
        } catch (Throwable t) {
            log.error("process unkown error", t);
            FullHttpResponse httpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
            doWriteAndRelease(ctx, request, httpResponse);
        }

    }

    private void doWriteAndRelease(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse httpResponse) {
        ctx.writeAndFlush(httpResponse)
                .addListener(ChannelFutureListener.CLOSE); //释放资源后关闭channel
        ReferenceCountUtil.release(request);
    }

    private void route(GatewayContext gatewayContext) {
        Request request = gatewayContext.getRequest().build();
        CompletableFuture<Response> future = AsyncHttpHelper.getInstance().executeRequest(request);

        boolean whenComplete = ConfigLoader.getConfig().isWhenComplete();
        //判断是否是单异步
        if (whenComplete) {
            future.whenComplete((response, throwable) -> {
               complete(request, response, throwable, gatewayContext);
            });
        } else {
            future.whenCompleteAsync((response, throwable) -> {
                complete(request, response, throwable, gatewayContext);
            });
        }
    }

    private void complete(Request request,
                          Response response,
                          Throwable throwable,
                          GatewayContext gatewayContext) {
        gatewayContext.releaseRequest();

        try {
            //异常信息处理
            if (Objects.nonNull(throwable)) {
                String url = request.getUrl();
                if (throwable instanceof TimeoutException) {
                    log.warn("complete time out {}", url);
                    gatewayContext.setThrowable(new ResponseException(ResponseCode.REQUEST_TIMEOUT));
                } else {
                    gatewayContext.setThrowable(new ConnectException(throwable,
                            gatewayContext.getUniqueId(),
                            url, ResponseCode.HTTP_RESPONSE_ERROR));
                }
            } else {
                //没有异常则正常响应结果
                gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(response));
            }
        } catch (Throwable t) {
            gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
            log.error("complete error", t);
        } finally {
            gatewayContext.written();
            ResponseHelper.writeResponse(gatewayContext);
        }
    }
}

实现NettyHttpClient

上面我们已经实现了Server服务端，现在开始实现客户端。大概步骤如下： 1：实现LifeCyclke接口 2：封装属性 3：实现init方法 4：实现start方法 5：实现shutdown方法可以发现其实client的实现和server的实现都是差不多的。


java复制代码
@Slf4j
public class NettyHttpClient implements LifeCycle {
    private final Config config;

    private final EventLoopGroup eventLoopGroupWoker;

    private AsyncHttpClient asyncHttpClient;

    public NettyHttpClient(Config config, EventLoopGroup eventLoopGroupWoker) {
        this.config = config;
        this.eventLoopGroupWoker = eventLoopGroupWoker;
        init();
    }

    @Override
    public void init() {
        DefaultAsyncHttpClientConfig.Builder builder = new DefaultAsyncHttpClientConfig.Builder()
                .setEventLoopGroup(eventLoopGroupWoker)
                .setConnectTimeout(config.getHttpConnectTimeout())
                .setRequestTimeout(config.getHttpRequestTimeout())
                .setMaxRedirects(config.getHttpMaxRequestRetry())
                .setAllocator(PooledByteBufAllocator.DEFAULT) //池化的byteBuf分配器，提升性能
                .setCompressionEnforced(true)
                .setMaxConnections(config.getHttpMaxConnections())
                .setMaxConnectionsPerHost(config.getHttpConnectionsPerHost())
                .setPooledConnectionIdleTimeout(config.getHttpPooledConnectionIdleTimeout());
        this.asyncHttpClient = new DefaultAsyncHttpClient(builder.build());
    }

    @Override
    public void start() {
        AsyncHttpHelper.getInstance().initialized(asyncHttpClient);
    }

    @Override
    public void shutdown() {
        if (asyncHttpClient != null) {
            try {
                this.asyncHttpClient.close();
            } catch (IOException e) {
                log.error("NettyHttpClient shutdown error", e);
            }
        }
    }
}

实现核心容器

在上面我们已经实现完毕了Netty相关的代码，接下来我们要去实现核心容器。大概步骤如下： 1：实现LifeCyclke接口 2：封装属性 3：实现init方法 4：实现start方法 5：实现shutdown方法实现了核心容器，就可以用来启动我们的Netty的客户端和服务端了，只要完成了这一步，我们简答的请求转发和接收就已经完成了。


java复制代码
package blossom.project.core;

import blossom.project.core.netty.NettyHttpClient;
import blossom.project.core.netty.NettyHttpServer;
import blossom.project.core.netty.processor.NettyCoreProcessor;
import blossom.project.core.netty.processor.NettyProcessor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Container implements LifeCycle {
    private final Config config;

    private NettyHttpServer nettyHttpServer;

    private NettyHttpClient nettyHttpClient;

    private NettyProcessor nettyProcessor;

    public Container(Config config) {
        this.config = config;
        init();
    }

    @Override
    public void init() {
        this.nettyProcessor = new NettyCoreProcessor();

        this.nettyHttpServer = new NettyHttpServer(config, nettyProcessor);

        this.nettyHttpClient = new NettyHttpClient(config,
                nettyHttpServer.getEventLoopGroupWoker());
    }

    @Override
    public void start() {
        nettyHttpServer.start();;
        nettyHttpClient.start();
        log.info("api gateway started!");
    }

    @Override
    public void shutdown() {
        nettyHttpServer.shutdown();
        nettyHttpClient.shutdown();
    }
}