ShenYu网关注册中心之HTTP注册原理

news2024/11/26 4:48:14

文章目录

  • 1、客户端注册流程
    • 1.1、读取配置
      • 1.1.1、用于注册的 HttpClientRegisterRepository
      • 1.1.2、用于扫描构建 元数据 和 URI 的 SpringMvcClientEventListener
    • 1.2、扫描注解,注册元数据和URI
      • 1.2.1、构建URI并写入Disruptor
      • 1.2.2、构建元数据并写入Disruptor
      • 1.2.3、Disruptor消费数据并向shenyu-admin注册数据
  • 2、服务端注册流程
    • 2.1、读取配置
      • 2.1.1、用于监听的ShenyuClientServerRegisterRepository
    • 2.2、注册元数据和URI
      • 2.2.1、注册接口接收数据写入Disruptor
      • 2.2.2、Disruptor消费数据并持久化

1、客户端注册流程

当客户端启动后,根据相关配置,读取属性信息,然后写入队列。以官方提供的 shenyu-examples-http 为例,开始源码分析。

1.1、读取配置

该例子是一个springboot项目,所以注册的入口往往在自动装配类中。不妨可以先看下项目的pom文件中引入了什么依赖:

<dependencies>
    <dependency>
        <groupId>org.apache.shenyu</groupId>
        <artifactId>shenyu-spring-boot-starter-client-springmvc</artifactId>
        <version>${project.version}</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-logging</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

这里面看到就shenyu-spring-boot-starter-client-springmvc是跟ShenYu相关的,所以入口应该就在这个依赖内了,看下这个依赖的项目结构:
在这里插入图片描述
发现就是两个配置类,ShenyuSpringMvcClientInfoRegisterConfiguration由于使用了@Configuration(proxyBeanMethods = false),暂时不用关注,重点关注ShenyuSpringMvcClientConfiguration,它是shenyu客户端http注册配置类。

/**
 * shenyu 客户端http注册配置类
 */
@Configuration
// shenyu客户端通用配置类
@ImportAutoConfiguration(ShenyuClientCommonBeanConfiguration.class)
@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")
public class ShenyuSpringMvcClientConfiguration {

    static {
        VersionUtils.checkDuplicate(ShenyuSpringMvcClientConfiguration.class);
    }

    /**
     *
     * 监听并处理http元数据和URI信息的注册
     *
     * @param clientConfig                   客户端注册配置
     * @param shenyuClientRegisterRepository 客户端注册类
     */
    @Bean
    @ConditionalOnMissingBean(ClientRegisterConfiguration.class)
    // 这里的两个参数是由ShenyuClientCommonBeanConfiguration导入的
    public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,
                                                                          final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
        return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);
    }
}

通过@Configuration表示这是一个配置类,通过@ImportAutoConfiguration引入ShenyuClientCommonBeanConfiguration配置类。

/**
 * shenyu客户端通用配置类,创建注册中心客户端通用的bean
 */
@Configuration
@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")
public class ShenyuClientCommonBeanConfiguration {
    
    /**
     * 根据注册中心配置通过SPI方式创建客户端注册类
     */
    @Bean
    public ShenyuClientRegisterRepository shenyuClientRegisterRepository(final ShenyuRegisterCenterConfig config) {
        return ShenyuClientRegisterRepositoryFactory.newInstance(config);
    }
    
    /**
     * Shenyu 客户端注册中心配置,读取shenyu.register属性配置
     */
    @Bean
    @ConfigurationProperties(prefix = "shenyu.register")
    public ShenyuRegisterCenterConfig shenyuRegisterCenterConfig() {
        return new ShenyuRegisterCenterConfig();
    }
    
    /**
     * Shenyu 客户端配置,读取shenyu.client属性配置
     */
    @Bean
    @ConfigurationProperties(prefix = "shenyu")
    public ShenyuClientConfig shenyuClientConfig() {
        return new ShenyuClientConfig();
    }
}

ShenyuClientCommonBeanConfigurationShenYu客户端的通用配置类,创建了3个通用bean。

  • ShenyuClientRegisterRepository:客户端注册类,用于将客户端接口信息注册到注册中心。
  • ShenyuRegisterCenterConfig:ShenYu客户端注册中心配置类,读取shenyu.register属性配置。
  • ShenyuClientConfig:ShenYu客户端配置类,读取shenyu.client属性配置。

1.1.1、用于注册的 HttpClientRegisterRepository

上面生成的ShenyuClientRegisterRepository是用于实现客户端注册的接口,会根据注册中心的配置通过SPI方式创建客户端注册类,每一个注册方式都对应一个实现类。
在这里插入图片描述
目前支持7种注册类型:

  • Http:HttpClientRegisterRepository
  • Apollo:ApolloClientRegisterRepository
  • Zookeeper:ZookeeperClientRegisterRepository
  • Etcd:EtcdClientRegisterRepository
  • Nacos:NacosClientRegisterRepository
  • Consul:ConsulClientRegisterRepository
  • Polaris:PolarisClientRegisterRepository
public final class ShenyuClientRegisterRepositoryFactory {
    
    private static final Map<String, ShenyuClientRegisterRepository> REPOSITORY_MAP = new ConcurrentHashMap<>();
    
    /**
     * 根据注册中心类型实例化注册服务
     */
    public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {
        if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {
            // 通过SPI方式创建客户端注册类
            ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());
            // 初始化对应客户端注册类,比如创建zookeeper client,etcd client,admin平台的token等
            result.init(shenyuRegisterCenterConfig);
            ShenyuClientShutdownHook.set(result, shenyuRegisterCenterConfig.getProps());
            REPOSITORY_MAP.put(shenyuRegisterCenterConfig.getRegisterType(), result);
            return result;
        }
        return REPOSITORY_MAP.get(shenyuRegisterCenterConfig.getRegisterType());
    }
}

加载类型通过registerType指定,也就是我们在配置文件中指定的类型:

shenyu:
  register:
    registerType: http
    serverLists: http://localhost:9095
    props:
      username: admin
      password: 123qweQWE$$

这里指定的是http,所以这里创建的就是HttpClientRegisterRepositoryprops是用于连接注册中心的一些额外属性,比如用户名,密码,命名空间等。
创建对应的注册客户端后,会调用init方法根据shenyu.register下的配置进行初始化:

@Join
public class HttpClientRegisterRepository extends FailbackRegistryRepository {

    public HttpClientRegisterRepository() {
    }

    public HttpClientRegisterRepository(final ShenyuRegisterCenterConfig config) {
        init(config);
    }

    @Override
    public void init(final ShenyuRegisterCenterConfig config) {
        // shenyu-admin用户名
        this.username = config.getProps().getProperty(Constants.USER_NAME);
        // shenyu-admin密码
        this.password = config.getProps().getProperty(Constants.PASS_WORD);
        // shenyu-admin集群地址
        this.serverList = Lists.newArrayList(Splitter.on(",").split(config.getServerLists()));
        // 根据上面3个信息请求shenyu-admin获取accessToken,用于后面调用注册接口
        this.accessToken = Caffeine.newBuilder()
        //see org.apache.shenyu.admin.config.properties.JwtProperties#expiredSeconds
        .expireAfterWrite(24L, TimeUnit.HOURS)
        .build(new CacheLoader<String, String>() {
            @Override
            public @Nullable String load(@NonNull final String server) throws Exception {
                try {
                    // 调用shenyu-admin的登录接口(/platform/login),获取accessToken
                    Optional<?> login = RegisterUtils.doLogin(username, password, server.concat(Constants.LOGIN_PATH));
                    return login.map(String::valueOf).orElse(null);
                } catch (Exception e) {
                    LOGGER.error("Login admin url :{} is fail, will retry. cause: {} ", server, e.getMessage());
                    return null;
                }
            }
        });
    }
}

这里主要就是去调shenyu-admin的登录接口获取accessToken,为后面的发送注册数据做准备。其他注册类型的ShenyuClientRegisterRepository也一样,创建各自注册中心的client,连接注册中心,为发送数据做准备。类注解@Join用于SPI的加载。

1.1.2、用于扫描构建 元数据 和 URI 的 SpringMvcClientEventListener

