Apache ShenYu 入门

介绍

最近在了解网关相关的东西,发现了shenyu这款异步的、高性能的、跨语言的、响应式的 API 网关。 花了一两天时间做了一个入门体验,在此记录一下。

具体介绍大家可以到官网去查阅

本地运行

在本地启动之前需要先了解两个模块:

1-shenyu-admin : 插件和其他信息配置的管理后台,启动类如下

ShenyuAdminBootstrap

2-shenyu-bootstrap : 用于启动项目,启动类如下

ShenyuBootstrapApplication

3-在启动之前需要先配置一下db的信息,这里我选择mysql,修改shenyu-admin下面的application.yml中的内容如下，然后配置application-mysql.yml中的连接信息即可。

spring:
  profiles:
    active: mysql

4-最后初始化一下SQL脚本:

incubator-shenyu/db/init/mysql/schema.sql

5-运行两个启动类 访问地址: http://localhost:9095/#/home 用户名密码 admin 123456

到这里整个网关服务就在本地启动了,但是此时还没有我们自己的服务接入进来。

服务接入入门

我们可以直接在shenyu-examples中找到想接入的服务端demo。比如http,dubbo,motan,springmvc,springcloud等等。

这里我选择了shenyu-examples-http来进行测试,其实就是一个springboot项目. 因为我们最终是需要通过网关访问,需要让网关感知到,因此需要先做一些配置(example中已经配置好,可以选择修改,这里我修改了一下contextPath和appName)

application.yml:
shenyu:
  register:
    registerType: http #zookeeper #etcd #nacos #consul
    serverLists: http://localhost:9095 #localhost:2181 #http://localhost:2379 #localhost:8848
    props:
      username: admin
      password: 123456
  client:
      http:
        props:
          contextPath: /api/test
          appName: testApp
          port: 8189

上面主要是进行配置我们启动的服务如何注册到网关: registerType代表类型包括http,zk,nacos等,这里默认是http。 client中则配置了当前服务在网关中的一些标识。 然后就可以将应用启动起来了。接下来可以使用postman进行调用测试.可以从demo中的HttpTestController选一个接口分别直接访问和通过网关进行访问来测试。

直接访问当前应用 http://localhost:8189/test/payment:

基于网关来访问 http://localhost:9195/api/test/test/payment:

通过访问地址就可以指定上面配置的contextPath作用的什么了。如果基于网关的访问的路径前缀不是我们配置的contextPath则会提示如下错: "message": "divide:Can not find selector, please check your configuration!"

应用接入的原理浅析

上面做了最基础的入门之后,不禁想探究一下其背后的原理。于是在接入的http example的pom文件中发现其引入了一个starter（springboot中的starter就不介绍了） 官方介绍地址:shenyu.apache.org/zh/docs/des…

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

然后找到这个starter模块,发现shenyu还提供了其他starter,比如dubbo的,motan的,可以让这些RPC框架接入我们的网关中。

我们这里还是继续看shenyu-spring-boot-starter-client-springmvc。在ShenyuSpringMvcClientConfiguration中定义了多个bean,主要看

SpringMvcClientBeanPostProcessor
实现了BeanPostProcessor接口,在bean实例化、依赖注入、初始化完毕时执行会调用postProcessAfterInitialization方法。具体源码如下:

@Override
public Object postProcessAfterInitialization(@NonNull final Object bean, @NonNull final String beanName) throws BeansException {
    // Filter out is not controller out
    if (Boolean.TRUE.equals(isFull) || !hasAnnotation(bean.getClass(), Controller.class)) {
        return bean;
    }
    //获取路径,先获取ShenyuSpringMvcClient注解上的，如果没有则获取RequestMapping上的
    final ShenyuSpringMvcClient beanShenyuClient = AnnotationUtils.findAnnotation(bean.getClass(), ShenyuSpringMvcClient.class);
    final String superPath = buildApiSuperPath(bean.getClass());
    // Compatible with previous versions
    if (Objects.nonNull(beanShenyuClient) && superPath.contains("*")) {
        publisher.publishEvent(buildMetaDataDTO(beanShenyuClient, pathJoin(contextPath, superPath)));
        return bean;
    }
    
    //获取方法上面先获取ShenyuSpringMvcClient注解,解析path
    final Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(bean.getClass());
    for (Method method : methods) {
        final RequestMapping requestMapping = AnnotatedElementUtils.findMergedAnnotation(method, RequestMapping.class);
        ShenyuSpringMvcClient methodShenyuClient = AnnotationUtils.findAnnotation(method, ShenyuSpringMvcClient.class);
        methodShenyuClient = Objects.isNull(methodShenyuClient) ? beanShenyuClient : methodShenyuClient;
        // the result of ReflectionUtils#getUniqueDeclaredMethods contains method such as hashCode, wait, toSting
        // add Objects.nonNull(requestMapping) to make sure not register wrong method
        //
        if (Objects.nonNull(methodShenyuClient) && Objects.nonNull(requestMapping)) {
            publisher.publishEvent(buildMetaDataDTO(methodShenyuClient, buildApiPath(method, superPath)));
        }
    }
    
    return bean;
}

