Nacos注册中心：发现篇

经过上一节的努力，我们已经将RPC服务成功的注册到Nacos上了。

我们还是以老生常谈的A调用B为例，B的所有实例B1、B2…都在Nacos上了。我们本节要实现的，都客户端，也就是A的部分。

老规矩，先引入依赖：

  1. implementation 'com.alibaba.nacos:nacos-client:2.0.3'
  2. implementation 'org.springframework.boot:spring-boot-autoconfigure:2.2.0.RELEASE'

上述除了引入nacos的依赖外，还引入了spring-boot的自动配置包，后续做客户端的自动装配时会用到。

客户端改造

在正式对接Nacos前，我们先对客户端的包做一些改造。

首先，引入一个通用的Grpc客户端实现：

  1. public abstract class HSGrpcClient implements AutoCloseable {
  3.     private ManagedChannel channel;
  5.     private String ip;
  7.     private int port;
  9.     public HSGrpcClient(String ip, int port) {
  10.         this.ip = ip;
  11.         this.port = port;
  12.     }
  14.     public void init() {
  15.         channel = ManagedChannelBuilder
  16.                 .forTarget(ip + ":" + port)
  17.                 .usePlaintext()
  18.                 .build();
  19.         initSub(channel);
  20.     }
  22.     protected abstract void initSub(Channel channel);
  24.     public void close() throws InterruptedException {
  25.         channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
  26.     }
  28. }

代码如上所示：

  • HSGrpcClient管理了ManagedChannel，这是用于实际网络通信的连接池。

  • 提供了initStub抽象方法，让子类根据自己的需求，去初始化自己的stub。

  • 实现了AutoCloseable接口，让客户端可以通过close方法自动关闭。

在这个基础上，我们改造之前的具体RPC客户端，如下：

  1. public class HomsDemoGrpcClient extends HSGrpcClient {
  3.     private Logger LOG = LoggerFactory.getLogger(HomsDemoGrpcClient.class);
  6.     private HomsDemoGrpc.HomsDemoFutureStub futureStub;
  8.     /**
  9.      * Construct client for accessing HelloWorld server using the existing channel.
  10.      */
  11.     public HomsDemoGrpcClient(String ip, int port) {
  12.         super(ip, port);
  13.     }
  15.     @Override
  16.     protected void initSub(Channel channel) {
  17.         futureStub = HomsDemoGrpc.newFutureStub(channel);
  18.     }
  20.     public Optional<Integer> add(int val1, int val2) {
  21.         AddRequest request = AddRequest.newBuilder().setVal1(val1).setVal2(val2).build();
  22.         try {
  24.             AddResponse response = futureStub.add(request).get();
  25.             return Optional.ofNullable(response.getVal());
  26.         } catch (Exception e) {
  27.             LOG.error("grpc add exception", e);
  28.             return Optional.empty();
  29.         }
  30.     }
  32. }

如上，我们改用了FutureStub，并且将Manage的管理部分，移到了基类中。

SimpleGrpcClientManager的实现

在正式引入Nacos之前，我们先实现一个“看起来没什么营养”的SimpleGrpcClientManager，它可以提供IP、Port直连的客户端管理。

首先是基类：

  1. public abstract class AbstractGrpcClientManager<T extends HSGrpcClient> {
  3.     protected Logger LOG = LoggerFactory.getLogger(getClass());
  5.     protected volatile CopyOnWriteArrayList<T> clientPools = new CopyOnWriteArrayList<>();
  7.     protected Class<T> kind;
  9.     public AbstractGrpcClientManager(Class<T> kind) {
  10.         this.kind = kind;
  11.     }
  13.     public Optional<T> getClient() {
  14.         if (clientPools.size() == 0) {
  15.             return Optional.empty();
  16.         }
  17.         int pos = ThreadLocalRandom.current().nextInt(clientPools.size());
  18.         return Optional.ofNullable(clientPools.get(pos));
  19.     }
  21.     public abstract void init() throws Exception;
  23.     public void shutdown() {
  24.         clientPools.forEach(c -> {
  25.             try {
  26.                 shutdown(c);
  27.             } catch (InterruptedException e) {
  28.                 LOG.error("shutdown client exception", e);
  29.             }
  30.         });
  31.     }
  33.     protected void shutdown(HSGrpcClient client) throws InterruptedException {
  34.         client.close();
  35.     }
  37.     protected Optional<HSGrpcClient> buildHsGrpcClient(String ip, int port) {
  38.         try {
  39.             Class[] cArg = {String.class, int.class};
  40.             HSGrpcClient client = kind.getDeclaredConstructor(cArg)
  41.                     .newInstance(ip, port);
  42.             client.init();
  43.             return Optional.ofNullable(client);
  44.         } catch (Exception e) {
  45.             LOG.error("build MyGrpcClient exception, ip = "+ ip + " port = "+ port, e);
  46.             return Optional.empty();
  47.         }
  48.     }
  50. }

代码如上，解释一下：

  • clientPools是一组HSGrpcClient对象，即支持同时与多个微服务实例(多组不同的ip和端口)建立连接。在微服务场景下，这一特性尤为重要。
  • 而从每一个HSGrpcClient的视角来看，其内置的ManagedChannel内部实现了连接池。因此针对同一个微服务的ip和端口，我们只需要一个HSGrpcClient的实例即可。

