三、eureka服务端获取服务列表

所有文章

https://www.cnblogs.com/lay2017/p/11908715.html

正文

eureka服务端维护了一个服务信息的列表，服务端节点之间相互复制服务信息。而作为eureka的客户端将会从eureka服务端请求这个服务信息列表，选择对应的实例。本文就来看看eureka服务端对客户端提供的获取服务信息列表的http接口。

eureka服务端基于jersey来提供http服务调用，所以我们先找到它的Resource。

ApplicationsResource

@GET
public Response getContainers(@PathParam("version") String version,
                              @HeaderParam(HEADER_ACCEPT) String acceptHeader,
                              @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
                              @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
                              @Context UriInfo uriInfo,
                              @Nullable @QueryParam("regions") String regionsStr) {

    // ...

    Key cacheKey = new Key(Key.EntityType.Application,
            ResponseCacheImpl.ALL_APPS,
            keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
    );

    Response response;
    if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) {
        response = Response.ok(responseCache.getGZIP(cacheKey))
                .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                .header(HEADER_CONTENT_TYPE, returnMediaType)
                .build();
    } else {
        response = Response.ok(responseCache.get(cacheKey))
                .build();
    }
    return response;
}

这里从responseCache当中获取了Applications的序列号结果直接返回了，所以我们先看看ResponseCache是从哪里来的

可以看到ResponseCache是再ApplicationsResource构造的时候从Registry中获取的

@Inject
ApplicationsResource(EurekaServerContext eurekaServer) {
    this.serverConfig = eurekaServer.getServerConfig();
    this.registry = eurekaServer.getRegistry();
    this.responseCache = registry.getResponseCache();
}

我们再看看ResponseCache的get方法做了什么

String get(final Key key, boolean useReadOnlyCache) {
    Value payload = getValue(key, useReadOnlyCache);
    if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
        return null;
    } else {
        return payload.getPayload();
    }
}

继续跟进getValue

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;
}

可以看到，其实只是从ReadWriteCacheMap当中获取对应的值，那么我们再看看ReadWriteCacheMap是怎么被构造的

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;
                                }
                            });

从这里可以看出，调用ReadWriteCacheMap的get方法，将会触发这里的generatePayload方法

我们跟进generatePayload

private Value generatePayload(Key key) {
    Stopwatch tracer = null;
    try {
        String payload;
        switch (key.getEntityType()) {
            case Application:
                boolean isRemoteRegionRequested = key.hasRegions();
                // 获取所有Application
                if (ALL_APPS.equals(key.getName())) {
                    if (isRemoteRegionRequested) {
                        tracer = serializeAllAppsWithRemoteRegionTimer.start();
                        payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
                    } else {
                        tracer = serializeAllAppsTimer.start();
                        payload = getPayLoad(key, registry.getApplications());
                    }
                } else if (ALL_APPS_DELTA.equals(key.getName())) {
                    // ...
                } else {
                    tracer = serializeOneApptimer.start();
                    // 获取某个Application
                    payload = getPayLoad(key, registry.getApplication(key.getName()));
                }
                break;
            // ...
        }
        return new Value(payload);
    } finally {
        
    }
}

我们看到这里payload主要构成元素是Application，也就是我们需要的服务列表信息。

最后，我们跟进getPayLoad方法，看看这些服务列表信息是怎么被序列号成payload的

private String getPayLoad(Key key, Applications apps) {
    EncoderWrapper encoderWrapper = serverCodecs.getEncoder(key.getType(), key.getEurekaAccept());
    String result;
    try {
        result = encoderWrapper.encode(apps);
    } catch (Exception e) {
        return "";
    }
    return result;
}

编码器的实现比较多种，这里就不展开了

总结

获取服务信息列表其实就是从registry当中获取Applications，然后做一次序列化，最后通过http响应回去。总体来说还是比较简单的。