回到一开始的ShenyuSpringMvcClientConfiguration配置类:

/**
 * shenyu 客户端http注册配置类
 */
@Configuration
// shenyu客户端通用配置类
@ImportAutoConfiguration(ShenyuClientCommonBeanConfiguration.class)
@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")
public class ShenyuSpringMvcClientConfiguration {

    static {
        VersionUtils.checkDuplicate(ShenyuSpringMvcClientConfiguration.class);
    }

    /**
     *
     * 监听并处理http元数据和URI信息的注册
     *
     * @param clientConfig                   客户端注册配置
     * @param shenyuClientRegisterRepository 客户端注册类
     */
    @Bean
    @ConditionalOnMissingBean(ClientRegisterConfiguration.class)
    // 这里的两个参数是由ShenyuClientCommonBeanConfiguration导入的
    public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,
                                                                          final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
        return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);
    }
}

创建了SpringMvcClientEventListener,负责客户端 元数据URI 数据的构建和注册。SpringMvcClientEventListener继承了AbstractContextRefreshedEventListener,而AbstractContextRefreshedEventListener是一个抽象类,它实现了ApplicationListener接口,并重写了onApplicationEvent()方法,当有Spring事件发生后,该方法会执行。每一种后端服务RPC调用协议都对应了一个监听类。
在这里插入图片描述

public class SpringMvcClientEventListener extends AbstractContextRefreshedEventListener<Object, ShenyuSpringMvcClient> {

    public SpringMvcClientEventListener(final PropertiesConfig clientConfig,
                                        final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
        super(clientConfig, shenyuClientRegisterRepository);
        // client配置
        Properties props = clientConfig.getProps();
        // 是否是全部接口都注册
        this.isFull = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.IS_FULL, Boolean.FALSE.toString()));
        // http协议
        this.protocol = props.getProperty(ShenyuClientConstants.PROTOCOL, ShenyuClientConstants.HTTP);
        this.addPrefixed = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.ADD_PREFIXED,
                Boolean.FALSE.toString()));
        mappingAnnotation.add(ShenyuSpringMvcClient.class);
        mappingAnnotation.add(RequestMapping.class);
    }

    // ...

}

SpringMvcClientEventListener的构造函数主要就是调用父类AbstractContextRefreshedEventListener的构造函数,传入客户端配置和客户端注册类,客户端配置指shenyu.client.http下的配置:

shenyu:
  client:
      http:
        props:
          contextPath: /http
          appName: http-appName
          port: 8189
          isFull: false
public abstract class AbstractContextRefreshedEventListener<T, A extends Annotation> implements ApplicationListener<ContextRefreshedEvent> {

    protected static final String PATH_SEPARATOR = "/";

    // Disruptor 发布器
    private final ShenyuClientRegisterEventPublisher publisher = ShenyuClientRegisterEventPublisher.getInstance();
	// ...
    
    public AbstractContextRefreshedEventListener(final PropertiesConfig clientConfig,
                                                 final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
        // 读取 shenyu.client.http 配置信息
        Properties props = clientConfig.getProps();
        this.appName = props.getProperty(ShenyuClientConstants.APP_NAME);
        this.contextPath = Optional.ofNullable(props.getProperty(ShenyuClientConstants.CONTEXT_PATH)).map(UriUtils::repairData).orElse("");
        if (StringUtils.isBlank(appName) && StringUtils.isBlank(contextPath)) {
            String errorMsg = "client register param must config the appName or contextPath";
            LOG.error(errorMsg);
            throw new ShenyuClientIllegalArgumentException(errorMsg);
        }
        this.ipAndPort = props.getProperty(ShenyuClientConstants.IP_PORT);
        this.host = props.getProperty(ShenyuClientConstants.HOST);
        this.port = props.getProperty(ShenyuClientConstants.PORT);
        // 开始事件发布,启动 Disruptor
        publisher.start(shenyuClientRegisterRepository);
    }   

}

取出相关配置信息后,就启动 Disruptor 队列,ShenyuClientRegisterEventPublisher可以看作是一个生产者,用来向队列发送数据

public class ShenyuClientRegisterEventPublisher {
    
    private static final ShenyuClientRegisterEventPublisher INSTANCE = new ShenyuClientRegisterEventPublisher();
    
    private DisruptorProviderManage<DataTypeParent> providerManage;
 
    public static ShenyuClientRegisterEventPublisher getInstance() {
        return INSTANCE;
    }
    
    /**
     * Start.
     *
     * @param shenyuClientRegisterRepository shenyuClientRegisterRepository
     */
    public void start(final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
        // 注册任务工厂类,用于创建注册的任务,客户端使用的是RegisterClientExecutorFactory, 
        // 而在服务端(shenyu-admin)用于处理注册任务的是RegisterServerConsumerExecutor,
        // 都是用于消费Disruptor数据的任务
        RegisterClientExecutorFactory factory = new RegisterClientExecutorFactory();
        // 添加元数据订阅器
        factory.addSubscribers(new ShenyuClientMetadataExecutorSubscriber(shenyuClientRegisterRepository));
        // 添加URI订阅器
        factory.addSubscribers(new ShenyuClientURIExecutorSubscriber(shenyuClientRegisterRepository));
        // 添加ApiDoc订阅器
        factory.addSubscribers(new ShenyuClientApiDocExecutorSubscriber(shenyuClientRegisterRepository));
        providerManage = new DisruptorProviderManage<>(factory);
        // 启动Disruptor队列,并创建消费者
        providerManage.startup();
    }
    
    /**
     * 发布事件,向Disruptor队列发数据
     *
     * @param data the data
     */
    public void publishEvent(final DataTypeParent data) {
        DisruptorProvider<DataTypeParent> provider = providerManage.getProvider();
        provider.onData(data);
    }
}

start方法主要是为队列添加订阅器,会由消费者接收到信息后调用这些订阅器。然后启动启动Disruptor队列,并创建消费者。

public class DisruptorProviderManage<T> {

    public void startup() {
        this.startup(false);
    }
    
    public void startup(final boolean isOrderly) {
        OrderlyExecutor executor = new OrderlyExecutor(isOrderly, consumerSize, consumerSize, 0, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                DisruptorThreadFactory.create("shenyu_disruptor_consumer_", false), new ThreadPoolExecutor.AbortPolicy());
        int newConsumerSize = this.consumerSize;
        EventFactory<DataEvent<T>> eventFactory;
        if (isOrderly) {
            newConsumerSize = 1;
            eventFactory = new OrderlyDisruptorEventFactory<>();
        } else {
            eventFactory = new DisruptorEventFactory<>();
        }
        Disruptor<DataEvent<T>> disruptor = new Disruptor<>(eventFactory,
                size,
                DisruptorThreadFactory.create("shenyu_disruptor_provider_" + consumerFactory.fixName(), false),
                ProducerType.MULTI,
                new BlockingWaitStrategy());
        // 创建消费者
        @SuppressWarnings("all")
        QueueConsumer<T>[] consumers = new QueueConsumer[newConsumerSize];
        for (int i = 0; i < newConsumerSize; i++) {
            consumers[i] = new QueueConsumer<>(executor, consumerFactory);
        }
        // 设置消费者
        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
        // 真正调用disruptor的api启动
        disruptor.start();
        RingBuffer<DataEvent<T>> ringBuffer = disruptor.getRingBuffer();
        // disruptor的生产者
        provider = new DisruptorProvider<>(ringBuffer, disruptor, isOrderly);
    }  

}

这里就是准备Disruptor队列的一些逻辑,就不细讲了,其中QueueConsumerDisruptor的消费者,后面就是由它接收数据。

1.2、扫描注解,注册元数据和URI

上面说到SpringMvcClientEventListener继承了AbstractContextRefreshedEventListener,而AbstractContextRefreshedEventListener实现了ApplicationListener接口,并重写了onApplicationEvent()方法,当有Spring事件发生后,该方法会执行。

