Eureka Server源码解析（全量拉取注册表流程）

刺骨的言语ヽ痛彻心扉 2022-09-09 02:26 167阅读 0赞

> 原创不易，转载请注明出处

# 系列文章目录 #

1.  《[SpringCloud Eureka Server源码解析（启动流程）][SpringCloud Eureka Server]》
2.  《[Eureka Server源码解析（服务注册流程）][Eureka Server]》
3.  《[Eureka Server源码解析（服务续约流程）][Eureka Server 1]》
4.  《[Eureka Server源码解析（服务主动下线流程）][Eureka Server 2]》
5.  《[Eureka Server源码解析（服务故障下线剔除流程）][Eureka Server 3]》
6.  《[深度解析Eureka的自我保护机制][Eureka]》

### 文章目录 ###

*  系列文章目录
 *  前言
 *  1.源码解析
 *   *  1.1 多级缓存初始化
     *  1.2 处理全量拉取注册表请求
 *  2.流程图
 *  总结

--------------------

# 前言 #

在创建服务发现客户端对象的时候，默认是会到Eureka Server 拉取一下全量的注册表信息（不管是Spring Cloud 那套还是你单独使用netflix的Eureka ），接下来解析下Eureka Server是怎样处理拉取全量注册表请求的源码。  
本文涉及到Eureka Server的多级缓存设计，还是很重要的。

# 1.源码解析 #

## 1.1 多级缓存初始化 ##

在Eureka Server 初始化的时候，会调用注册表的init方法进行初始化，这个时候就会创建多级缓存并初始化。  
PeerAwareInstanceRegistryImpl\# init  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAJOeggeWHuuacquadpQ_size_20_color_FFFFFF_t_70_g_se_x_16]  
我们看看initializedResponseCache 方法干了什么  
![在这里插入图片描述][34d1128aa70141dbbfc09df38c42702c.png]  
其实就是创建了一个ResponseCacheImpl对象。它里面个成员变量需要我们注意下

private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();

也就是只读缓存，它是第一级缓存，是个ConcurrentMap  
在ResponseCacheImpl 构造中，有2个点我们需要看下，1是创建ReadWrite缓存，2是启动一个定时任务用来更新缓存。  
先来看看ReadWrite缓存的初始化

this.readWriteCacheMap =
                    CacheBuilder.newBuilder().initialCapacity(serverConfig.getInitialCapacityOfResponseCache())
                            .expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
                            .removalListener(new RemovalListener<Key, Value>() { 
                                @Override
                                public void onRemoval(RemovalNotification<Key, Value> notification) { 
                                    Key removedKey = notification.getKey();
                                    if (removedKey.hasRegions()) { 
                                        Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
                                        regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
                                    }
                                }
                            })
                            .build(new CacheLoader<Key, Value>() { 
                                @Override
                                public Value load(Key key) throws Exception { 
                                    if (key.hasRegions()) { 
                                        Key cloneWithNoRegions = key.cloneWithoutRegions();
                                        regionSpecificKeys.put(cloneWithNoRegions, key);
                                    }
                                    Value value = generatePayload(key);
                                    return value;
                                }
                            });

这个ReadWrite 缓存是使用的google guava包里面的CacheLoader缓存。  
里面的细节我们先不用关心，这里暂时只需要知道它是在ResponseCacheImpl 构造中创建的即可。  
再来看看第2个重要的点，刷新缓存定时任务。

if (shouldUseReadOnlyResponseCache) { 
                timer.schedule(getCacheUpdateTask(),
                        new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                                + responseCacheUpdateIntervalMs),
                        responseCacheUpdateIntervalMs);
            }

如果启用readOnly缓存的话，就会创建一个定时任务，时间间隔是30s，这个任务主要是更新readOnly一级缓存的。最后我们看下是怎样更新readOnly一级缓存的。

private TimerTask getCacheUpdateTask() { 
        return new TimerTask() { 
            @Override
            public void run() { 
                for (Key key : readOnlyCacheMap.keySet()) { 
                    CurrentRequestVersion.set(key.getVersion());
                      //从readWrite 缓存中获取该key对应的value
                      Value cacheValue = readWriteCacheMap.get(key);
                      // 从readOnly中获取该key对应缓存的值
                      Value currentCacheValue = readOnlyCacheMap.get(key);
                      // 如果两个value不一致的话，说明出现了缓存不一致的情况，这个时候就会更新readOnly里面的缓存
                      if (cacheValue != currentCacheValue) { 
                          readOnlyCacheMap.put(key, cacheValue);
                      }
                }
            }
        };
    }