//上面最终是将解析的注解构建为MetaDataRegisterDTO,并通过publisher.publishEvent发送出去

private MetaDataRegisterDTO buildMetaDataDTO(@NonNull final ShenyuSpringMvcClient shenyuSpringMvcClient, final String path) {
    return MetaDataRegisterDTO.builder()
            .contextPath(contextPath)  //yml配置的
            .appName(appName)          //yml配置的
            .path(path)
            .pathDesc(shenyuSpringMvcClient.desc())
            .rpcType(RpcTypeEnum.HTTP.getName())
            .enabled(shenyuSpringMvcClient.enabled())
            .ruleName(StringUtils.defaultIfBlank(shenyuSpringMvcClient.ruleName(), path))
            .registerMetaData(shenyuSpringMvcClient.registerMetaData())
            .build();
}

ShenyuClientRegisterEventPublisher
上面的publisher.publishEvent就是指ShenyuClientRegisterEventPublisher。
他是基于Disruptor高性能队列来实现的一个生产消费的模型。
提供了publishEvent方法来生产消息
并且提供了QueueConsumer来进行异步消费
最终会由RegisterClientConsumerExecutor来进行消费

private final ShenyuClientRegisterEventPublisher publisher = ShenyuClientRegisterEventPublisher.getInstance();


//启动方法,指定了ShenyuClientMetadataExecutorSubscriber和ShenyuClientURIExecutorSubscriber,在RegisterClientConsumerExecutor消费的时候会使用.
public void start(final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {
    RegisterClientExecutorFactory factory = new RegisterClientExecutorFactory();
    factory.addSubscribers(new ShenyuClientMetadataExecutorSubscriber(shenyuClientRegisterRepository));
    factory.addSubscribers(new ShenyuClientURIExecutorSubscriber(shenyuClientRegisterRepository));
    providerManage = new DisruptorProviderManage<>(factory);
    providerManage.startup();
}

//ShenyuClientMetadataExecutorSubscriber的内容就是去向shenyu-admin注册Metadata了。

//ShenyuClientRegisterRepository 就是在starter中定义的bean,下面有介绍,总之我们example获取到的是HttpClientRegisterRepository

private final ShenyuClientRegisterRepository shenyuClientRegisterRepository;

/**
 * Instantiates a new shenyu client metadata executor subscriber.
 *
 * @param shenyuClientRegisterRepository the shenyu client register repository
 */
public ShenyuClientMetadataExecutorSubscriber(finalShenyuClientRegisterRepository shenyuClientRegisterRepository) {
    this.shenyuClientRegisterRepository = shenyuClientRegisterRepository;
}

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

//消息类型是DataType.META_DATA的,消费者最终会调用此方法处理消息
@Override
public void executor(final Collection<MetaDataRegisterDTO> metaDataRegisterDTOList) {
    for (MetaDataRegisterDTO metaDataRegisterDTO : metaDataRegisterDTOList) {
        shenyuClientRegisterRepository.persistInterface(metaDataRegisterDTO);
    }
}

//具体的实现就是基于http请求将metaData注册到了admin中
@Override
public void doPersistInterface(final MetaDataRegisterDTO metadata) {
    doRegister(metadata, Constants.META_PATH, Constants.META_TYPE);
}

接口地址:
String META_PATH = "/shenyu-client/register-metadata";

我们可以在shenyu-admin中的ShenyuClientHttpRegistryController中找到对应的地址。
shenyu-admin如何接收新的信息变动在后面会继续说明。这里先了解.

ContextRegisterListener
在启动的时候往publisher中生产URIRegisterDTO类型的消息

@Override
public void onApplicationEvent(@NonNull final ContextRefreshedEvent contextRefreshedEvent) {
    if (!registered.compareAndSet(false, true)) {
        return;
    }
    if (Boolean.TRUE.equals(isFull)) {
        publisher.publishEvent(buildMetaDataDTO());
    }
    try {
        final int mergedPort = port <= 0 ? PortUtils.findPort(beanFactory) : port;
        publisher.publishEvent(buildURIRegisterDTO(mergedPort));
    } catch (ShenyuException e) {
        throw new ShenyuException(e.getMessage() + "please config ${shenyu.client.http.props.port} in xml/yml !");
    }
}

private URIRegisterDTO buildURIRegisterDTO(final int port) {
    return URIRegisterDTO.builder()
        .contextPath(this.contextPath)
        .appName(appName)
        .protocol(protocol)
        .host(IpUtils.isCompleteHost(this.host) ? this.host : IpUtils.getHost(this.host))
        .port(port)
        .rpcType(RpcTypeEnum.HTTP.getName())
        .build();
}

ShenyuClientRegisterRepository
根据配置来获取具体的实现,默认是http

/**
 * New instance shenyu client register repository.
 *
 * @param shenyuRegisterCenterConfig the shenyu register center config
 * @return the shenyu client register repository
 */
public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {
    if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {
        //spi机制获取具体实现,我们的demo中是HttpClientRegisterRepository
        ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());
        result.init(shenyuRegisterCenterConfig);
        ShenyuClientShutdownHook.set(result, shenyuRegisterCenterConfig.getProps());
        REPOSITORY_MAP.put(shenyuRegisterCenterConfig.getRegisterType(), result);
        return result;
    }
    return REPOSITORY_MAP.get(shenyuRegisterCenterConfig.getRegisterType());
}