// 当有上下文刷新事件ContextRefreshedEvent发生时,该方法会执行,算是客户端的执行入口吧
@Override
public void onApplicationEvent(@NonNull final ContextRefreshedEvent event) {
    context = event.getApplicationContext();
    // 获取客户端的接口类,比如http就是Controller类,dubbo就是@DubboService类,由子类实现
    Map<String, T> beans = getBeans(context);
    if (MapUtils.isEmpty(beans)) {
        return;
    }
    // 保证只注册一次
    if (!registered.compareAndSet(false, true)) {
        return;
    }
    // 构建URI并写入Disruptor,由子类实现
    publisher.publishEvent(buildURIRegisterDTO(context, beans));
    // 构建元数据并写入Disruptor
    beans.forEach(this::handle);
    Map<String, Object> apiModules = context.getBeansWithAnnotation(ApiModule.class);
    apiModules.forEach((k, v) -> handleApiDoc(v, beans));
}

获取客户端服务的接口类,由具体的子类实现,http就是Controller类,这里对应的子类就是SpringMvcClientEventListener

@Override
protected Map<String, Object> getBeans(final ApplicationContext context) {
    // Filter out
    if (Boolean.TRUE.equals(isFull)) {
        // isFull=true,表示代理整个服务,就不需要注解扫描了,
        // 直接构建元数据和URI,写入Disruptor
        getPublisher().publishEvent(MetaDataRegisterDTO.builder()
                .contextPath(getContextPath())
                .addPrefixed(addPrefixed)
                .appName(getAppName())
                .path(PathUtils.decoratorPathWithSlash(getContextPath()))
                .rpcType(RpcTypeEnum.HTTP.getName())
                .enabled(true)
                .ruleName(getContextPath())
                .build());
        LOG.info("init spring mvc client success with isFull mode");
        // 构建URI
        publisher.publishEvent(buildURIRegisterDTO(context, Collections.emptyMap()));
        return Collections.emptyMap();
    }
    // 否则获取@Controller注解的bean
    return context.getBeansWithAnnotation(Controller.class);
}

这里会判断配置文件中的shenyu.client.http.props.isFull,如果是true,则直接构建一个元数据URI,写入到Disruptor中,然后返回一个空集合,后续的逻辑就没执行了。如果是false,则从spring容器中获取带@Controller注解的bean返回。

1.2.1、构建URI并写入Disruptor

构建一个URI数据写入到Disruptor,这个也是由子类实现的:

// 构建URI
@Override
protected URIRegisterDTO buildURIRegisterDTO(final ApplicationContext context,
                                             final Map<String, Object> beans) {
    try {
        return URIRegisterDTO.builder()
                .contextPath(getContextPath()) // shneyu得contextPath
                .appName(getAppName()) // appName
                .protocol(protocol) // 服务协议
                .host(super.getHost()) // 服务host
                .port(Integer.valueOf(getPort())) // 服务端口
                .rpcType(RpcTypeEnum.HTTP.getName()) // rpc类型
                .eventType(EventType.REGISTER) // 事件类型
                .build();
    } catch (ShenyuException e) {
        throw new ShenyuException(e.getMessage() + "please config ${shenyu.client.http.props.port} in xml/yml !");
    }
}

可以看出来URI跟接口类没有关系,一个后端服务实例生成一个URI。

1.2.2、构建元数据并写入Disruptor

之后遍历每个接口构建元数据beans.forEach(this::handle)

/**
 * 构建元数据并写入Disruptor
 */
protected void handle(final String beanName, final T bean) {
    Class<?> clazz = getCorrectedClass(bean);
    // 获取当前bean的对应shenyu客户端的注解,比如http是@ShenyuSpringMvcClient, 
    // dubbo是@ShenyuDubboClient
    final A beanShenyuClient = AnnotatedElementUtils.findMergedAnnotation(clazz, getAnnotationType());
    // 获取bean对应的path(类上注解的路径),由子类实现
    final String superPath = buildApiSuperPath(clazz, beanShenyuClient);
    // 如果有shenyu客户端注解并且path中包含*,则表示要注册整个类的方法,只需要构建一个类元数据
    if (Objects.nonNull(beanShenyuClient) && superPath.contains("*")) {
        // 由具体的子类构建类元数据写入Disruptor
        handleClass(clazz, bean, beanShenyuClient, superPath);
        return;
    }
    // 类上没有shenyu客户端注解(类上没有注解,但方法上有注解,也是可以注册的),
    // 或者有注解但是path没有包含*,则就要遍历每个方法,为每个需要注册的方法构建方法元数据
    final Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(clazz);
    for (Method method : methods) {
        // 由具体子类构建方法元数据写入Disruptor,并将每个method对应的元数据对象缓存在当前类里
        handleMethod(bean, clazz, beanShenyuClient, method, superPath);
    }
}

protected void handleClass(final Class<?> clazz,
                           final T bean,
                           @NonNull final A beanShenyuClient,
                           final String superPath) {
    publisher.publishEvent(buildMetaDataDTO(bean, beanShenyuClient, pathJoin(contextPath, superPath), clazz, null));
}

protected void handleMethod(final T bean,
                            final Class<?> clazz,
                            @Nullable final A beanShenyuClient,
                            final Method method,
                            final String superPath) {
    // 如果方法上有Shenyu客户端注解,就表示该方法需要注册
    A methodShenyuClient = AnnotatedElementUtils.findMergedAnnotation(method, getAnnotationType());
    if (Objects.nonNull(methodShenyuClient)) {
        final MetaDataRegisterDTO metaData = buildMetaDataDTO(bean, methodShenyuClient,
                buildApiPath(method, superPath, methodShenyuClient), clazz, method);
        publisher.publishEvent(metaData);
        metaDataMap.put(method, metaData);
    }
}
// 获取接口对应路径,如果shenyu注解上没有,就用@RequestMapping上的路径,
// 但是这个只支持第一个路径
@Override
protected String buildApiSuperPath(final Class<?> clazz, @Nullable final ShenyuSpringMvcClient beanShenyuClient) {
    if (Objects.nonNull(beanShenyuClient) && StringUtils.isNotBlank(beanShenyuClient.path())) {
        return beanShenyuClient.path();
    }
    RequestMapping requestMapping = AnnotationUtils.findAnnotation(clazz, RequestMapping.class);
    // Only the first path is supported temporarily
    if (Objects.nonNull(requestMapping) && ArrayUtils.isNotEmpty(requestMapping.path()) && StringUtils.isNotBlank(requestMapping.path()[0])) {
        return requestMapping.path()[0];
    }
    return "";
}

// springmvc接口上需要有 ShenyuSpringMvcClient 注解,
// 并且包含RequestMapping注解(表示是一个接口),才进行注册
protected void handleMethod(final Object bean, final Class<?> clazz,
                            @Nullable final ShenyuSpringMvcClient beanShenyuClient,
                            final Method method, final String superPath) {
    final RequestMapping requestMapping = AnnotatedElementUtils.findMergedAnnotation(method, RequestMapping.class);
    ShenyuSpringMvcClient methodShenyuClient = AnnotatedElementUtils.findMergedAnnotation(method, ShenyuSpringMvcClient.class);
    methodShenyuClient = Objects.isNull(methodShenyuClient) ? beanShenyuClient : methodShenyuClient;
    // 如果有 ShenyuSpringMvcClient 注解并且包含RequestMapping注解(表示是一个接口),则进行注册
    if (Objects.nonNull(methodShenyuClient) && Objects.nonNull(requestMapping)) {
        // 构建元数据
        final MetaDataRegisterDTO metaData = buildMetaDataDTO(bean, methodShenyuClient,
                // 构建path = contextPath + 类上的路径 + 方法上的路径                                        
                buildApiPath(method, superPath, methodShenyuClient), clazz, method);
        // 发布元数据
        getPublisher().publishEvent(metaData);
        getMetaDataMap().put(method, metaData);
    }
}

// path = contextPath + 类上的路径 + 方法上的路径,
// 如果@ShenyuSpringMvcClient注解上的路径不为空,则方法上的路径=@ShenyuSpringMvcClient上的value,
// 否则,方法上的路径=@RequestMapping上的value
@Override
protected String buildApiPath(final Method method, final String superPath,
                              @NonNull final ShenyuSpringMvcClient methodShenyuClient) {
    String contextPath = getContextPath();
    if (StringUtils.isNotBlank(methodShenyuClient.path())) {
        return pathJoin(contextPath, superPath, methodShenyuClient.path());
    }
    final String path = getPathByMethod(method);
    if (StringUtils.isNotBlank(path)) {
        return pathJoin(contextPath, superPath, path);
    }
    return pathJoin(contextPath, superPath);
}

