Trino源码学习-Airlift源码分析

public class HttpDiscoveryAnnouncementClient
        implements DiscoveryAnnouncementClient
{
    @Inject
    public HttpDiscoveryAnnouncementClient(
            @ForDiscoveryClient Supplier<URI> discoveryServiceURI,
            NodeInfo nodeInfo,
            JsonCodec<Announcement> announcementCodec,
            @ForDiscoveryClient HttpClient httpClient)
    {
        requireNonNull(discoveryServiceURI, "discoveryServiceURI is null");
        requireNonNull(nodeInfo, "nodeInfo is null");
        requireNonNull(announcementCodec, "announcementCodec is null");
        requireNonNull(httpClient, "httpClient is null");

        this.nodeInfo = nodeInfo;
        this.discoveryServiceURI = discoveryServiceURI;
        this.announcementCodec = announcementCodec;
        this.httpClient = httpClient;
    }
    // 注册节点
    @Override
    public ListenableFuture<Duration> announce(Set<ServiceAnnouncement> services)
    {
        requireNonNull(services, "services is null");

        URI uri = discoveryServiceURI.get();
        if (uri == null) {
            return immediateFailedFuture(new DiscoveryException("No discovery servers are available"));
        }
        // 注册信息
        Announcement announcement = new Announcement(nodeInfo.getEnvironment(), nodeInfo.getNodeId(), nodeInfo.getPool(), nodeInfo.getLocation(), services);
        Request request = preparePut()
                .setUri(createAnnouncementLocation(uri, nodeInfo.getNodeId()))
                .setHeader("User-Agent", nodeInfo.getNodeId())
                .setHeader("Content-Type", MEDIA_TYPE_JSON.toString())
                .setBodyGenerator(jsonBodyGenerator(announcementCodec, announcement))
                .build();
        return httpClient.executeAsync(request, new DiscoveryResponseHandler<Duration>("Announcement", uri)
        {
            @Override
            public Duration handle(Request request, Response response)
                    throws DiscoveryException
            {
                int statusCode = response.getStatusCode();
                if (!isSuccess(statusCode)) {
                    throw new DiscoveryException(String.format("Announcement failed with status code %s: %s", statusCode, getBodyForError(response)));
                }

                Duration maxAge = extractMaxAge(response); //抽取超时时间，默认是10s
                return maxAge;
            }
        });
    }
    // 注册服务URL
    @VisibleForTesting
    static URI createAnnouncementLocation(URI baseUri, String nodeId)
    {
        return uriBuilderFrom(baseUri)
                .appendPath("/v1/announcement")
                .appendPath(nodeId)
                .build();
    }
    // 取消注册
    @Override
    public ListenableFuture<Void> unannounce()
    {
        URI uri = discoveryServiceURI.get();
        if (uri == null) {
            return immediateFuture(null);
        }

        Request request = prepareDelete()
                .setUri(createAnnouncementLocation(uri, nodeInfo.getNodeId()))
                .setHeader("User-Agent", nodeInfo.getNodeId())
                .build();
        return httpClient.executeAsync(request, new DiscoveryResponseHandler<>("Unannouncement", uri));
    }
    ... ...
}

// io.airlift.discovery.client.Announcer
// announce 准备开始
public void start()
{
    checkState(!executor.isShutdown(), "Announcer has been destroyed");
    if (started.compareAndSet(false, true)) {
        // announce immediately, if discovery is running
        ListenableFuture<Duration> announce = announce(System.nanoTime(), new Duration(0, SECONDS));
        try {
            announce.get(30, SECONDS); // 30秒的最大等待时间
        }
        catch (Exception ignored) {
        }
    }
}
// announce 异步发送，在回调函数中进行延迟等待，然后准备下一次的announce
private ListenableFuture<Duration> announce(long delayStart, Duration expectedDelay)
{
    // log announcement did not happen within 5 seconds of expected delay
    if (System.nanoTime() - (delayStart + expectedDelay.roundTo(NANOSECONDS)) > SECONDS.toNanos(5)) {
        log.error("Expected service announcement after %s, but announcement was delayed %s", expectedDelay, Duration.nanosSince(delayStart));
    }

    long requestStart = System.nanoTime();
    // 在announcementClient中发起请求
    ListenableFuture<Duration> future = announcementClient.announce(getServiceAnnouncements());

    // 通过future的回调函数异步获得结果
    Futures.addCallback(future, new FutureCallback<Duration>()
    {
        @Override
        public void onSuccess(Duration expectedDelay)
        {
            errorBackOff.success();

            // wait 80% of the suggested delay，防止超出maxAge
            expectedDelay = new Duration(expectedDelay.toMillis() * 0.8, MILLISECONDS);
            scheduleNextAnnouncement(expectedDelay); // 下一次的announce
        }

        @Override
        public void onFailure(Throwable t)
        {
            Duration duration = errorBackOff.failed(t);
            scheduleNextAnnouncement(duration); // 下一次的announce
        }
    }, executor);
    return future;
}
// 下一次的announce
private void scheduleNextAnnouncement(Duration expectedDelay)
{
    // already stopped?  avoids rejection exception
    if (executor.isShutdown()) {
        return;
    }
    long delayStart = System.nanoTime();
    executor.schedule(() -> announce(delayStart, expectedDelay), expectedDelay.toMillis(), MILLISECONDS);
}

