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來對請求進行處理的。
這塊還沒做更深入的研究,准備放到後面繼續來學習。