1.2.3、Disruptor消费数据并向shenyu-admin注册数据

上面启动Disruptor的时候说到QueueConsumer实现了WorkHandler接口,是Disruptor的消费者,消费逻辑就在它的onEvent方法中:

public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {
    
    private final OrderlyExecutor executor;
    
    private final QueueConsumerFactory<T> factory;
    
    /**
     * Instantiates a new Queue consumer.
     *
     * @param executor the executor
     * @param factory  the factory
     */
    public QueueConsumer(final OrderlyExecutor executor, final QueueConsumerFactory<T> factory) {
        this.executor = executor;
        this.factory = factory;
    }
    
    @Override
    public void onEvent(final DataEvent<T> t) {
        if (Objects.nonNull(t)) {
            // 根据事件类型使用不同的线程池
            ThreadPoolExecutor executor = orderly(t);
            // 通过工厂创建队列消费任务 RegisterClientConsumerExecutor
            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();
            // 为消费任务设置数据
            queueConsumerExecutor.setData(t.getData());
            t.setData(null);
            // 放在线程池中执行 消费任务
            executor.execute(queueConsumerExecutor);
        }
    }

    // ...
}

QueueConsumerExecutor是实现了Runnable的消费任务,它有两个实现:

  • RegisterClientConsumerExecutor:客户端消费者任务
  • RegisterServerConsumerExecutor:服务端消费者任务

从名字也可以看出,RegisterClientConsumerExecutor负责处理客户端任务,shenyu客户端将元数据和URI写入disruptor后由这个消费者任务来消费数据,执行实际向注册中心注册的操作。RegisterServerConsumerExecutor负责处理服务端(shenyu-admin)任务,服务端从注册中心监听到元数据URI后写入disruptor,然后由RegisterServerConsumerExecutor任务来消费数据,处理数据入库操作和发布事件。
RegisterClientConsumerExecutor的消费逻辑:

public final class RegisterClientConsumerExecutor<T extends DataTypeParent> extends QueueConsumerExecutor<T> {
    
    private final Map<DataType, ExecutorTypeSubscriber<T>> subscribers;
    
    private RegisterClientConsumerExecutor(final Map<DataType, ExecutorTypeSubscriber<T>> executorSubscriberMap) {
        this.subscribers = new EnumMap<>(executorSubscriberMap);
    }

    @Override
    public void run() {
        // 获取数据
        final T data = getData();
        // 根据数据类型获取对应的处理器进行处理,即在disruptor启动的时候添加的订阅器
        subscribers.get(data.getType()).executor(Lists.newArrayList(data));
    }

    // ... 
}

根据不同的数据类型使用不同的订阅器执行器去执行,这些订阅器是在disruptor启动的时候设置的。目前注册的数据类型有3种,元数据URIAPI文档

public enum DataType {
    
    /**
     * Meta data data type enum.
     */
    META_DATA,
    
    /**
     * Uri data type enum.
     */
    URI,

    /**
     * Api doc type enum.
     */
    API_DOC,
}

所以相对应的订阅器也分为3类,分别处理元数据URI和API文档。在客户端和服务端分别有两个,所以一共是6个。
在这里插入图片描述
元数据处理

public class ShenyuClientMetadataExecutorSubscriber implements ExecutorTypeSubscriber<MetaDataRegisterDTO> {
    
    private final ShenyuClientRegisterRepository shenyuClientRegisterRepository;
    
    // ...

    /**
     * 遍历元数据,对数据注册到注册中心
     */
    @Override
    public void executor(final Collection<MetaDataRegisterDTO> metaDataRegisterDTOList) {
        for (MetaDataRegisterDTO metaDataRegisterDTO : metaDataRegisterDTOList) {
            // 调用响应注册中心的客户端注册类注册元数据
            shenyuClientRegisterRepository.persistInterface(metaDataRegisterDTO);
        }
    }
}

遍历数据,然后又将数据委托给ShenyuClientRegisterRepository执行。ShenyuClientRegisterRepository是在一开始读取配置的时候就创建了,是客户端注册类,用来将数据发送到注册中心的类,不同的注册方式有不同的实现类,该示例使用http方式注册(shenyu.register.registerType=http)的实现类是HttpClientRegisterRepositoryHttpClientRegisterRepository并没有直接实现ShenyuClientRegisterRepository接口,而是继承FailbackRegistryRepositoryFailbackRegistryRepository实现了ShenyuClientRegisterRepository接口,FailbackRegistryRepository本身主要用于对http注册过程中的失败重试。

@Override
public void persistInterface(final MetaDataRegisterDTO metadata) {
    try {
        this.doPersistInterface(metadata);
    } catch (Exception ex) {
        //If a failure occurs, it needs to be added to the retry list.
        // 如果注册失败,则添加到重试列表,过一段时间后重新注册
        logger.warn("Failed to persistInterface {}, cause:{}", metadata, ex.getMessage());
        this.addFailureMetaDataRegister(metadata);
    }
}

@Override
public void doPersistInterface(final MetaDataRegisterDTO metadata) {
    //  META_PATH = "/shenyu-client/register-metadata"
    doRegister(metadata, Constants.META_PATH, Constants.META_TYPE);
}

// 发送注册数据到admin
private <T> void doRegister(final T t, final String path, final String type) {
    int i = 0;
    // admin集群中的每个都要去注册(admin之间没有同步机制么?)
    for (String server : serverList) {
        i++;
        // 拼接上admin的地址和具体的接口路径,构成完整的请求地址
        String concat = server.concat(path);
        try {
            // 调用admin接口的token
            String accessToken = this.accessToken.get(server);
            if (StringUtils.isBlank(accessToken)) {
                throw new NullPointerException("accessToken is null");
            }
            // 通过工具类发送http请求
            RegisterUtils.doRegister(GsonUtils.getInstance().toJson(t), concat, type, accessToken);
            // considering the situation of multiple clusters, we should continue to execute here
        } catch (Exception e) {
            LOGGER.error("Register admin url :{} is fail, will retry. cause:{}", server, e.getMessage());
            if (i == serverList.size()) {
                throw new RuntimeException(e);
            }
        }
    }
}

http注册方式比较简单,遍历每个admin服务,获取到accessToken后,向/shenyu-client/register-metadata接口地址发起http请求,将数据发送给admin。
URI处理

public class ShenyuClientURIExecutorSubscriber implements ExecutorTypeSubscriber<URIRegisterDTO> {
 
    @Override
    public void executor(final Collection<URIRegisterDTO> dataList) {
        for (URIRegisterDTO uriRegisterDTO : dataList) {
            Stopwatch stopwatch = Stopwatch.createStarted();
            // 这里的逻辑是为了探测客户端是否已经启动
            while (true) {
                try (Socket ignored = new Socket(uriRegisterDTO.getHost(), uriRegisterDTO.getPort())) {
                    break;
                } catch (IOException e) {
                    long sleepTime = 1000;
                    // maybe the port is delay exposed
                    if (stopwatch.elapsed(TimeUnit.SECONDS) > 5) {
                        LOG.error("host:{}, port:{} connection failed, will retry",
                                uriRegisterDTO.getHost(), uriRegisterDTO.getPort());
                        // If the connection fails for a long time, Increase sleep time
                        if (stopwatch.elapsed(TimeUnit.SECONDS) > 180) {
                            sleepTime = 10000;
                        }
                    }
                    try {
                        TimeUnit.MILLISECONDS.sleep(sleepTime);
                    } catch (InterruptedException ex) {
                        LOG.error("interrupted when sleep", ex);
                    }
                }
            }
            ShenyuClientShutdownHook.delayOtherHooks();
            // 向注册中心注册URI数据
            shenyuClientRegisterRepository.persistURI(uriRegisterDTO);
            // 优雅停机
            ShutdownHookManager.get().addShutdownHook(new Thread(() -> {
                final URIRegisterDTO offlineDTO = new URIRegisterDTO();
                BeanUtils.copyProperties(uriRegisterDTO, offlineDTO);
                offlineDTO.setEventType(EventType.OFFLINE);
                shenyuClientRegisterRepository.offline(offlineDTO);
            }), 2);
        }
    }
}

