vertx 实现动态 RPC

需求:替换ali lightApi 动态rpc的实现,因为api为商业版,不是开源的,是基于pandora 的EDAS平台的。那么我们如何实现开源的动态RPC呢?

定义:动态RPC指的是,可以动态的让一个服务上线和下线,换句话说就是动态的从注册中心剔除服务,而不是停止这个服务,启动这个服务。这个需求是来自链路的原路返回

简介 vertx:

vertx 是一套全异步的基于netty通信的框架,Vertx中核心组建有 verticle, eventbus, circuit breaker,service discovery and register, Router等等。后期我会一一的去探索这里面的组件的原理。而本文介绍的vertx 的动态rpc,就是使用vertx proxy service 这个功能来实现的。另外我们得知道vertx 中的线程模型是基于netty的event loop,一个verticle 是一个微服务,而且具有HA的机制的微服务。服务之间的通信是根据event bus 的netty 通信机制。但是event bus 通信不是100%的可靠的(这点很要命,后期我会写博客探索的)。

此外vertx需要依赖 分布式缓存建立集群(hazelcast, ignite等),基于gossip协议的p2p网络。

探索过程:

如果要实现需求的话,首先第一想到的还是spring cloud,dubbo,thrift 等RPC框架,hsf 是阿里pandora rpc 框架,肯定不能用。但是经一番折腾之后,没有可以让我觉得可以在代码里面直接让服务上线下线,从注册中心剔除掉或者注册到注册中心的。那怎么办呢?曾经也想过会用rabbitmq,这样的MQ 去做动态的RPC,但是发现很复杂,关键很难做scalablity,如果有上千个链路,就需要上千个topic,或者tag 这类的标记。感觉很复杂,也很难维护。

solutions: vertx proxy service 

这个方案上,我们不打算使用verticle的概念,仅仅使用vertx 代理服务的概念。首先我们要使用vertx 的服务发现和注册,其次我们要使用proxy service。

第一个坑:代理服务的自动生成。请在pom或者gradle里面加入 生成代理的依赖(对接口的代理)。然后创建 package-info.java,在root package。所谓的root package就是基package,web developer 开的springboot 应该都很清楚,需要将app.class 建在基package,以便扫描都可以扫描到

    <groupId>io.vertx</groupId>

    <artifactId>vertx-service-proxy</artifactId>

    <classifier>processor</classifier>

    <groupId>io.vertx</groupId>

    <artifactId>vertx-codegen</artifactId>

    <classifier>processor</classifier>

@ModuleGen(name =”ap-common-vertx”, groupPackage =”com.xxxx.xx.xx.vertx”)

package com.xxxx.xx.xx.vertx;

import io.vertx.codegen.annotations.ModuleGen;

其中groupPackage =”com.xxxx.xx.xx.vertx”), 是基package, 也可以是 你定义interface 所在的包。完成之后,使用maven clean install, 你就会发现在target 目录下面会有 …Proxy 的class,这两个class 就是代理类,对于原理,后面的博客我会慢慢分析。

第二个坑是:服务发现publish 服务之后,需要注册服务代理,不然注册的服务,在event bus 上面是找不到的,看看代码怎么写吧:

Record record = EventBusService.createRecord(servicePublishRequestBean.getServiceName(), servicePublishRequestBean.getEventBusAddress(), servicePublishRequestBean.getClazz());

servicePublishRequestBean.getDiscovery().publish(record, ar -> {

if (ar.succeeded() && ar.result() !=null) {

publishedRecords.add(record);

        recordMessageConsumerMap.putIfAbsent(record, ProxyServiceUtil.registerProxyService(servicePublishRequestBean.getEventBusAddress(), servicePublishRequestBean.getClazz(), servicePublishRequestBean.getService()));

    }

handler.handle(ar.map(ar.result()));

});

servicePublishRequestBean就是我封装的bean,里面de属性有:

private StringserviceName;

private StringeventBusAddress;

private ServiceDiscoverydiscovery;

private Classclazz;

private T service;

此外我们一定要注册这样的proxy service才能够生效,不然是没有用的:

ProxyServiceUtil.registerProxyService(servicePublishRequestBean.getEventBusAddress(), servicePublishRequestBean.getClazz(), servicePublishRequestBean.getService()));

看看里面怎么写的

public static MessageConsumerregisterProxyService(String eventBusAddress, Class clazz, T service) {

return ServiceBinderUtil.getBinderInstance()

.setAddress(eventBusAddress)

.register(clazz, service);

}

第三个坑是: 和第二个坑一样的,需要注销服务,注销服务的话,我们也要调用proxyService 去把服务注销了,而不能单纯的调用 service discovery and register unpublish 方法,这个是很恶心的:看看我怎么写的把,一些很细节的东西需要自己探索,我不会贴出所有的东西的。比如下面的discover record 有许多的状态,这里一不小心就会找不到你所发布的服务,另外注销代理服务,需要个奇怪的参数,这个参数我存储在map里面:recordMessageConsumerMap, 在服务publish时候,就会生成的。