// io.airlift.discovery.server.DynamicAnnouncementResource 服务端接收注册的代码
@PUT
@Consumes(MediaType.APPLICATION_JSON)
public Response put(@PathParam("node_id") Id<Node> nodeId, @Context UriInfo uriInfo, DynamicAnnouncement announcement)
{
  if (!nodeInfo.getEnvironment().equals(announcement.getEnvironment())) {
    return Response.status(BAD_REQUEST)
            .entity(format("Environment mismatch. Expected: %s, Provided: %s", nodeInfo.getEnvironment(), announcement.getEnvironment()))
            .build();
    }

  String location = firstNonNull(announcement.getLocation(), "/somewhere/" + nodeId.toString());

  DynamicAnnouncement announcementWithLocation = DynamicAnnouncement.copyOf(announcement)
    .setLocation(location)
    .build();

  dynamicStore.put(nodeId, announcementWithLocation);

  return Response.status(ACCEPTED).build();
}

{
   "environment":"test",
   "services":[
      {
         "id":"C8A9EE64-0476-452C-8638-8E72F3EE3CA6",
         "nodeId":"597A741E-9968-40E2-BB4D-7AF26DE18689",
         "type":"discovery",
         "pool":"general",
         "location":"/172.17.31.245",
         "state":"RUNNING",
         "properties":{
            "http":"http://172.17.31.245:8411"
         }
      },
      {
         "id":"370AF416-5F44-47D3-BFB6-D93A92676D49",
         "nodeId":"0BA42FDB-5DBA-4A2C-BE26-9596B7B4368E",
         "type":"discovery",
         "pool":"general",
         "location":"/172.17.31.246",
         "state":"RUNNING",
         "properties":{
            "http":"http://172.17.31.246:8411"
         }
      }
   ]
}

//io.airlift.discovery.store.DistributedStore
@PostConstruct
public void start(){
  garbageCollector.scheduleAtFixedRate(new Runnable(){
    @Override
    public void run()
    {
      removeExpiredEntries();
    }
  }, 0, garbageCollectionInterval.toMillis(), TimeUnit.MILLISECONDS);
}

@Managed
public void removeExpiredEntries(){
  for (Entry entry : localStore.getAll()) {
    if (isExpired(entry)) {
        localStore.delete(entry.getKey(), entry.getVersion());
    }
  }

  lastGcTimestamp.set(System.currentTimeMillis());
}

// io.airlift.discovery.client.CachingServiceSelector
@Override
public ListenableFuture<List<ServiceDescriptor>> refresh()
{
    ServiceDescriptors oldDescriptors = this.serviceDescriptors.get();
    ListenableFuture<ServiceDescriptors> future;
    if (oldDescriptors == null) {
        future = lookupClient.getServices(type, pool);
    }
    else {
        future = lookupClient.refreshServices(oldDescriptors);
    }
    future = chainedCallback(future, new FutureCallback<ServiceDescriptors>()
    {
        @Override
        public void onSuccess(ServiceDescriptors newDescriptors)
        {
            serviceDescriptors.set(newDescriptors);
            errorBackOff.success();
            Duration delay = newDescriptors.getMaxAge();
            if (delay == null) {
                delay = DEFAULT_DELAY; // 这个变量的默认是10秒
            }
            scheduleRefresh(delay); // 刷新
        }
        @Override
        public void onFailure(Throwable t) { ... }
    }, executor);
    ...
}