下面，我们看一下基础的、不带服务发现的实现：

  1. package com.coder4.homs.demo.client;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  6. import java.util.Arrays;
  7. import java.util.concurrent.CopyOnWriteArrayList;
  9. /**
  10.  * @author coder4
  11.  */
  12. public class SimpleGrpcClientManager<T extends HSGrpcClient> extends AbstractGrpcClientManager<T> {
  14.     protected Logger LOG = LoggerFactory.getLogger(SimpleGrpcClientManager.class);
  16.     private String ip;
  18.     private int port;
  20.     public SimpleGrpcClientManager(Class<T> kind, String ip, int port) {
  21.         super(kind);
  22.         this.ip = ip;
  23.         this.port = port;
  24.     }
  26.     public void init() {
  27.         // init one client only
  28.         HSGrpcClient client = buildHsGrpcClient(ip, port)
  29.                 .orElseThrow(() -> new RuntimeException("build HsGrpcClient fail"));
  30.         clientPools = new CopyOnWriteArrayList(Arrays.asList(client));
  31.     }
  33.     public static void main(String[] args) throws Exception {
  34.         SimpleGrpcClientManager<HomsDemoGrpcClient> manager = new SimpleGrpcClientManager(HomsDemoGrpcClient.class, "127.0.0.1", 5000);
  35.         manager.init();
  36.         manager.getClient().ifPresent(t -> System.out.println(t.add(1, 2)));
  37.         manager.shutdown();
  38.     }
  40. }

从上述实现中不难发现：

  • 该实现中，默认只与预先设定的IP和端口，构造一个单独的HSGrpcClient。

  • 由于IP和端口通过外部指定，因此使用了CopyOnWriteArrayList以保证线程安全。

NacosGrpcClientManager的实现

下面，我们着手实现带Nacos服务发现的版本。

  1. package com.coder4.homs.demo.client;
  3. import com.alibaba.nacos.api.naming.NamingFactory;
  4. import com.alibaba.nacos.api.naming.NamingService;
  5. import com.alibaba.nacos.api.naming.listener.NamingEvent;
  6. import com.alibaba.nacos.api.naming.pojo.Instance;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10. import java.util.concurrent.CopyOnWriteArrayList;
  12. /**
  13.  * @author coder4
  14.  */
  15. public class NacosGrpcClientManager<T extends HSGrpcClient> extends AbstractGrpcClientManager<T> {
  17.     protected String serviceName;
  19.     protected String nacosServer;
  21.     protected NamingService namingService;
  23.     public NacosGrpcClientManager(Class<T> kind, String nacosServer, String serviceName) {
  24.         super(kind);
  25.         this.nacosServer = nacosServer;
  26.         this.serviceName = serviceName;
  27.     }
  29.     @Override
  30.     public void init() throws Exception {
  31.         namingService = NamingFactory
  32.                 .createNamingService(nacosServer);
  33.         namingService.subscribe(serviceName, e -> {
  34.             if (e instanceof NamingEvent) {
  35.                 NamingEvent event = (NamingEvent) e;
  36.                 rebuildClientPools(event.getInstances());
  37.             }
  38.         });
  39.         rebuildClientPools(namingService.selectInstances(serviceName, true));
  40.     }
  42.     private void rebuildClientPools(List<Instance> instanceList) {
  43.         ArrayList<HSGrpcClient> list = new ArrayList<>();
  44.         for (Instance instance : instanceList) {
  45.             buildHsGrpcClient(instance.getIp(), instance.getPort()).ifPresent(c -> list.add(c));
  46.         }
  47.         CopyOnWriteArrayList<T> oldClientPools = clientPools;
  48.         clientPools = new CopyOnWriteArrayList(list);
  49.         // destory old ones
  50.         oldClientPools.forEach(c -> {
  51.             try {
  52.                 c.close();
  53.             } catch (InterruptedException e) {
  54.                 LOG.error("MyGrpcClient shutdown exception", e);
  55.             }
  56.         });
  57.     }
  59. }

解释如下：

  • 在init方法中，初始化了NamingService，并订阅对应serviceName服务的更新事件。

  • 当第一次，或者有服务更新时，我们会根据最新列表，重建所有的HSGrpcClient

  • 每次重建后，关闭老的HSGrpcClient

为了让上述客户端使用更加方便，我们添加了如下的自动配置：

  1. @Configuration
  2. public class HomsDemoGrpcClientManagerConfiguration {
  4.     @Bean(name = "homsDemoGrpcClientManager")
  5.     @ConditionalOnMissingBean(name = "homsDemoGrpcClientManager")
  6.     @ConditionalOnProperty(name = {"nacos.server"})
  7.     public AbstractGrpcClientManager<HomsDemoGrpcClient> nacosManager(
  8.             @Value("${nacos.server}") String nacosServer) throws Exception {
  9.         NacosGrpcClientManager<HomsDemoGrpcClient> manager =
  10.                 new NacosGrpcClientManager<>(HomsDemoGrpcClient.class,
  11.                         nacosServer, HomsDemoConstant.SERVICE_NAME);
  12.         manager.init();
  13.         return manager;
  14.     }
  15. }

如上所示：

  • nacos的server地址由yaml中配置

  • serviceName由client包中的常量文件HomsDemoConstant提供(即homs-demo)

为了让上述自动配置自动生效，我们还需要添加META-INF/spring.factories文件

  1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  2. com.coder4.homs.demo.configuration.HomsDemoGrpcClientManagerConfiguration

最后，我们来实验一下服务发现的效果

  1. 启动Server进程，检查Nacos上，应当出现了自动注册的RPC服务。

  2. 开发客户端驱动的项目，引用上述client包、配置yaml中的nacos服务地址

  3. 最后，在客户端驱动项目中，通过Autowired自动装配，代码类似：

  1. @Autowired
  2. private AbstractGrpcClientManager<HomsDemoGrpcClient> homsClientManager;
  5. // Usage
  6. homsClientManager.getClient().ifPresent(client -> client.add(1, 2));

如果一切顺利，会自动发现nacos上已经注册的服务实例，并成功执行rpc调用。