URI注册逻辑基本相似,只是比元数据多了一步探测客户端时候已经启动完成的操作,保证客户端启动完成后再注册URI,后面的逻辑就跟元数据一样了。
分析到这里就将客户端的注册逻辑分析完了,通过读取自定义的注解信息构造元数据和URI,将数据发到Disruptor队列,然后从队列中消费数据,将消费者放到线程池中去执行,最终通过发送http请求到admin,元数据注册接口是/shenyu-client/register-metadataURI注册接口是/shenyu-client/register-uri

2、服务端注册流程

2.1、读取配置

从前面分析到,admin的两个注册接口分别是/shenyu-client/register-metadata/shenyu-client/register-uri,通过全局搜索发现这两个接口在ShenyuClientHttpRegistryController类。

@RequestMapping("/shenyu-client")
@Join
public class ShenyuClientHttpRegistryController implements ShenyuClientServerRegisterRepository {

    /**
     * 注册元数据
     */
    @PostMapping("/register-metadata")
    @ResponseBody
    public String registerMetadata(@RequestBody final MetaDataRegisterDTO metaDataRegisterDTO) {
        // 直接将元数据发布到Disruptor
        publisher.publish(metaDataRegisterDTO);
        return ShenyuResultMessage.SUCCESS;
    }
    
    /**
     * 注册URI
     */
    @PostMapping("/register-uri")
    @ResponseBody
    public String registerURI(@RequestBody final URIRegisterDTO uriRegisterDTO) {
        // 直接将URI发布到Disruptor
        publisher.publish(uriRegisterDTO);
        return ShenyuResultMessage.SUCCESS;
    }
}

但是这个类上并没有@RestController注解,那它就不能做为一个bean被spring扫描到,那它是如何创建的呢?
不着急,想要从注册中心中监听到数据(http注册方式可以将admin当作注册中心),自然需要有注册中心配置,来表明使用哪个注册中心。admin的注册中心配置是RegisterCenterConfiguration,我们先看这个配置类:

/**
 * 注册中心配置类
 */
@Configuration
public class RegisterCenterConfiguration {

    /**
     * 读取shenyu.register配置
     */
    @Bean
    @ConfigurationProperties(prefix = "shenyu.register")
    public ShenyuRegisterCenterConfig shenyuRegisterCenterConfig() {
        return new ShenyuRegisterCenterConfig();
    }
    
    /**
     * 创建用于服务端的注册类,从注册中心中监听数据,然后将数据写入Disruptor队列中
     */
    @Bean(destroyMethod = "close")
    public ShenyuClientServerRegisterRepository shenyuClientServerRegisterRepository(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig,
                                                                               final List<ShenyuClientRegisterService> shenyuClientRegisterService) {
        // 从配置中获取注册类型
        String registerType = shenyuRegisterCenterConfig.getRegisterType();
        // 根据注册类型通过SPI方式创建对应的ShenyuClientServerRegisterRepository
        ShenyuClientServerRegisterRepository registerRepository = ExtensionLoader.getExtensionLoader(ShenyuClientServerRegisterRepository.class).getJoin(registerType);
        // 创建Disruptor发布者
        RegisterClientServerDisruptorPublisher publisher = RegisterClientServerDisruptorPublisher.getInstance();
        // 每种客户端类型(rpc类型)的处理类
        Map<String, ShenyuClientRegisterService> registerServiceMap = shenyuClientRegisterService.stream().collect(Collectors.toMap(ShenyuClientRegisterService::rpcType, Function.identity()));
        // 启动Disruptor,添加元数据和URI的订阅器
        publisher.start(registerServiceMap);
        // 初始化注册中心
        registerRepository.init(publisher, shenyuRegisterCenterConfig);
        return registerRepository;
    }
}

该配置类创建了2个bean:

  • ShenyuRegisterCenterConfig:shenyu-admin注册中心配置,读取shenyu.register属性配置。
  • ShenyuClientServerRegisterRepository:服务端注册类,用于从注册中心中监听数据,然后将数据写入Disruptor队列中。

这里的创建Disruptor发布者,启动Disruptor等逻辑跟在客户端那边的一样,只是类是服务端这边的,就不再分析了。

2.1.1、用于监听的ShenyuClientServerRegisterRepository

上面生成的ShenyuClientServerRegisterRepository是用于实现服务端注册的接口,会根据注册中心的配置通过SPI方式创建注册类,每一个注册方式都对应一个实现类。
在这里插入图片描述
目前支持7种注册类型:

  • Http:ShenyuClientHttpRegistryController
  • Apollo:ApolloClientServerRegisterRepository
  • Zookeeper:ZookeeperClientServerRegisterRepository
  • Etcd:EtcdClientServerRegisterRepository
  • Nacos:NacosClientServerRegisterRepository
  • Consul:ConsulClientServerRegisterRepository
  • Polaris:PolarisClientServerRegisterRepository
    加载类型通过registerType指定,也就是我们在配置文件中指定的类型:
shenyu:
  register:
    registerType: http

服务端的注册类型必须跟客户端的注册类型一致,这样服务端才可以监听到注册信息。这里要指定的是http,所以这里创建的就是ShenyuClientHttpRegistryController,这个不就是前面说到的注册接口所在的类么,所以回到前面的ShenyuClientHttpRegistryController

2.2、注册元数据和URI

2.2.1、注册接口接收数据写入Disruptor

@RequestMapping("/shenyu-client")
@Join
public class ShenyuClientHttpRegistryController implements ShenyuClientServerRegisterRepository {

    /**
     * 注册元数据
     */
    @PostMapping("/register-metadata")
    @ResponseBody
    public String registerMetadata(@RequestBody final MetaDataRegisterDTO metaDataRegisterDTO) {
        // 直接将元数据发布到Disruptor
        publisher.publish(metaDataRegisterDTO);
        return ShenyuResultMessage.SUCCESS;
    }
    
    /**
     * 注册URI
     */
    @PostMapping("/register-uri")
    @ResponseBody
    public String registerURI(@RequestBody final URIRegisterDTO uriRegisterDTO) {
        // 直接将URI发布到Disruptor
        publisher.publish(uriRegisterDTO);
        return ShenyuResultMessage.SUCCESS;
    }
}

两个注册接口获取到数据后,就直接调用了publisher.publish()方法,把数据发布到Disruptor队列中。

2.2.2、Disruptor消费数据并持久化

QueueConsumer实现了WorkHandler接口,是Disruptor的消费者,消费逻辑就在它的onEvent方法中:

public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {
    
    private final OrderlyExecutor executor;
    
    private final QueueConsumerFactory<T> factory;
    
    /**
     * Instantiates a new Queue consumer.
     *
     * @param executor the executor
     * @param factory  the factory
     */
    public QueueConsumer(final OrderlyExecutor executor, final QueueConsumerFactory<T> factory) {
        this.executor = executor;
        this.factory = factory;
    }
    
    @Override
    public void onEvent(final DataEvent<T> t) {
        if (Objects.nonNull(t)) {
            // 根据事件类型使用不同的线程池
            ThreadPoolExecutor executor = orderly(t);
            // 通过工厂创建队列消费任务 RegisterServerConsumerExecutor
            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();
            // 为消费任务设置数据
            queueConsumerExecutor.setData(t.getData());
            t.setData(null);
            // 放在线程池中执行 消费任务
            executor.execute(queueConsumerExecutor);
        }
    }

    // ...
}

分析客户端注册流程的时候说到RegisterServerConsumerExecutor是服务端消费者任务,处理数据入库操作和发布事件。
RegisterServerConsumerExecutor消费逻辑:

public final class RegisterServerConsumerExecutor extends QueueConsumerExecutor<Collection<DataTypeParent>> {