discovery.getRecord(info -> info.getName().equals(serviceName), res -> {

if (res.succeeded() && res.result() !=null && res.result().getStatus() == Status.UP) {

Record record = res.result();

                discovery.unpublish(record.getRegistration(), ar -> {

if(ar.succeeded()) {

List records =recordMessageConsumerMap.keySet().stream().filter(map-> map.getName().equals(serviceName)).collect(Collectors.toList());

                        offlineRecord(records);

                    }

handler.handle(ar.map((Void)null));

                });

            }else {

handler.handle(res.map((Void)null));

            }

}

);

OfflineRecord 主要是调用ProxyServiceUtil.unregisterProxyService(recordMessageConsumerMap.get(records.get(0)));,其中unregisterProxyService 方法的实现如下,是不是比较简单?

ServiceBinderUtil.getBinderInstance().unregister(consumer);

然后基本上做到这里,vertx 动态的rpc 就可以实现了,主要是用了vertx 服务发现组件的,publish, unpublish方式,和proxy service 的 register和unregister方法。但是里面的坑比较多。此外除了上面所描述的坑之外,我还想告诉小伙伴们,event bus 通信exception 会出现一些奇怪的错误,所有event bus 通信也是个坑,其实proxy sevice 本质就是 event bus,所以event bus 网络连接不通的话,确实很头疼,经过我这边的测试,不管ecs 集群的部署,还是ecs 和docker 的混合部署,网络都是可以通的(我使用的是ignite分布式缓存)。暂时我先贴出来关于网络的代码,如果你们遇到了问题,先暂时按照我这个来,不会让你很恼火。

@Bean

public IgniteConfigurationgetIgniteSelfConfiguration()throws Exception{

IgniteConfiguration igniteConfiguration =new IgniteConfiguration();

    igniteConfiguration.setClientMode(false);

    igniteConfiguration.setPeerClassLoadingEnabled(true);

    igniteConfiguration.setDeploymentMode(DeploymentMode.CONTINUOUS);

    igniteConfiguration.setPeerClassLoadingMissedResourcesCacheSize(0);

    igniteConfiguration.setDiscoverySpi(getTcpDiscoverySpi());

    igniteConfiguration.setCacheConfiguration(getCacheConfiguration());

  // igniteConfiguration.setLocalHost(IPUtil.getLocalIp());

    return igniteConfiguration;

}

@Bean

public TcpDiscoverySpigetTcpDiscoverySpi()throws Exception{

TcpDiscoverySpi tcpDiscoverySpi =new TcpDiscoverySpi();

    tcpDiscoverySpi.setIpFinder(getTcpDiscoveryMulticastIpFinder());

    tcpDiscoverySpi.setNetworkTimeout(10000);

    System.out.println(“setting success for host”);

    tcpDiscoverySpi.setLocalAddress(IPUtil.getLocalIp());

    return tcpDiscoverySpi;

}

@Bean

public TcpDiscoveryMulticastIpFindergetTcpDiscoveryMulticastIpFinder(){

TcpDiscoveryMulticastIpFinder tcpDiscoveryMulticastIpFinder =new TcpDiscoveryMulticastIpFinder();

    tcpDiscoveryMulticastIpFinder.setMulticastGroup(“224.0.0.100”);

    return tcpDiscoveryMulticastIpFinder;

}

@Bean

public CacheConfigurationgetCacheConfiguration(){

CacheConfiguration cacheConfiguration =new CacheConfiguration();

    cacheConfiguration.setName(“myCache”);

    cacheConfiguration.setCacheMode(CacheMode.PARTITIONED);

    cacheConfiguration.setBackups(1);

    return cacheConfiguration;

}

@Bean

public VertxClusterStartervertxClusterStarter() {

VertxClusterStarter vertxClusterStarter =new VertxClusterStarter();

    return vertxClusterStarter;

}

ClusterManager clusterManager =new IgniteClusterManager(igniteSelfConfiguration);

TcpDiscoverySpi discoverySpi = (TcpDiscoverySpi)igniteSelfConfiguration.getDiscoverySpi();

TcpDiscoveryMulticastIpFinder tcpDiscoveryMulticastIpFinder = (TcpDiscoveryMulticastIpFinder) discoverySpi.getIpFinder();

tcpDiscoveryMulticastIpFinder.setAddresses(Arrays.asList(propertiesHolderUtils.getVertxClusterIps().split(CommonConstants.CHARACTER_SEPARATOR_COMMA)));

VertxOptions vertxOptions =new VertxOptions().setClustered(true).setClusterHost(IPUtil.getLocalIp()).setClusterPort(Integer.valueOf(propertiesHolderUtils.getVertxClusterPorts().split(CommonConstants.CHARACTER_SEPARATOR_COMMA)[0])).setClusterManager(clusterManager);

ServiceDiscoveryOptions discoveryOptions =new ServiceDiscoveryOptions();

谢谢,希望对大家有所帮助,后面我会尝试探索event bus 通信的原理。

    原文作者:Double_winter
    原文地址: https://www.jianshu.com/p/1cc263bdccb4
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