其实很简单，就是遍历readOnly 一级缓存中所有的key ，拿着该key去二级缓存ReadWirte中获取对应的值，同时也从readOnly自己里面获取该key对应的value值，比较2个value值是否相等，如果相等的话，说明没有变，如果不相等，说明缓存不一致，就会将readWrite 里面的值更新到readOnly一级缓存中。

## 1.2 处理全量拉取注册表请求 ##

客户端服务发送get请求/apps去Eureka Server 拉取全量的注册表。  
是由ApplicationsResource 的getContainers 方法处理的。

Key cacheKey = new Key(Key.EntityType.Application,
             ResponseCacheImpl.ALL_APPS,
             keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
     );
    
     Response response = Response.ok(responseCache.getGZIP(cacheKey))
                 .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                 .header(HEADER_CONTENT_TYPE, returnMediaType)
                 .build();

先是构建一个key对象，注意ALL\_APPS 这，说明是获取所有的注册实例信息。接着调用responseCache 的getGZIP 方法获取注册表信息。这个responseCache就是1.1小节我们介绍的ResponseCacheImpl对象。

public byte[] getGZIP(Key key) { 
        Value payload = getValue(key, shouldUseReadOnlyResponseCache);
        if (payload == null) { 
            return null;
        }
        return payload.getGzipped();
    }

调用getValue 方法获取一个value。

Value getValue(final Key key, boolean useReadOnlyCache) { 
        Value payload = null;
        try { 
            if (useReadOnlyCache) { 
                final Value currentPayload = readOnlyCacheMap.get(key);
                if (currentPayload != null) { 
                    payload = currentPayload;
                } else { 
                    payload = readWriteCacheMap.get(key);
                    readOnlyCacheMap.put(key, payload);
                }
            } else { 
                payload = readWriteCacheMap.get(key);
            }
        } catch (Throwable t) { 
            logger.error("Cannot get value for key : {}", key, t);
        }
        return payload;
    }

如果使用readOnly缓存的话，获取先根据key到readOnly缓存中获取，这里我们是启用的，如果是从readOnly一级缓存中能够获取到的话，直接返回了，如果获取不到的话，就从readWrite 二级缓存中获取，并且将值缓存到readOnly一级缓存中。  
那如果readWrite 这个二级缓存中也没有怎么办？ guava这个缓存没有的话会去CacheLoader 的load方法加载，在创建缓存的时候设置的这个CacheLoader，这个CacheLoader 会将获取到value 再缓存到readWrite缓存中，这个是他的一个机制，我们看看

new CacheLoader<Key, Value>() { 
        @Override
        public Value load(Key key) throws Exception { 
            if (key.hasRegions()) { 
                Key cloneWithNoRegions = key.cloneWithoutRegions();
                regionSpecificKeys.put(cloneWithNoRegions, key);
            }
            Value value = generatePayload(key);
            return value;
        }
    }

调用generatePayload方法，这个方法体很长，我们只贴出来与之相关的代码部分即可。

if (ALL_APPS.equals(key.getName())) { 
    	tracer = serializeAllAppsTimer.start();
    	payload = getPayLoad(key, registry.getApplications());
    }

请求过来的时候，封装的缓存key就是ALL\_APPS 。  
有个非常重要的地方，就是registry.getApplications() ，获取注册表中所有的实例信息，这里怎样获取的我们不用看也能猜到，就是遍历注册表中所有的租约信息，取到对应的实例信息放到一个集合中。

Applications apps = new Applications();
    apps.setVersion(1L);
    for (Entry<String, Map<String, Lease<InstanceInfo>>> entry : registry.entrySet()) { 
       Application app = null;
       if (entry.getValue() != null) { 
           for (Entry<String, Lease<InstanceInfo>> stringLeaseEntry : entry.getValue().entrySet()) { 
               Lease<InstanceInfo> lease = stringLeaseEntry.getValue();
               if (app == null) { 
                   app = new Application(lease.getHolder().getAppName());
               }
               app.addInstance(decorateInstanceInfo(lease));
           }
       }
       if (app != null) { 
           apps.addApplication(app);
       }
    }