    // 每种数据类型的订阅器执行器
    private final Map<DataType, ExecutorSubscriber<DataTypeParent>> subscribers;
    
    private RegisterServerConsumerExecutor(final Map<DataType, ExecutorTypeSubscriber<DataTypeParent>> executorSubscriberMap) {
        this.subscribers = new HashMap<>(executorSubscriberMap);
    }
    
    @Override
    public void run() {
        Collection<DataTypeParent> results = getData()
                .stream()
                .filter(this::isValidData)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(results)) {
            return;
        }
        // 选择对应的数据类型的订阅器执行器去执行
        selectExecutor(results).executor(results);
    }
    
    private ExecutorSubscriber<DataTypeParent> selectExecutor(final Collection<DataTypeParent> list) {
        final Optional<DataTypeParent> first = list.stream().findFirst();
        return subscribers.get(first.orElseThrow(() -> new RuntimeException("the data type is not found")).getType());
    }
    
    // ...
}

根据不同的数据类型使用不同的订阅器执行器去执行,这些订阅器是在disruptor启动的时候设置的。
服务端的订阅器有3个,分别为MetadataExecutorSubscriberURIRegisterExecutorSubscriberApiDocExecutorSubscriber,分别处理元数据URIAPI文档

元数据的处理

public class MetadataExecutorSubscriber implements ExecutorTypeSubscriber<MetaDataRegisterDTO> {

    // 每种客户端类型的注册服务
    private final Map<String, ShenyuClientRegisterService> shenyuClientRegisterService;

    public MetadataExecutorSubscriber(final Map<String, ShenyuClientRegisterService> shenyuClientRegisterService) {
        this.shenyuClientRegisterService = shenyuClientRegisterService;
    }

    @Override
    public DataType getType() {
        return DataType.META_DATA;
    }

    @Override
    public void executor(final Collection<MetaDataRegisterDTO> metaDataRegisterDTOList) {
        // 遍历元数据
        metaDataRegisterDTOList.forEach(meta -> {
            // 根据客户端类型
            Optional.ofNullable(this.shenyuClientRegisterService.get(meta.getRpcType()))
                    .ifPresent(shenyuClientRegisterService -> {
                        // 加锁,保证数据顺序执行,防止并发
                        synchronized (shenyuClientRegisterService) {
                            // 处理数据
                            shenyuClientRegisterService.register(meta);
                        }
                    });
        });
    }
}

ShenyuClientRegisterService是注册方法接口,它有多个实现类:
在这里插入图片描述

  • AbstractContextPathRegisterService:抽象类,处理部分公共逻辑;
  • AbstractShenyuClientRegisterServiceImpl::抽象类,处理部分公共逻辑;
  • ShenyuClientRegisterDivideServiceImpl:divide类,处理http注册类型;
  • ShenyuClientRegisterDubboServiceImpl:dubbo类,处理dubbo注册类型;
  • ShenyuClientRegisterGrpcServiceImpl:gRPC类,处理gRPC注册类型;
  • ShenyuClientRegisterBrpcServiceImpl:bRPC类,处理bRPC注册类型;
  • ShenyuClientRegisterMotanServiceImpl:Motan类,处理Motan注册类型;
  • ShenyuClientRegisterSofaServiceImpl:Sofa类,处理Sofa注册类型;
  • ShenyuClientRegisterSpringCloudServiceImpl:SpringCloud类,处理SpringCloud注册类型;
  • ShenyuClientRegisterTarsServiceImpl:Tars类,处理Tars注册类型;
  • ShenyuClientRegisterWebSocketServiceImpl:Websocket类,处理Websocket注册类型;

每一种rpc类型都对应一个注册处理类,所以本文是使用ShenyuClientRegisterDivideServiceImpl来处理。

public abstract class AbstractShenyuClientRegisterServiceImpl extends FallbackShenyuClientRegisterService implements ShenyuClientRegisterService {

    @Resource
    private ApplicationEventPublisher eventPublisher;

    // 这几个就是操作数据库的service
    @Resource
    private SelectorService selectorService;

    @Resource
    private MetaDataService metaDataService;

    @Resource
    private RuleService ruleService;

    @Override
    public String register(final MetaDataRegisterDTO dto) {
        
        // 1、注册选择器(可以认为一个服务就是一个选择器)
        
        // 选择器执行逻辑,默认情况是空的,需要在控制台另外手动配置
        // 子类实现
        String selectorHandler = selectorHandler(dto);
        // 持久化选择器并发布选择器变更事件(不存在的时候)ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE
        String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);
        
        // 2、注册规则(可以认为一个元数据就是一个规则,根据path判断是否同一个)
        
        // 规则处理逻辑
        // 子类实现,,都是直接创建一个各自rpc类型的默认逻辑
        String ruleHandler = ruleHandler();
        // 构建规则DTO
        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);
        // 持久化规则并发布规则变更事件(不存在的时候)ConfigGroupEnum.RULE, DataEventTypeEnum.UPDATE
        ruleService.registerDefault(ruleDTO);
        
        // 3、注册元数据,并发布元数据变更事件(已存在,发布元数据更新事件,不存在,发布元数据创建事件)
       
        // 子类实现
        registerMetadata(dto);
        
        // 4、注册contextPath(只有http,springCloud,webSocket类型才有)
        String contextPath = dto.getContextPath();
        if (StringUtils.isNotEmpty(contextPath)) {
            registerContextPath(dto);
        }
        return ShenyuResultMessage.SUCCESS;
    }

}

整个注册处理逻辑可以分为4步:

  1. 注册选择器,构建选择器,默认情况下一个服务就是一个选择器。之后将选择器插入数据库并发布选择器变更事件。
@Override
public String registerDefault(final MetaDataRegisterDTO dto, final String pluginName, final String selectorHandler) {
    // 以contextPath或appName作为选择器名称
    String contextPath = ContextPathUtils.buildContextPath(dto.getContextPath(), dto.getAppName());
    // 根据选择器名和插件名从数据库中查询选择器
    SelectorDO selectorDO = findByNameAndPluginName(contextPath, pluginName);
    // 如果还不存在,就创建一个选择器插入数据库
    if (Objects.isNull(selectorDO)) {
        // 构建选择器DTO
        SelectorDTO selectorDTO = SelectorUtil.buildSelectorDTO(contextPath, pluginMapper.selectByName(pluginName).getId());
        selectorDTO.setHandle(selectorHandler);
        // 注册选择器并发布事件 ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE
        return registerDefault(selectorDTO);
    }
    return selectorDO.getId();
}
  1. 注册规则,可以认为一个元数据就是一个规则,根据path判断是否同一个。
    构建规则:
private RuleDTO buildRpcDefaultRuleDTO(final String selectorId, final MetaDataRegisterDTO metaDataDTO, final String ruleHandler) {
    return buildRuleDTO(selectorId, ruleHandler, metaDataDTO.getRuleName(), metaDataDTO.getPath());
}

private RuleDTO buildRuleDTO(final String selectorId, final String ruleHandler, final String ruleName, final String path) {
    // 构建规则DTO
    RuleDTO ruleDTO = RuleDTO.builder()
            .selectorId(selectorId)
            .name(ruleName)
            .matchMode(MatchModeEnum.AND.getCode())
            .enabled(Boolean.TRUE)
            .loged(Boolean.TRUE)
            .matchRestful(Boolean.FALSE)
            .sort(1)
            .handle(ruleHandler)
            .build();

    // 将{xxx}替换成**
    String conditionPath = this.rewritePath(path);
    RuleConditionDTO ruleConditionDTO = RuleConditionDTO.builder()
            .paramType(ParamTypeEnum.URI.getName())
            .paramName("/")
            .paramValue(conditionPath)
            .build();
    // 设置规则条件
    if (conditionPath.endsWith(AdminConstants.URI_SLASH_SUFFIX)) {
        ruleConditionDTO.setOperator(OperatorEnum.STARTS_WITH.getAlias());
    } else if (conditionPath.endsWith(AdminConstants.URI_SUFFIX)) {
        ruleConditionDTO.setOperator(OperatorEnum.PATH_PATTERN.getAlias());
    } else if (conditionPath.indexOf("*") > 1) {
        ruleConditionDTO.setOperator(OperatorEnum.MATCH.getAlias());
    } else {
        ruleConditionDTO.setOperator(OperatorEnum.EQ.getAlias());
    }
    ruleDTO.setRuleConditions(Collections.singletonList(ruleConditionDTO));
    return ruleDTO;
}

