(八) Nepxion-Thunder分布式RPC集成框架 - 注册中心

Nepxion-Thunder（QQ 群 471164539）发布在https://github.com/Nepxion/

 

1. 存储结构

点击查看大图

• 应用存储区
  位于顶级节点application下，分不同的通信中间件进行存储。树状结构：
  对于服务端，[中间件名]/[组名]/[应用名]/service/[接口名]/[ApplicationEntity的Json串]
  对于调用端，[中间件名]/[组名]/[应用名]/reference/[接口名]/[ApplicationEntity的Json串]
  其中[应用名]节点下存储ApplicationConfig.java对象，内容如下：
  

  public class ApplicationConfig implements Serializable {
      private static final long serialVersionUID = -6769132534750910991L;
  
      private String application; // 应用名
      private String group;       // 组名
      private int frequency;      // 令牌刷新的时钟周期
  }
  

  其中service/[接口名]节点下存储ServiceConfig.java对象，内容如下：
  

  public class ServiceConfig implements Serializable {
      private static final long serialVersionUID = 5777064003531668211L;
  
      private String interfaze;     // 接口名
      private List<String> methods; // 方法列表
      private String secretKey;     // 密钥
      private int version;          // 版本号
      private long token;           // 最大令牌数
  }

  其中reference/[接口名]节点下存储ReferenceConfig.java对象，内容如下：
  

  public class ReferenceConfig implements Serializable {
      private static final long serialVersionUID = -4278894097968838119L;
  
      private String interfaze; // 接口名    
      private String secretKey; // 密钥    
      private int version; // 版本号
  } 

  其中ApplicationEntity的Json串属于临时节点，当服务端或者调用端下线后，在一定的超时时间后，会自动被删除。注意：最新版本加上了UUID和上下时间，便于服务治理，从存储结构图上并没有反应出来
• 配置存储区
  位于顶级节点configuration下，集成式配置，所有通信中间件的配置都可以集中管理，服务端和调用端配置也集中管理。树状结构：
  [组名]/[应用名]
  其中[应用名]节点下存储Properties文本，内容跟thunder-ext.properties配置文件一致
• 监控存储区
  位于顶级节点monitor下，子节点为第三方WebService监控系统的URL
• 用户存储区
  位于顶级节点user下，子节点为UserEntity序列化对象

2. 实现思路

  1）RegistryInitializer.java - 暴露的注册中心初始化接口。默认的注册中心是用Zookeeper来实现，也可以换做其它开源框架来实现(例如Redis)，只需要实现该接口即可

public interface RegistryInitializer extends ThunderDelegate {
    
    // 启动和注册中心的连接
    void start(RegistryEntity registryEntity) throws Exception;

    // 启动和注册中心的连接
    void start(RegistryEntity registryEntity, ThunderProperties properties) throws Exception;

    // 停止注册中心的连接
    void stop() throws Exception;
}

  2）RegistryExecutor.java - 暴露的注册中心查询，执行等功能接口。默认的注册中心是用Zookeeper来实现，也可以换做其它开源框架来实现(例如Redis)，只需要实现该接口即可

public interface RegistryExecutor extends ThunderDelegate {

    // 设置注册中心初始化器
    void setRegistryInitializer(RegistryInitializer registryInitializer);
    
    // 设置协议实体
    void setProtocolEntity(ProtocolEntity protocolEntity);
    
    // 设置名称空间
    void setNamespace(String namespace);
    
    // 初始化注册中心Application相关环境
    void registerApplicationEnvironment(ApplicationEntity applicationEntity) throws Exception;
    
    // 初始化注册中心Configuration相关环境
    void registerConfigurationEnvironment() throws Exception;
    
    // 初始化注册中心Monitor相关环境
    void registerMonitorEnvironment() throws Exception;
    
    // 初始化注册中心User相关环境
    void registerUserEnvironment() throws Exception;

    // 注册Application
    void registerApplication(ApplicationEntity applicationEntity) throws Exception;
    
    // 注册Service目录
    void registerServiceCategory(String interfaze, ApplicationEntity applicationEntity) throws Exception;