这里我们还是贴出来，就是遍历注册表，将实例信息塞到集合中。  
这个getPayLoad 方法其实没啥好看的，就是编码。

最后一个问题，注册表中的实例信息改变了，这个readWrite 二级缓存是怎样保证缓存一致的。  
我们在服务注册、 服务续约、服务下线的时候，都有说过一个事情，就是更新了完了注册表，就会去删除对应的缓存。其实是调用的ResponseCache 的invalidateCache 方法

private void invalidateCache(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress) { 
        // invalidate cache
        responseCache.invalidate(appName, vipAddress, secureVipAddress);
    }

@Override
    public void invalidate(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress) { 
      for (Key.KeyType type : Key.KeyType.values()) { 
          for (Version v : Version.values()) { 
          		//调用invalidate 方法清除
              invalidate(
                      new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.compact),
                      new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.compact),
                      new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.compact)
              );
              if (null != vipAddress) { 
                  invalidate(new Key(Key.EntityType.VIP, vipAddress, type, v, EurekaAccept.full));
              }
              if (null != secureVipAddress) { 
                  invalidate(new Key(Key.EntityType.SVIP, secureVipAddress, type, v, EurekaAccept.full));
              }
          }
      }
    }

这个方法主要是封装该实例多有类型的key，然后调用invalidate 方法进行清理。

public void invalidate(Key... keys) { 
        for (Key key : keys) { 
            logger.debug("Invalidating the response cache key : {} {} {} {}, {}",
                    key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
            // 清理对应key的所有缓存
            readWriteCacheMap.invalidate(key);
            Collection<Key> keysWithRegions = regionSpecificKeys.get(key);
            if (null != keysWithRegions && !keysWithRegions.isEmpty()) { 
                for (Key keysWithRegion : keysWithRegions) { 
                    logger.debug("Invalidating the response cache key : {} {} {} {} {}",
                            key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
                    readWriteCacheMap.invalidate(keysWithRegion);
                }
            }
        }
    }

遍历所有类型的key，去readWrite这个二级缓存中清理。

# 2.流程图 #

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAJOeggeWHuuacquadpQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]

# 总结 #

本文主要解析了客户端去Eureka Server 拉取全量注册表处理源码，可以看到，Eureka Server 是用了多级缓存来提高性能，一级缓存是readOnly 一个map ，二级缓存就是readWrite guava提供的一个本地缓存工具，三级缓存可以理解为注册表。  
流程是，先去readOnly里面获取，如果readOnly缓存中没有就去readWrite 缓存中获取，获取到后将值缓存到readOnly一级缓存中，readWrite也没有的话，就会去注册表中获取，并缓存到readWrite缓存中。  
readOnly 是怎样保证数据一致性的？缓存组件创建的时候，会创建一个定时任务，默认30s 比对readOnly 与readWrite的值是否相等，不相等更新readOnly里面的值。  
readWrite是怎样保证数据一致性的？  
在服务注册、服务续约 、服务下线的时候，都会删除对应的readWrite 中的缓存。

[SpringCloud Eureka Server]: https://blog.csdn.net/yuanshangshenghuo/article/details/119920341
[Eureka Server]: https://blog.csdn.net/yuanshangshenghuo/article/details/119931667
[Eureka Server 1]: https://blog.csdn.net/yuanshangshenghuo/article/details/119951880
[Eureka Server 2]: https://blog.csdn.net/yuanshangshenghuo/article/details/119960880
[Eureka Server 3]: https://blog.csdn.net/yuanshangshenghuo/article/details/119961592
[Eureka]: https://inetyoung.blog.csdn.net/article/details/110131223
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAJOeggeWHuuacquadpQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/2a1bc58b119348c8b51c26319852074c.png
[34d1128aa70141dbbfc09df38c42702c.png]: /images/20220829/721ecd3cd8f04dd2a63ee28c0e21ae23.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAJOeggeWHuuacquadpQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]: /images/20220829/08baa34bad9f4a758b24b4bbfd23d83a.png