到这里我们的应用已经将信息告知到了shenyu-admin。

shenyu-admin如何接收更新的消息

shenyu-admin作为管理后台会将数据存储到db中,并且同步数据到网关服务。

在上面http example中已经知道了我们的服务是基于 ShenyuClientRegisterRepository 来像shenyu-admin来注册MetaData等信息的。

ShenyuClientRegisterRepository有多种实现根据我们的配置来进行实例化。如http,nacos... 现在就来看看admin这里是如何接收注册消息的。

基于http的注册方式是基于ShenyuClientHttpRegistryController里面的接口来接收消息实现注册的

@PostMapping("/register-metadata")
@ResponseBody
public String registerMetadata(@RequestBody final MetaDataRegisterDTO metaDataRegisterDTO) {
    publisher.publish(metaDataRegisterDTO);
    return ShenyuResultMessage.SUCCESS;
}

也可以看一下基于nacos的注册方式,如果是基于nacos进行注册,则shenyu-admin就会依赖 shenyu-register-client-server-nacos模块来监听注册信息。

//shenyu admin启动的时候会初始化bean,和注册ShenyuClientRegisterRepository相呼应
@Bean(destroyMethod = "close")
public ShenyuClientServerRegisterRepository shenyuClientServerRegisterRepository(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig,
                                                                           final List<ShenyuClientRegisterService> shenyuClientRegisterService) {
    String registerType = shenyuRegisterCenterConfig.getRegisterType();
    ShenyuClientServerRegisterRepository registerRepository = ExtensionLoader.getExtensionLoader(ShenyuClientServerRegisterRepository.class).getJoin(registerType);
    RegisterClientServerDisruptorPublisher publisher = RegisterClientServerDisruptorPublisher.getInstance();
    Map<String, ShenyuClientRegisterService> registerServiceMap = shenyuClientRegisterService.stream().collect(Collectors.toMap(ShenyuClientRegisterService::rpcType, e -> e));
    publisher.start(registerServiceMap);
    registerRepository.init(publisher, shenyuRegisterCenterConfig);
    return registerRepository;
}

//基于nacos的实现类
NacosClientServerRegisterRepository
//上面声明bean的时候就会调用其init方法。最终会调用subscribe方法进行监听

try {
    this.configService = ConfigFactory.createConfigService(nacosProperties);
    this.namingService = NamingFactory.createNamingService(nacosProperties);
} catch (NacosException e) {
    throw new ShenyuException(e);
}

subscribe();

//subscribe方法最终会调用到,就是基于nacos的API来监听

private void subscribeMetadata(final String serviceConfigName) {
    registerMetadata(readData(serviceConfigName));
    LOGGER.info("subscribe metadata: {}", serviceConfigName);
    try {
        configService.addListener(serviceConfigName, defaultGroup, new Listener() {

            @Override
            public Executor getExecutor() {
                return null;
            }

            @Override
            public void receiveConfigInfo(final String config) {
                registerMetadata(config);
            }
        });
    } catch (NacosException e) {
        throw new ShenyuException(e);
    }
}

//最终还是会调用publisher.publish,和基于http的接口 /register-metadata 最终的实现是一样的。
private void publishMetadata(final String data) {
    LOGGER.info("publish metadata: {}", data);
    publisher.publish(Lists.newArrayList(GsonUtis.getInstance().fromJson(data, MetaDataRegisterDTO.class)));
}

//这里的publisher 和上面介绍服务接入中的publisher原理是一样的
也是基于Disruptor高性能队列来实现的一个生产消费的模型。

//消费者最终会调用 MetadataExecutorSubscriber中的
shenyuClientRegisterService.register(metaDataRegisterDTO);

