Nacos源码系列—订阅机制的前因后果(下)

点赞再看,养成习惯,微信搜索【牧小农】关注我获取更多资讯,风里雨里,小农等你,很高兴能够成为你的朋友。
项目源码地址:公众号回复 nacos,即可免费获取源码

事件发布

在上一节中我们讲解了在NotifyCenter中维护了事件名称和事件发布者的关系,而默认的事件发布者为DefaultPublisher,今天我们就来讲一下DefaultPublisher的事件发布的具体逻辑

首先我们来看一下DefaultPublisher的源码:

public class DefaultPublisher extends Thread implements EventPublisher {     @Override     public void init(Class<? extends Event> type, int bufferSize) {         //守护线程         setDaemon(true);         //设置线程名         setName("nacos.publisher-" + type.getName());         this.eventType = type;         this.queueMaxSize = bufferSize;         //阻塞队列初始化         this.queue = new ArrayBlockingQueue<>(bufferSize);         //启动线程         start();     }          @Override     public synchronized void start() {         if (!initialized) {             // start just called once             //启动run方法             super.start();             if (queueMaxSize == -1) {                 queueMaxSize = ringBufferSize;             }             initialized = true;         }     } } 

我们可以看到这个类继承自Thread,说明他是一个线程类,同时实现了EventPublisher说明他也是一个发布者,在init()中,是以守护线程的方式运作的,同时初始化了一个阻塞队列,最后调用start()启动线程。

在start()里面,其实就是启动run():

@Override public void run() {     openEventHandler(); }     void openEventHandler() {     try {          // This variable is defined to resolve the problem which message overstock in the queue.         int waitTimes = 60;         // To ensure that messages are not lost, enable EventHandler when         // waiting for the first Subscriber to register         //死循环遍历,线程启动设置最大延迟60秒,用来解决消息积压问题         for (; ; ) {             if (shutdown || hasSubscriber() || waitTimes <= 0) {                 break;             }             ThreadUtils.sleep(1000L);             waitTimes--;         }         //死循环从队列中取出event对象,同时通知订阅者(subscriber)执行event对象         for (; ; ) {             if (shutdown) {                 break;             }             final Event event = queue.take();             receiveEvent(event);             UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));         }     } catch (Throwable ex) {         LOGGER.error("Event listener exception : ", ex);     } } 

在上述代码中我们可以看到for (; ; )这个循环出现了两次,这个就是循环遍历(死循环),第一个死循环我们可以理解成延时效果,里面最大延时60秒,每隔一秒运行一次,判断(当前线程是否关闭、是否有订阅者、是否超过60秒)只要满足其中任意一个条件,跳出循环
第二个死循环,是我们业务逻辑处理,用来消费,从队列中取出event事件,然后通过receiveEvent()执行。

那么我们可以从队列中取出事件,那么这个事件又在哪一步注入进去的呢,我们还是在当前类里面,找到一个叫publish()的方法

@Override public boolean publish(Event event) {     checkIsStart();     //向队列中插入元素     boolean success = this.queue.offer(event);     //判断是否插入成功     if (!success) {         LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);         //失败直接执行         receiveEvent(event);         return true;     }     return true; } 

这个方法其实就是发布事件调用了publish往阻塞队列中存入事件,如果失败那么立即执行receiveEvent(),不在继续走队列方法

void receiveEvent(Event event) {     final long currentEventSequence = event.sequence();      if (!hasSubscriber()) {         LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);         return;     }      // Notification single event listener     //循环遍历subscribers对象     for (Subscriber subscriber : subscribers) {         // Whether to ignore expiration events         if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {             LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",                     event.getClass());             continue;         }          // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.         // Remove original judge part of codes.         //通知订阅者执行event         notifySubscriber(subscriber, event);     } } 

而在receiveEvent()方法中,这里其实就是遍历的subscribers集合(订阅者),然后通过notifySubscriber() 通知订阅者方法,而这个subscribers集合就是在我们之前讲到的NacosNamingService.init()方法中设置的。

public class NacosNamingService implements NamingService {  private void init(Properties properties) throws NacosException {    //将Subscribe注册到Publisher    NotifyCenter.registerSubscriber(changeNotifier);  } } 

NotifyCenter.registerSubscriber(changeNotifier);会调用NotifyCenter.addSubscriber()方法,进行最终的操作。

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,       EventPublisherFactory factory) {    final String topic = ClassUtils.getCanonicalName(subscribeType);   synchronized (NotifyCenter.class) {       // MapUtils.computeIfAbsent is a unsafe method.       MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);   }   //获取对应的publisher   EventPublisher publisher = INSTANCE.publisherMap.get(topic);   if (publisher instanceof ShardedEventPublisher) {       ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);   } else {       //添加到subscribers集合       publisher.addSubscriber(consumer);   } } 

addSubscriber()方法的逻辑就是讲订阅事件、发布中、订阅者三个关系进行绑定,而发布者和事件通过Map进行维护,发布者与订阅者通过关联关系进行维护。

我们回到刚刚DefaulePublisher.notifySubscriber()方法,这里是最后执行订阅者事件的方法

@Override public void notifySubscriber(final Subscriber subscriber, final Event event) {    LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);   //执行订阅者事件   final Runnable job = () -> subscriber.onEvent(event);   //执行者   final Executor executor = subscriber.executor();    if (executor != null) {       executor.execute(job);   } else {       try {           job.run();       } catch (Throwable e) {           LOGGER.error("Event callback exception: ", e);       }   } } 

到这里,订阅机制就讲完了,可能会有点绕,最好是我们能够去跟着代码走一遍,这样会比较理解和记忆,在这里我们重点需要理解NotifyCenter对事件发布者、订阅者以及之间关系的维护,关系维护的入口就在NacosNamingService.init()中,我们来看一下他的核心逻辑

Nacos源码系列—订阅机制的前因后果(下)

首先ServiceInfoHolder中通过NotifyCenter发布InstancesChangeEvent事件.

NotifyCenter获取对应的CanonicalName,并将这个参数作为key,从NotifyCenter.publisherMap中获取对应的事件发布者,然后将InstancesChangeEvent事件进行发布.

InstancesChangeEvent事件发布主要是通过EventPublisher的实现类,DefaultPublisher进行InstancesChangeEvent事件发布,而DefaultPublisher本身作为守护线程的方式进行运作,在执行业务逻辑时判断是否线程启动,如果启动,将事件添加到队列中,如果成功,则发布过程完成,如果添加失败,立即执行DefaultPublisher.receiveEvent,接收事件通知订阅者,创建一个Runnable对象,执行订阅者的Event事件。

在添加到队列成功的时候,DefaultPublisher会创建一个阻塞队列(BlockingQueue),标记线程启动,当他执行 super.start(),会调用它的run方法,在这个run方法里面核心的业务逻辑就是openEventHandler(),里面会有两个死循环,第一个是在线程启动的60秒内执行条件,第二个是从阻塞队列中获取Event事件,调用DefaultPublisher.receiveEvent()通知订阅者,流程结束

本地缓存

我们在之前的系列中,客户端会缓存一些信息在本地中,来获取ServiceInfo的信息,但是在执行本地缓存的时候,难免会有一些故障,有故障就需要进行处理,在这里主要涉及到两个类ServiceInfoHolderFailoverReactor

Nacos源码系列—订阅机制的前因后果(下)

Nacos缓存主要是分为两个方面,一个从注册中心获取实例信息缓存到内存中,通过ConcurrentMap进行存储,一个是通过磁盘文件的形式定时缓存。

同时故障处理也分为两个部分,一个是故障处理的开关通过文件进行标记,一个是当起来故障处理后,可以从故障备份的文件中获取服务实例信息。

介绍完上面几点,我们先来详细讲解第一个核心类ServiceInfoHolder

ServiceInfoHolder

ServiceInfoHolder类,主要是用来处理服务信息的,每次客户端从服务端拉取服务信息时,都用经过这个类,而processServiceInfo用来处理本地信息(缓存、发布、更新、本地目录初始化)等

Nacos源码系列—订阅机制的前因后果(下)

ServiceInfo: 注册服务的信息,主要包含(服务名、分组名、集群信息、实例列表、最后一次更新时间),客户端获取的信息,都是通过ServiceInfo作为承载体,ServiceInfoHolder.ServiceInfo,通过ConcurrentMap进行存储,如下所示:

public class ServiceInfoHolder implements Closeable {    private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;       public ServiceInfoHolder(String namespace, Properties properties) {     initCacheDir(namespace, properties);     //启动是判断是否从缓存信息中获取,默认为false     if (isLoadCacheAtStart(properties)) {         //从缓存目录中读取信息         this.serviceInfoMap = new ConcurrentHashMap<>(DiskCache.read(this.cacheDir));     } else {         //创建空集合对象         this.serviceInfoMap = new ConcurrentHashMap<>(16);     }     this.failoverReactor = new FailoverReactor(this, cacheDir);     this.pushEmptyProtection = isPushEmptyProtect(properties); }      public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {         //判断服务key是否为空         String serviceKey = serviceInfo.getKey();         if (serviceKey == null) {             return null;         }         ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());         if (isEmptyOrErrorPush(serviceInfo)) {             //empty or error push, just ignore             return oldService;         }         //将缓存信息放置到map中         serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);         //判断实例信息是否发生改变         boolean changed = isChangedServiceInfo(oldService, serviceInfo);         if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {             serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));         }         //监控服务缓存map的大小         MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());         if (changed) {             NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(),                     JacksonUtils.toJson(serviceInfo.getHosts()));             //添加实例变更事件,被订阅者执行             NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),                     serviceInfo.getClusters(), serviceInfo.getHosts()));             //写入本地文件             DiskCache.write(serviceInfo, cacheDir);         }         return serviceInfo;     } } 

这里就是Nacos获取注册信息的缓存,之前我们有讲过,当服务信息变更的时候会第一时间更新ServiceInfoMap中的信息,通过isChangedServiceInfo进行判断,当发生变动时,serviceInfoMap.put最新数据,当我们需要使用的时候,通过key进行get操作,ServiceInfoMap默认创建空的对象,但如果配置启动从缓存文件中获取,则会从缓存中获取信息。而且当我们服务实例发生变更的时候,会通过DiskCache.write()向对应的目录文件中写入ServiceInfo信息

本地缓存地址

本地缓存的地址通过cacheDir进行执行本地缓存和故障处理的根目录,在ServiceInfoHolder构造方法中,会默认生成缓存目录,默认路径为${user}/nacos/naming/public,我们也可以需要通过System.setProperty("JM.SNAPSHOT.PATH")指定。

public class ServiceInfoHolder implements Closeable {     private String cacheDir;          public ServiceInfoHolder(String namespace, Properties properties) {     //初始化生成缓存目录     initCacheDir(namespace, properties);     ......     }          private void initCacheDir(String namespace, Properties properties) {         String jmSnapshotPath = System.getProperty(JM_SNAPSHOT_PATH_PROPERTY);          String namingCacheRegistryDir = "";         if (properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR) != null) {             namingCacheRegistryDir = File.separator + properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR);         }          if (!StringUtils.isBlank(jmSnapshotPath)) {             cacheDir = jmSnapshotPath + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir                     + File.separator + FILE_PATH_NAMING + File.separator + namespace;         } else {             cacheDir = System.getProperty(USER_HOME_PROPERTY) + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir                     + File.separator + FILE_PATH_NAMING + File.separator + namespace;         }     }      } 

故障处理

ServiceInfoHolder构造方法中,还会初始化一个FailoverReactor的类,这个类主要是用来故障处理。

public class ServiceInfoHolder implements Closeable {   private final FailoverReactor failoverReactor;      public ServiceInfoHolder(String namespace, Properties properties) {     ....     //为两者相互持有对方的引用     this.failoverReactor = new FailoverReactor(this, cacheDir);     .....   }       public FailoverReactor(ServiceInfoHolder serviceInfoHolder, String cacheDir) {         //获取serviceInfoHolder引用对象         this.serviceInfoHolder = serviceInfoHolder;         //故障目录${user}/nacos/naming/public/failover         this.failoverDir = cacheDir + FAILOVER_DIR;         //初始化executorService         this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {             @Override             public Thread newThread(Runnable r) {                 Thread thread = new Thread(r);                 //开启守护线程                 thread.setDaemon(true);                 thread.setName("com.alibaba.nacos.naming.failover");                 return thread;             }         });         //其他信息初始化         this.init();     }           public void init() {         //执行初始化操作,间隔5秒,执行SwitchRefresher()任务         executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);          //初始化操作,延迟30分钟执行,间隔24小时,执行DiskFileWriter()任务         executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);                   //初始化操作,间隔10秒,核心方法为DiskFileWriter         executorService.schedule(new Runnable() {             @Override             public void run() {                 try {                     File cacheDir = new File(failoverDir);                                          if (!cacheDir.exists() && !cacheDir.mkdirs()) {                         throw new IllegalStateException("failed to create cache dir: " + failoverDir);                     }                                          File[] files = cacheDir.listFiles();                     //如果故障目录为空,启动立即执行,备份文件                     if (files == null || files.length <= 0) {                         new DiskFileWriter().run();                     }                 } catch (Throwable e) {                     NAMING_LOGGER.error("[NA] failed to backup file on startup.", e);                 }                              }         }, 10000L, TimeUnit.MILLISECONDS);     } } 

init()代码中,开启了三个定时任务,三个任务都是FailoverReactor内部类,

  • 执行初始化操作,间隔5秒,执行SwitchRefresher()任务
  • 初始化操作,延迟30分钟执行,间隔24小时,执行DiskFileWriter()任务
  • 初始化操作,间隔10秒,核心方法为DiskFileWriter

我们这里先来看一下核心方法DiskFileWriter,这里主要是获取服务信息,判断是否能够写入磁盘,条件满足,写入拼接的故障目录中,因为第一个和第二个初始化操作,都会用到DiskFileWriter,当我们第三个定时判断如果文件不存在,则会将文件写入本地磁盘中

class DiskFileWriter extends TimerTask {    @Override   public void run() {       Map<String, ServiceInfo> map = serviceInfoHolder.getServiceInfoMap();       for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) {           ServiceInfo serviceInfo = entry.getValue();           //主要是判断服务信息是否完整           if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils                   .equals(serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY) || StringUtils                   .equals(serviceInfo.getName(), UtilAndComs.ENV_CONFIGS) || StringUtils                   .equals(serviceInfo.getName(), UtilAndComs.VIP_CLIENT_FILE) || StringUtils                   .equals(serviceInfo.getName(), UtilAndComs.ALL_HOSTS)) {               continue;           }           //将文件写入磁盘中           DiskCache.write(serviceInfo, failoverDir);       }   } } 

接下来我们再看一下,第一个定时任务SwitchRefresher的业务逻辑,

class SwitchRefresher implements Runnable {  long lastModifiedMillis = 0L;  @Override public void run() {   try {       File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);       //如果文件不存在返回       if (!switchFile.exists()) {           switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());           NAMING_LOGGER.debug("failover switch is not found, {}", switchFile.getName());           return;       }        long modified = switchFile.lastModified();       //判断文件修改时间       if (lastModifiedMillis < modified) {           lastModifiedMillis = modified;           //获取故障处理文件内容           String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH,                   Charset.defaultCharset().toString());           if (!StringUtils.isEmpty(failover)) {               String[] lines = failover.split(DiskCache.getLineSeparator());                for (String line : lines) {                   String line1 = line.trim();                   //"1" 开启故障处理                   if (IS_FAILOVER_MODE.equals(line1)) {                       switchParams.put(FAILOVER_MODE_PARAM, Boolean.TRUE.toString());                       NAMING_LOGGER.info("failover-mode is on");                       new FailoverFileReader().run();                       //"0" 关闭故障处理                   } else if (NO_FAILOVER_MODE.equals(line1)) {                       switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());                       NAMING_LOGGER.info("failover-mode is off");                   }               }           } else {               switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());           }       }    } catch (Throwable e) {       NAMING_LOGGER.error("[NA] failed to read failover switch.", e);   } } } 

这里面主要是判断故障处理文件是否存在,不存在直接返回,再去比较文件的修改时间,如果已经修改,则获取文件中的内容,继续进行判断,当我们开启故障处理时,执行线程FailoverFileReader().run()

class FailoverFileReader implements Runnable {  @Override public void run() {   Map<String, ServiceInfo> domMap = new HashMap<String, ServiceInfo>(16);    BufferedReader reader = null;   try {       //读取failoverDir目录下的文件       File cacheDir = new File(failoverDir);       //不存在返回错误       if (!cacheDir.exists() && !cacheDir.mkdirs()) {           throw new IllegalStateException("failed to create cache dir: " + failoverDir);       }       //获取文件       File[] files = cacheDir.listFiles();       //文件不存在返回       if (files == null) {           return;       }       //遍历处理       for (File file : files) {           //文件不存在跳过           if (!file.isFile()) {               continue;           }           //如果是故障处理标志文件,跳过这一步           if (file.getName().equals(UtilAndComs.FAILOVER_SWITCH)) {               continue;           }            ServiceInfo dom = new ServiceInfo(file.getName());            //读取备份中的内容,转换为ServiceInfo对象           try {               String dataString = ConcurrentDiskUtil                       .getFileContent(file, Charset.defaultCharset().toString());               reader = new BufferedReader(new StringReader(dataString));                String json;               if ((json = reader.readLine()) != null) {                   try {                       dom = JacksonUtils.toObj(json, ServiceInfo.class);                   } catch (Exception e) {                       NAMING_LOGGER.error("[NA] error while parsing cached dom : {}", json, e);                   }               }            } catch (Exception e) {               NAMING_LOGGER.error("[NA] failed to read cache for dom: {}", file.getName(), e);           } finally {               try {                   if (reader != null) {                       reader.close();                   }               } catch (Exception e) {                   //ignore               }           }           if (!CollectionUtils.isEmpty(dom.getHosts())) {               //将ServiceInfo对象放入domMap中               domMap.put(dom.getKey(), dom);           }       }   } catch (Exception e) {       NAMING_LOGGER.error("[NA] failed to read cache file", e);   }   //如果不为空,赋值serviceMap   if (domMap.size() > 0) {       serviceMap = domMap;   } } } 

FailoverFileReader主要是操作读取failover目录存储的备份服务信息文件内容,然后装换成ServiceInfo信息,并将所有的ServiceInfo储存在FailoverReactorServiceMap属性中。

总结

到这里我们Nacos订阅机制核心流程就讲完了,整体订阅机制的流程还是比较复杂的,因为还涉及到之前将的逻辑,会有点绕,并且用到了保证线程Map、守护线程、阻塞队列、线程的使用等等,我们需要重点掌握的主要是事件发布者、订阅者之间的关系,这里还是推荐大家有机会的话可以自己跟着源码走一遍,会有更深的体验。

如果觉得文中有帮助的,记得点赞支持,你的支持是我创作的最大动力!

我是牧小农,怕什么真理无穷,进一步有进一步的欢喜,大家加油!

发表评论

相关文章

当前内容话题