保存规则:

@Override
public String registerDefault(final RuleDTO ruleDTO) {
    // 选择器下已经存在同名的规则,则直接返回,什么也不干
    if (Objects.nonNull(ruleMapper.findBySelectorIdAndName(ruleDTO.getSelectorId(), ruleDTO.getName()))) {
        return "";
    }
    RuleDO ruleDO = RuleDO.buildRuleDO(ruleDTO);
    if (StringUtils.isEmpty(ruleDTO.getId())) {
        // 插入规则
        ruleMapper.insertSelective(ruleDO);
        // 插入规则条件
        addCondition(ruleDO, ruleDTO.getRuleConditions());
    }
    // 发布规则变更事件 ConfigGroupEnum.RULE, DataEventTypeEnum.UPDATE
    ruleEventPublisher.onRegister(ruleDO, ruleDTO.getRuleConditions());
    return ruleDO.getId();
}

具体的规则设计建议去看官方文档。
3. 注册元数据,直接将注册上来的元数据保存

@Override
protected void registerMetadata(final MetaDataRegisterDTO dto) {
    if (dto.isRegisterMetaData()) {
        MetaDataService metaDataService = getMetaDataService();
        // 根据路径查询元数据时候已存在
        MetaDataDO exist = metaDataService.findByPath(dto.getPath());
        // 已存在,就更新,发布元数据更新事件,不存在,就插入,发布元数据创建事件
        metaDataService.saveOrUpdateMetaData(exist, dto);
    }
}

4、注册ContextPath,只有httpspringCloudwebSocket类型才有。处理的逻辑在AbstractContextPathRegisterService中。

public abstract class AbstractContextPathRegisterService extends AbstractShenyuClientRegisterServiceImpl {
    
    @Override
    public void registerContextPath(final MetaDataRegisterDTO dto) {
        // 持久化contextPath插件下的选择器并发布选择器变更事件
        String contextPathSelectorId = getSelectorService().registerDefault(dto, PluginEnum.CONTEXT_PATH.getName(), "");
        // 创建规则处理逻辑
        ContextMappingRuleHandle handle = new ContextMappingRuleHandle();
        handle.setContextPath(PathUtils.decoratorContextPath(dto.getContextPath()));
        handle.setAddPrefixed(dto.getAddPrefixed());
        // 注册contextPath插件默认的规则,contextPath就是规则名,并发布规则变更事件
        getRuleService().registerDefault(buildContextPathDefaultRuleDTO(contextPathSelectorId, dto, handle.toJson()));
    }
}

● URI的处理
URI数据是由URIRegisterExecutorSubscriber订阅器处理:

public class URIRegisterExecutorSubscriber implements ExecutorTypeSubscriber<URIRegisterDTO> {
    
    @Override
    public void executor(final Collection<URIRegisterDTO> dataList) {
        if (CollectionUtils.isEmpty(dataList)) {
            return;
        }
        // 根据rpc类型分类
        final Map<String, List<URIRegisterDTO>> groupByRpcType = dataList.stream()
                .filter(data -> StringUtils.isNotBlank(data.getRpcType()))
                .collect(Collectors.groupingBy(URIRegisterDTO::getRpcType));
        for (Map.Entry<String, List<URIRegisterDTO>> entry : groupByRpcType.entrySet()) {
            // 根据不同rpc类型使用对应的shenyuClientRegisterService处理
            final String rpcType = entry.getKey();
            Optional.ofNullable(shenyuClientRegisterService.get(rpcType))
                    .ifPresent(service -> {
                        final List<URIRegisterDTO> list = entry.getValue();
                        // 再以contextPath/appName分类
                        Map<String, List<URIRegisterDTO>> listMap = buildData(list);
                        listMap.forEach((selectorName, uriList) -> {
                            final List<URIRegisterDTO> register = new LinkedList<>();
                            final List<URIRegisterDTO> offline = new LinkedList<>();
                            for (URIRegisterDTO d : uriList) {
                                final EventType eventType = d.getEventType();
                                // 判断是注册类型还是下线类型
                                if (Objects.isNull(eventType) || EventType.REGISTER.equals(eventType)) {
                                    // eventType is null, should be old versions
                                    register.add(d);
                                } else if (EventType.OFFLINE.equals(eventType)) {
                                    offline.add(d);
                                }
                            }
                            if (CollectionUtils.isNotEmpty(register)) {
                                // 注册URI
                                service.registerURI(selectorName, register);
                            }
                            if (CollectionUtils.isNotEmpty(offline)) {
                                // 下线URI
                                service.offline(selectorName, offline);
                            }
                        });
                    });
        }
    }
    
    private Map<String, List<URIRegisterDTO>> buildData(final Collection<URIRegisterDTO> dataList) {
        Map<String, List<URIRegisterDTO>> resultMap = new HashMap<>(8);
        for (URIRegisterDTO dto : dataList) {
            String contextPath = dto.getContextPath();
            String key = StringUtils.isNotEmpty(contextPath) ? contextPath : dto.getAppName();
            if (StringUtils.isNotEmpty(key)) {
                if (resultMap.containsKey(key)) {
                    List<URIRegisterDTO> existList = resultMap.get(key);
                    existList.add(dto);
                    resultMap.put(key, existList);
                } else {
                    resultMap.put(key, Lists.newArrayList(dto));
                }
            }
        }
        return resultMap;
    }
}

调到FallbackShenyuClientRegisterServiceregisterURI()方法

@Override
public String registerURI(final String selectorName, final List<URIRegisterDTO> uriList) {
    String result;
    String key = key(selectorName);
    try {
        this.removeFallBack(key);
        result = this.doRegisterURI(selectorName, uriList);
        logger.info("Register success: {},{}", selectorName, uriList);
    } catch (Exception ex) {
        logger.warn("Register exception: cause:{}", ex.getMessage());
        result = "";
        this.addFallback(key, new FallbackHolder(selectorName, uriList));
    }
    return result;
}

FallbackShenyuClientRegisterService是用来异常处理的,然后调用doRegisterURI()做真正处理。

@Override
public String doRegisterURI(final String selectorName, final List<URIRegisterDTO> uriList) {
    if (CollectionUtils.isEmpty(uriList)) {
        return "";
    }
    // 查询对应的选择器
    SelectorDO selectorDO = selectorService.findByNameAndPluginName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));
    if (Objects.isNull(selectorDO)) {
        throw new ShenyuException("doRegister Failed to execute,wait to retry.");
    }
  
    // 过滤port或host为空的URI
    List<URIRegisterDTO> validUriList = uriList.stream().filter(dto -> Objects.nonNull(dto.getPort()) && StringUtils.isNotBlank(dto.getHost())).collect(Collectors.toList());
    // 由URI构建处理选择器中的handler信息,更新选择器中的handler
    // 应该就是相当于添加上服务实例信息
    String handler = buildHandle(validUriList, selectorDO);
    if (handler != null) {
        selectorDO.setHandle(handler);
        SelectorData selectorData = selectorService.buildByName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));
        selectorData.setHandle(handler);
        // 更新数据库
        selectorService.updateSelective(selectorDO);
        // 发布选择器变更事件
        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE, Collections.singletonList(selectorData)));
    }
    return ShenyuResultMessage.SUCCESS;
}

总结就是admin拿到URI数据后,更新选择器中的handler信息,然后写入到数据库,最后发布事件。
更新的就是这里的信息:
在这里插入图片描述
至此,服务端注册流程也就分析完了,主要通过对外提供的接口,接受客户端的注册信息,然后写入到Disruptor队列,再从中消费数据,根据接收到的元数据和URI数据更新admin的选择器、规则、元数据和选择器的handler。

参考资料:
官方博客

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1311747.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

Java获取当前用户当前工作目录