    // 注册Service，并把服务所在的应用信息写入
    void registerService(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 注册Reference目录
    void registerReferenceCategory(String interfaze, ApplicationEntity applicationEntity) throws Exception;

    // 注册Reference，并把服务所在的应用信息写入
    void registerReference(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 注册Configuration
    void registerConfiguration(ApplicationEntity applicationEntity) throws Exception;
    
    // 注册Monitor
    void registerMonitor(String address) throws Exception;
    

    // 获取Application配置信息
    ApplicationConfig retrieveApplication(ApplicationEntity applicationEntity) throws Exception;

    // 获取Service配置信息
    ServiceConfig retrieveService(String interfaze, ApplicationEntity applicationEntity) throws Exception;

    // 获取Reference配置信息
    ReferenceConfig retrieveReference(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 获取Property文本配置信息
    String retrieveProperty(ApplicationEntity applicationEntity) throws Exception;
    

    // 持久化Application配置信息
    void persistApplication(ApplicationConfig applicationConfig) throws Exception;

    // 持久化Service配置信息
    void persistService(ServiceConfig serviceConfig, ApplicationEntity applicationEntity) throws Exception;

    // 持久化Reference配置信息
    void persistReference(ReferenceConfig referenceConfig, ApplicationEntity applicationEntity) throws Exception;
    
    // 持久化Property文本配置信息
    void persistProperty(String property, ApplicationEntity applicationEntity) throws Exception;
    

    // 获取服务实例列表
    List<ApplicationEntity> getServiceInstanceList(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 获取调用实例列表
    List<ApplicationEntity> getReferenceInstanceList(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 获取监控实例列表
    List<String> getMonitorInstanceList() throws Exception;
    
    // 判断服务实例是否Online
    boolean isServiceInstanceOnline(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 判断调用实例是否Online
    boolean isReferenceInstanceOnline(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 判断监控实例是否Online
    boolean isMonitorInstanceOnline(String monitorInstance) throws Exception;
    
    // 获取用户列表
    List<UserEntity> retrieveUserList() throws Exception;
    
    // 获得用户
    UserEntity retrieveUser(String name) throws Exception;
    
    // 持久化用户
    void persistUser(UserEntity userEntity) throws Exception;
    
    // 删除用户
    void deleteUser(UserEntity userEntity) throws Exception;
    

    // 监听Application配置信息变更
    void addApplicationConfigWatcher(ApplicationConfig applicationConfig) throws Exception;
    
    // 监听Service配置信息变更
    void addServiceConfigWatcher(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 监听Reference配置信息变更
    void addReferenceConfigWatcher(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 监听Service上下线
    void addServiceInstanceWatcher(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 监听Reference上下线
    void addReferenceInstanceWatcher(String interfaze, ApplicationEntity applicationEntity) throws Exception;

    // 监听Monitor上下线，用来保持注册中心和本地缓存一致
    void addMonitorInstanceWatcher() throws Exception;

    // 监听注册中心应用与注册中心断开后重连成功后，触发事件
    void addReconnectionListener();
    
    
    // 获取Category名称列表
    List<String> getCategoryList() throws Exception;
    
    // 获取Protocol名称列表
    List<String> getProtocolList() throws Exception;
    
    // 获取Group名称列表
    List<String> getGroupList() throws Exception;

    // 获取Application名称列表
    List<String> getApplicationList(String group) throws Exception;

    // 获取Service名称列表
    List<String> getServiceList(String application, String group) throws Exception;

    // 获取Reference名称列表
    List<String> getReferenceList(String application, String group) throws Exception;
    
    // 获取Configuration Group名称列表
    List<String> getConfigurationGroupList() throws Exception;

    // 获取Configuration Application名称列表
    List<String> getConfigurationApplicationList(String group) throws Exception;
    
    
    // 重置ApplicationConfig，所有属性值恢复为默认值
    void resetApplication(ApplicationEntity applicationEntity) throws Exception;

    // 重置ServiceConfig，所有属性值恢复为默认值
    void resetService(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 重置ReferenceConfig，所有属性值恢复为默认值
    void resetReference(String interfaze, ApplicationEntity applicationEntity) throws Exception;
    
    // 更改ApplicationConfig令牌刷新频率， ApplicationEntity只需要带application和group(下同)
    void modifyApplicationFrequency(ApplicationEntity applicationEntity, int frequency) throws Exception;
    
    // 更改ApplicationConfig单个属性
    void modifyApplication(ApplicationEntity applicationEntity, MethodInvocation invocation) throws Exception;
    
    // 更改ApplicationConfig批量属性
    void modifyApplication(ApplicationEntity applicationEntity, List<MethodInvocation> invocationList) throws Exception;
    
    // 更改ServiceConfig密钥
    void modifyServiceSecretKey(String interfaze, ApplicationEntity applicationEntity, String secretKey) throws Exception;
  
    // 更改ServiceConfig版本
    void modifyServiceVersion(String interfaze, ApplicationEntity applicationEntity, int version) throws Exception;
    
    // 更改ServiceConfig令牌数
    void modifyServiceToken(String interfaze, ApplicationEntity applicationEntity, long token) throws Exception;
    
    // 更改ServiceConfig单个属性
    void modifyService(String interfaze, ApplicationEntity applicationEntity, MethodInvocation invocation) throws Exception;
    
    // 更改ServiceConfig批量属性
    void modifyService(String interfaze, ApplicationEntity applicationEntity, List<MethodInvocation> invocationList) throws Exception;
    
    // 更改ReferenceConfig密钥
    void modifyReferenceSecretKey(String interfaze, ApplicationEntity applicationEntity, String secretKey) throws Exception;

    // 更改ServiceConfig版本
    void modifyReferenceVersion(String interfaze, ApplicationEntity applicationEntity, int version) throws Exception;
    
    // 更改ReferenceConfig单个属性
    void modifyReference(String interfaze, ApplicationEntity applicationEntity, MethodInvocation invocation) throws Exception;
    
    // 更改ReferenceConfig批量属性
    void modifyReference(String interfaze, ApplicationEntity applicationEntity, List<MethodInvocation> invocationList) throws Exception;
}

  3）RegistryLauncher.java - 暴露给外部程序访问注册中心的接口

// 提供给外部程序所用
public interface RegistryLauncher {

    // 启动注册中心连接
    void start(String address, ProtocolType protocolType) throws Exception;

    // 停止注册中心连接
    void stop() throws Exception;

    // 获取注册中心执行器
    RegistryExecutor getRegistryExecutor();
}

  4）ZookeeperRegistryInitializer.java - 实现RegistryInitializer.java
  5）ZookeeperRegistryExecutor.java - 实现RegistryExecutor.java
  6）ZookeeperRegistryLauncher.java - 实现RegistryLauncher.java
  7）ZookeeperApplicationConfigWatcher.java - 实现对“[中间件名]/[组名]/[应用名]“路径的EXISTS事件监听，对应的ApplicationConfig里内容改变，会触发监听事件(令牌刷新的时钟周期控制)
  8）ZookeeperServiceConfigWatcher.java - 实现对“[中间件名]/[组名]/[应用名]/service/[接口名]”路径的EXISTS类型监听，对应的ServiceConfig里内容改变，会触发监听事件(令牌，密钥，版本控制)
  9）ZookeeperReferenceConfigWatcher.java - 实现对“[中间件名]/[组名]/[应用名]/reference/[接口名]”路径的EXISTS类型监听，对应的ReferenceConfig里内 容改变，会触发监听事件(密钥，版本控制)
  10）ZookeeperInstanceWatcher.java -  实现对“[中间件名]/[组名]/[应用名]/service或reference/[接口名]”路径的GET_CHILDREN事件监听，对应的它的子节点增添或者删除会触发监听事件(服务/调用上下线)
  11）ZookeeperInstanceEventInterceptor.java -  实现对上下线的事件拦截
  12）ZookeeperMonitorInstancesWatcher.java - 实现对monitor”路径的GET_CHILDREN事件监听，对应的它的子节点增添或者删除会触发监听事件(监控中心上下线)
  13）ZookeeperReconnectionListener.java - 实现重连监听，一旦Zookeeper重启或者由于出网络问题后又恢复和Zookeeper连接等，会触发重连监听

  14）ZookeeperUserWatcher.java - 实现对“monitor”路径的GET_CHILDREN事件监听，对应的它的子节点增添或者删除会触发监听事件(监控中心上下线)
  15）ZookeeperUserWatcherCallback.java - 实现对“user/[userName]”路径的EXISTS事件监听，它被删除或者数据改变会触发监听事件