//这里register 了
public String register(final MetaDataRegisterDTO dto) {
    //handler plugin selector
    String selectorHandler = selectorHandler(dto);
    String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);
    //handler selector rule
    String ruleHandler = ruleHandler();
    RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);
    ruleService.registerDefault(ruleDTO);
    //handler register metadata
    registerMetadata(dto);
    //handler context path
    String contextPath = dto.getContextPath();
    if (StringUtils.isNotEmpty(contextPath)) {
        registerContextPath(dto);
    }
    return ShenyuResultMessage.SUCCESS;
}


基于http的实现类是:ShenyuClientRegisterDivideServiceImpl

protected void registerMetadata(final MetaDataRegisterDTO dto) {
    if (dto.isRegisterMetaData()) {
        MetaDataService metaDataService = getMetaDataService();
        MetaDataDO exist = metaDataService.findByPath(dto.getPath());
        metaDataService.saveOrUpdateMetaData(exist, dto);
    }
}

//最终会入库,并且进行了一个eventPublisher.publishEvent操作,这个操作就是同步信息到网关。后面详情说明一下。
public void saveOrUpdateMetaData(final MetaDataDO exist, final MetaDataRegisterDTO metaDataDTO) {
    DataEventTypeEnum eventType;
    MetaDataDO metaDataDO = MetaDataTransfer.INSTANCE.mapRegisterDTOToEntity(metaDataDTO);
    if (Objects.isNull(exist)) {
        Timestamp currentTime = new Timestamp(System.currentTimeMillis());
        metaDataDO.setId(UUIDUtils.getInstance().generateShortUuid());
        metaDataDO.setDateCreated(currentTime);
        metaDataDO.setDateUpdated(currentTime);
        metaDataMapper.insert(metaDataDO);
        eventType = DataEventTypeEnum.CREATE;
    } else {
        metaDataDO.setId(exist.getId());
        metaDataMapper.update(metaDataDO);
        eventType = DataEventTypeEnum.UPDATE;
    }
    // publish MetaData's event
    eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.META_DATA, eventType,
            Collections.singletonList(MetaDataTransfer.INSTANCE.mapToData(metaDataDO))));
}

到这里我们大致了解了整个MetaData(URIRegister原理一样)数据从服务注册到shenyu-admin的整个流程,后面就可以看看是怎么将数据同步到网关的了。也就是上面提到的:eventPublisher.publishEvent(new DataChangedEvent .....)

shenyu-admin同步数据到网关

上面的eventPublisher是ApplicationEventPublisher,它是spring自带的发布监听的功能。

eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.META_DATA, eventType,
        Collections.singletonList(MetaDataTransfer.INSTANCE.mapToData(metaDataDO))));

//找到监听的地方DataChangedEventDispatcher,发布的消息会在onApplicationEvent方法中接收

public void onApplicationEvent(final DataChangedEvent event) {
    for (DataChangedListener listener : listeners) {
        switch (event.getGroupKey()) {
            case APP_AUTH:
                listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
                break;
            case PLUGIN:
                listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
                break;
            case RULE:
                listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
                break;
            case SELECTOR:
                listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                break;
            case META_DATA:
                listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
        }
    }
}
//listener有多个实现类，到底使用的是哪一个


//先来看看DataSyncConfiguration配置类,里面配置了通过哪种方式同步数据到网关
shenyu-admin中的application.yml中找到配置的地方:
默认是websocket:
  sync:
      websocket:
         enabled: true
      messageMaxSize: 10240
#      zookeeper:
#        url: localhost:2181
#        sessionTimeout: 5000
#        connectionTimeout: 2000
#      http:
#        enabled: true

还有nacos等...
 
如果是websocket则listener对应的是WebsocketDataChangedListener 
如果是http则listener对应的是HttpLongPollingDataChangedListener
nacos对应的是NacosDataChangedListener
其他的可以自己查看一下。

如果是基于websocket,admin则会和网关服务建立了websocket连接,然后发送消息到网关。

shenyu-admin这里的DataChangedListener会和网关的SyncDataService相呼应。比如WebsocketDataChangedListener 就对应了网关的WebsocketSyncDataService。

网关同步数据的功能都集中在shenyu-sync-data-center模块中,也是提供了多种实现对应admin中同步数据的方式。
也是基于配置来看看到底使用哪个SyncDataService。下面是websocket的配置:
@Configuration
@ConditionalOnClass(WebsocketSyncDataService.class)
@ConditionalOnProperty(prefix = "shenyu.sync.websocket", name = "urls")

网关调用服务

关于网关是如何调用到服务的,这块主要是基于ShenyuWebHandler来对请求进行处理的。
这块还没做更深入的研究,准备放到后面继续来学习。