方法一&#xff1a;使用System.getProperty(“user.dir”)函数可以获取用户当前工作目录 例如&#xff0c;Java工程的文件布局如下&#xff1a; 主类文件&#xff0c;获取用户当前的工作目录&#xff1a; package com.thb;public class Test5 {public static void main(Stri…

RNN介绍及Pytorch源码解析

介绍一下RNN模型的结构以及源码&#xff0c;用作自己复习的材料。 RNN模型所对应的源码在&#xff1a;\PyTorch\Lib\site-packages\torch\nn\modules\RNN.py文件中。 RNN的模型图如下&#xff1a; 源码注释中写道&#xff0c;RNN的数学公式&#xff1a; 表示在时刻的隐藏状态…

KubeSphere应用【笔记四】自定义镜像

一、概述 在KubeSphere部署Redis负载时&#xff0c;想通过应用商店部署Redis&#xff0c;通过应用商店部署redis时可以指定访问密码&#xff0c;结果应用商店部署Redis时如下图所示&#xff0c;不能进行部署&#xff0c;所以打算自己制作有默认密码的镜像&#xff0c;上传至Ha…

C语言--有一个3*4的矩阵,求出其中最大值的那个元素的值,以及其所在的行号和列号

一.题目描述 有一个3*4的矩阵&#xff0c;要求求出其中最大值的那个元素的值&#xff0c;以及其所在的行号和列号 比如&#xff1a;给定一个3*4的矩阵如下 输出结果&#xff1a;最大值为 12 &#xff0c;行号为3&#xff0c; 列号为2 二.思路分析 打擂台算法&#xff1a; 先思考…

爬虫的分类

爬虫的分类 网络爬虫按照系统结构和实现技术&#xff0c;大致可分为4类&#xff0c;即通用网络爬虫、聚焦网络爬虫、增量网络爬虫和深层次网络爬虫。 1.通用网络爬虫&#xff1a;搜索引擎的爬虫 比如用户在百度搜索引擎上检索对应关键词时&#xff0c;百度将对关键词进行分析…

【深度学习目标检测】五、基于深度学习的安全帽识别(python,目标检测)

深度学习目标检测方法则是利用深度神经网络模型进行目标检测&#xff0c;主要有以下几种&#xff1a; R-CNN系列&#xff1a;包括R-CNN、Fast R-CNN、Faster R-CNN等&#xff0c;通过候选区域法生成候选目标区域&#xff0c;然后使用卷积神经网络提取特征&#xff0c;并通过分类…

Linux 链接器如何使用静态库来解析引用

文章目录 通过进行代码实践静态库在编译时被链接到可执行文件中的基本原理原理总结 QA:.obj文件是什么? 通过进行代码实践 链接器在解析引用时&#xff0c;可以使用静态库来满足对未定义符号的引用。以下是使用静态库的一般步骤&#xff1a; 编写代码&#xff1a; 首先&#…

大数据技术11:Hadoop 原理与运行机制

前言&#xff1a;HDFS &#xff08;Hadoop Distributed File System&#xff09;是 Hadoop 下的分布式文件系统&#xff0c;具有高容错、高吞吐量等特性&#xff0c;可以部署在低成本的硬件上。 一、Hadoop简介 1.1、Hadoop定义 Hadoop 作为一个开源分布式系统基础框架&#x…

gitlab下载安装

1.下载 官网rpm包 gitlab/gitlab-ce - Results in gitlab/gitlab-ce 国内镜像 Index of /gitlab-ce/yum/ | 清华大学开源软件镜像站 | Tsinghua Open Source Mirror 2.安装 rpm -ivh gitlab-ce-16.4.3-ce.0.el7.x86_64.rpm 3.配置 vim /etc/gitlab/gitlab.rb 将 externa…

第一个“hello Android”程序

1、首先安装Android studio&#xff08;跳过&#xff09; Android Studio是由Google推出的官方集成开发环境&#xff08;IDE&#xff09;&#xff0c;专门用于Android应用程序的开发。它是基于JetBrains的IntelliJ IDEA IDE构建的&#xff0c;提供了丰富的功能和工具&#xff0…

想要在电脑桌面上使用手机便签怎么操作?

作为一名上班族&#xff0c;我们时常需要在电脑和手机之间同步使用便签&#xff0c;以记录工作、生活中的重要事项。然而&#xff0c;有些时候我们可能更习惯在手机上使用便签&#xff0c;但又希望在电脑桌面上也能够方便地查看和编辑这些便签。那么&#xff0c;如何在电脑桌面…

TCP/IP详解——ICMP协议,Ping程序,Traceroute程序,IP源站选路选项

文章目录 一、ICMP 协议1. ICMP 概念2. ICMP 重定向3. ICMP 差错检测4. ICMP 错误报告/差错报文5. ICMP 差错报文的结构6. ICMP 源站抑制差错7. ICMP 数据包格式8. ICMP 消息类型和编码类型9. ICMP 应用-Ping10. ICMP 应用-Tracert11. BSD 对 ICMP 报文的处理12. 总结 PING 程序…

开源IPad Pro应用IDE:使用SSH远程连接服务器进行云端编程开发

&#x1f525;博客主页&#xff1a; 小羊失眠啦. &#x1f3a5;系列专栏&#xff1a;《C语言》 《数据结构》 《Linux》《Cpolar》 ❤️感谢大家点赞&#x1f44d;收藏⭐评论✍️ 前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;…

IDEA中Terminal配置为bash

简介 我们日常命令行都是使用Linux的bash指令&#xff0c;但是我们的开发基本都是基于Windows上的IDEA进行开发的&#xff0c;对此我们可以通过将IDEA将终端Terminal改为git bash自带的bash.exe解决问题。 配置步骤 安装GIT 这步无需多说了&#xff0c;读者可自行到官网下载…

PyTorch 的 10 条内部用法

欢迎阅读这份有关 PyTorch 原理的简明指南[1]。无论您是初学者还是有一定经验&#xff0c;了解这些原则都可以让您的旅程更加顺利。让我们开始吧&#xff01; 1. 张量&#xff1a;构建模块 PyTorch 中的张量是多维数组。它们与 NumPy 的 ndarray 类似&#xff0c;但可以在 GPU …

【方法】如何给PDF文件添加“限制密码”?

PDF是很多人工作中经常用到的电子文档格式&#xff0c;它可以保留原始文档的所有格式和布局&#xff0c;也不容易修改&#xff0c;想要编辑修改PDF文件需要用到编辑器。 但如果给PDF文件添加“限制密码”&#xff0c;就可以保护文件不被随意修改&#xff0c;即使使用编辑器也需…

基于PCIe的NVMe学习

一&#xff1a;基本概念 1.UltraScale&#xff1a;是Xilinx ZYNQ 系列产品 2.spec&#xff1a;大家现在别纠结于具体的命令&#xff0c;了解一下就好。老板交代干活的时候&#xff0c;再找spec一个一个看吧————猜测估计是命令表之类的。 3.TLP报文部分&#xff1a; 二&…

STM32--中断使用(超详细!)

写在前面&#xff1a;前面的学习中&#xff0c;我们接触了STM32的第一个外设GPIO&#xff0c;这也是最常用的一个外设&#xff1b;而除了GPIO外&#xff0c;中断也是一个十分重要且常用的外设&#xff1b;只有掌握了中断&#xff0c;再处理程序时才能掌握好解决实际问题的逻辑思…

云上丝绸之路| 云轴科技ZStack成功实践精选(西北)

古有“丝绸之路” 今有丝绸之路经济带 丝路焕发新生&#xff0c;数智助力经济 云轴科技ZStack用“云”护航千行百业 沿丝绸之路&#xff0c;领略西北数字化。 古丝绸之路起点-陕西 集历史与现代交融&#xff0c;不仅拥有悠久的历史文化积淀&#xff0c;而且现代化、数字化发…

论文润色降重哪个平台好 papergpt

大家好&#xff0c;今天来聊聊论文润色降重哪个平台好&#xff0c;希望能给大家提供一点参考。 以下是针对论文重复率高的情况&#xff0c;提供一些修改建议和技巧&#xff1a; 标题&#xff1a;论文润色降重哪个平台好――专业、高效、可靠的学术支持 一、引言 在学术研究中&…