InfiniGate自研网关实现四

news2024/11/28 15:59:18

13.服务发现组件搭建和注册网关连接

以封装 api-gateway-core 为目的，搭建 SpringBoot Starter 组件，用于服务注册发现的相关内容处理。

这里最大的目的在于搭建起用于封装网关算力服务的 api-gateway-core 系统，提供网关服务注册发现能力。那么之所以要开发一个这样的组件，也就是 SpringBoot Starter。是因为我们希望把这样的统一公用能力进行一致的管理，如果没有这样的组件服务，那么将需要每一个 SpringBoot 服务都要做类似这样的事情，整体来看就会耗费很大的成本，所以要把这样的功能进行收口。由于我在开发api-gateway-core时为了项目的可扩展性，没有使用SpringBoot框架，因此api-gateway-core还不能直接封装为一个SpringBoot Starter组件。所以我写了一个api-gateway-assist 这个辅助组件，它是一个 SpringBoot Starter 起到包装和连接的作用。

api-gateway-core 是网关的算力服务，api-gateway-center 是网关的注册中心，那么为了把这块服务链接起来，中间则需要一套 api-gateway-engin 网关的引擎，用于启动网关的算力服务。
但由于启动网关的算力服务还需要一些功能的整合，来包装网关算力到注册中心的连接，所以这部分需要整合到 api-gateway-assist 这个辅助组件，它是一个 SpringBoot Starter 起到包装和连接的作用。

具体实现：

不熟悉SpringBoot自定义Starter的可以先看一下这篇文章：中间件研发之Springboot自定义starter-CSDN博客

1.在config包下定义配置类GatewayServiceProperties，在类中添加属性信息，这些属性信息就是最后的配置到 yml 中的配置属性。

2.在service包下创建RegisterGatewayService 注册网关类，在该类中调用 api-gateway-center 提供的服务发现接口，向网关中心注册网关算力节点。这是第一步非常重要的关联作用，有了这块逻辑的处理，才能打通整个网关算力和网关注册中心

3.在application包下创建GatewayApplication类实现Spring框架中的ApplicationListener接口，用于监听Spring上下文刷新事件ContextRefreshedEvent。这个类的目的是在Spring应用上下文刷新时执行一些初始化操作，比如注册网关服务到配置中心。重复注册则是标记服务启动

包结构如下：

14.网关映射聚合信息查询实现

以封装 api-gateway-core 为目的，搭建 SpringBoot Starter 组件，用于服务注册发现的相关内容处理.

这里主要实现在注册中心提供一个接口，这个接口用来：将各个RPC服务配置【系统、接口、方法】信息拉取下来（其实就是一个根据网关算力节点查询对应的RPC服务），注册到网关算力中，完成RPC映射的过程。

首先通过 gateway_distribution 表，把网关和RPC应用服务关联起来，方便知道哪个网关算力处理哪些RPC映射管理。有了这个映射关系后，就可以把对应的 application_interface、application_interface_method、application_system 三个表维护应用的配置信息关联起来了。

具体实现：

1.在用户接口层interfaces包下的GatewayConfigManage类中新增一个queryApplicationSystemRichInfo接口，用于根据算力节点id查询对应的RPC服务。

2.在应用层application包下的IConfigManageService接口中新增一个方法queryApplicationSystemRichInfo（防腐层）

3.在领域层domain的对应领域包下实现上面接口中新增的方法，在这个方法中写查询RPC服务的逻辑。调用持久层的接口是该业务子领域的repository包下接口的方法（防腐层）。

4.在infrastructure基础层包下的repository包下实现防腐层的接口，实现的方法主要是对要增删改的数据进行一些填充或者是对查询出来的数据进行封装，然后再调用dao层的接口，真正地去操作数据库的数据。

15.服务配置拉取和组件使用验证

结合第13节组件的开发和第14节数据结构的提供，在第15节中重构网关助手组件服务，并把组件引入到 SpringBoot 中使用。

结合着第13节，网关算力的助手组件初步实现，本章需要进行扩展以及验证使用。这一节需要完成从第14节中提供的网关聚合配置信息，拉取到网关算力中，打通这部分的接口调用。后续再处理映射操作，因为映射还需要把网关核心算力引入到助手组件中进行包装使用。

具体实现：

1.把第13节中写的api-gateway-assist下的service包进行重构，重构成DDD架构下domain领域包。

2.在service包下的GatewayCenterService类中新增pullApplicationSystemRichInfo方法，用于拉取注册中心的网关算力节点对应的RPC服务信息。

16.网络通信配置提取

提取 Netty 通信服务配置信息，包括：IP、端口、线程等，到会话的 Configuration 配置类中进行统一的维护和管理，便于外部调用方向内部传递信息。

在网关会话中会从注册中心拉取网关算力节点的服务配置信息，这些配置信息存放到Configuration配置类。

具体实现:

只需修改Netty启动类的相关配置，改为从Configuration配置类中获取

public class GatewaySocketServer implements Callable<Channel> {

    private final Logger logger = LoggerFactory.getLogger(GatewaySocketServer.class);

    private final Configuration configuration;
    private DefaultGatewaySessionFactory gatewaySessionFactory;

    private EventLoopGroup boss;
    private EventLoopGroup work;
    private Channel channel;

    public GatewaySocketServer(Configuration configuration, DefaultGatewaySessionFactory gatewaySessionFactory) {
        this.configuration = configuration;
        this.gatewaySessionFactory = gatewaySessionFactory;
        this.initEventLoopGroup();
    }

    private void initEventLoopGroup(){
        boss = new NioEventLoopGroup(configuration.getBossNThreads());
        work = new NioEventLoopGroup(configuration.getWorkNThreads());
    }

    @Override
    public Channel call() throws Exception {
        ChannelFuture channelFuture = null;
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(boss, work)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new GatewayChannelInitializer(configuration, gatewaySessionFactory));
            // Docker 容器部署会自动分配IP，所以我们只设定端口即可。
            // channelFuture = b.bind(new InetSocketAddress(configuration.getHostName(), configuration.getPort())).syncUninterruptibly();
            channelFuture = b.bind(configuration.getPort()).syncUninterruptibly();
            this.channel = channelFuture.channel();
        } catch (Exception e) {
            logger.error("socket server start error.", e);
        } finally {
            if (null != channelFuture && channelFuture.isSuccess()) {
                logger.info("socket server start done.");
            } else {
                logger.error("socket server start error.");
            }
        }
        return channel;
    }

}